# -*- coding: utf-8 -*- """独立备份脚本: 把本地 数据.jsonl + 超长待重跑列表 上传到 COS。 不 import 项目里任何其他 .py(不依赖 tool/、main.py 等), 只用标准库 + 第三方 qcloud_cos SDK, 直接读 config/config.json —— 哪怕代码刚更新完、其他模块有问题, 这个脚本也能独立跑, 适合放在 重新部署/更新代码前后单独执行一次, 确保数据先落到 COS。 用法(项目根目录下运行): python tools/upload_data_standalone.py 行为跟 tools/upload_daily.py + tool/job_cos.py 里的 upload_daily_backup/upload_longlist_backup 一致(上传目标、命名、覆盖安全性都对齐): 数据.jsonl -> dispatch.daily_backup_path/<日期>/数据_<机器号>.jsonl (全量快照,覆盖安全) 已处理清单(轻量) -> dispatch.daily_backup_path/<日期>/已处理_<机器号>.txt (从jsonl提取的纯 video_cos_path 列表, 一行一个; 全局去重 global_processed_set() 只拉这个小文件, 不用为了去重下载/解析整份带状态耗时字段的jsonl) 超长待重跑列表 -> dispatch.long_list_backup_path/todo/<原文件名>_<机器号>_<日期>_<时分秒>.txt (增量式:传完清空本地已传部分; key精确到秒,同一天多次调用不会互相覆盖) """ import json import os import socket import subprocess from datetime import datetime from qcloud_cos import CosConfig, CosS3Client ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) CONFIG_PATH = os.path.join(ROOT, "config", "config.json") JSONL_PATH = os.path.join(ROOT, "数据.jsonl") def load_config() -> dict: with open(CONFIG_PATH, "r", encoding="utf-8") as f: return json.load(f) def parse_cos_url(url: str, default_bucket: str): """cos://bucket/a/b -> (bucket, 'a/b')。无 cos:// 前缀时按传入的默认桶补全。""" url = url.strip() if url.startswith("cos://"): rest = url[len("cos://"):] bucket, _, key = rest.partition("/") return bucket, key.strip("/") return default_bucket, url.strip("/") def detect_ip() -> str: """取本机对外默认路由的网卡IP(UDP connect触发内核选路由,不实际发包,天然跳过docker网桥/回环); 取不到退化到 `hostname -I` 第一个。跟 tool/job_cos.py 的 _detect_ip() 逻辑一致。""" try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: s.connect(("8.8.8.8", 80)) ip = s.getsockname()[0] finally: s.close() if ip and not ip.startswith("127."): return ip except OSError: pass try: r = subprocess.run(["hostname", "-I"], capture_output=True, text=True, timeout=5) ips = (r.stdout or "").split() return ips[0] if ips else "" except (OSError, subprocess.SubprocessError): return "" def get_machine_id(dcfg: dict) -> str: """优先级: config.dispatch.machine_id > 环境变量 MACHINE_ID > 本机固定IP > 主机名。 跟 tool/job_cos.py 的 get_machine_id() 保持一致, 这样跟常驻 worker 认领的是同一个身份。""" mid = (dcfg.get("machine_id") or "").strip() if mid: return mid mid = (os.environ.get("MACHINE_ID") or "").strip() if mid: return mid ip = detect_ip() if ip: return ip return socket.gethostname() # 数据.jsonl 是全量快照, 产能高的机器上百 MB/上 GB。put_object 单请求默认超时 30 秒, 大到 # 传不完就一直 "The write operation timed out"。超过这个大小改用 coscli 分块上传(几 GB 也行), # coscli 失败自动回退 put_object。跟 tool/job_cos.py 的 _upload_dated_file 是同一套策略。 _COSCLI_UPLOAD_THRESHOLD = 50 * 1024 * 1024 # 50MB def _coscli_put(sf_cfg: dict, local_path: str, cos_url: str) -> bool: """用 coscli 上传大文件。成功 True 失败 False。sf_cfg = config 的 splitframe 段。""" try: coscli = sf_cfg.get("coscli_path", "./coscli-linux-amd64") coscfg = sf_cfg.get("cosconfig_path", ".cos.yaml") r = subprocess.run([coscli, "-c", coscfg, "cp", local_path, cos_url], capture_output=True, text=True, timeout=1800) if r.returncode == 0: return True print(f" coscli 上传失败(回退 put_object): {(r.stderr or '').strip()[:200]}") return False except Exception as e: print(f" coscli 上传异常(回退 put_object): {type(e).__name__}: {e}") return False def upload_jsonl(client, default_bucket: str, dcfg: dict, sf_cfg: dict, machine_id: str): daily_root = dcfg.get("daily_backup_path") if not daily_root: print("✗ 未配置 dispatch.daily_backup_path, 跳过 数据.jsonl") return if not os.path.exists(JSONL_PATH): print(f"✗ 本地无 {JSONL_PATH}, 跳过") return bkt, base = parse_cos_url(daily_root, default_bucket) date = datetime.now().strftime("%Y%m%d") key = f"{base}/{date}/数据_{machine_id}.jsonl" cos_url = f"cos://{bkt}/{key}" with open(JSONL_PATH, "rb") as f: data = f.read() size = len(data) uploaded = False if size > _COSCLI_UPLOAD_THRESHOLD: print(f" 数据.jsonl 有 {size / 1024 / 1024:.0f}MB, 用 coscli 分块上传") uploaded = _coscli_put(sf_cfg, JSONL_PATH, cos_url) if not uploaded: client.put_object(Bucket=bkt, Body=data, Key=key) print(f"✓ 数据.jsonl 已上传 -> {cos_url}") # 顺带生成一份轻量"已处理清单"(只有 video_cos_path, 一行一个), 供 global_processed_set() # 只拉这个小文件去重用, 不用为了去重下载/解析整份带状态/耗时字段的 jsonl。 paths = [] for line in data.decode("utf-8", errors="ignore").splitlines(): line = line.strip() if not line: continue try: vid = json.loads(line).get("video_cos_path") if vid: paths.append(vid) except json.JSONDecodeError: pass if paths: list_key = f"{base}/{date}/已处理_{machine_id}.txt" client.put_object(Bucket=bkt, Body=("\n".join(paths) + "\n").encode("utf-8"), Key=list_key) print(f"✓ 已处理清单已上传 -> cos://{bkt}/{list_key} ({len(paths)} 条)") def upload_long_list(client, default_bucket: str, dcfg: dict, sub_cfg: dict, machine_id: str): if dcfg.get("run_long_video"): print("本机 run_long_video=true(专职长视频机), 不回传超长列表, 跳过") return long_root = dcfg.get("long_list_backup_path") if not long_root: print("✗ 未配置 dispatch.long_list_backup_path, 跳过超长待重跑列表") return local_long = sub_cfg.get("long_video_list", "./超长待重跑.txt") if not os.path.isabs(local_long): local_long = os.path.join(ROOT, local_long) if not os.path.exists(local_long): print(f"✗ 本地无 {local_long}, 跳过") return with open(local_long, "r", encoding="utf-8") as f: snapshot = f.readlines() if not any(ln.strip() for ln in snapshot): print("✗ 超长待重跑列表为空, 跳过") return bkt, base = parse_cos_url(long_root, default_bucket) stem = os.path.splitext(os.path.basename(local_long))[0] ts = datetime.now().strftime("%Y%m%d_%H%M%S") key = f"{base}/todo/{stem}_{machine_id}_{ts}.txt" client.put_object(Bucket=bkt, Body="".join(snapshot).encode("utf-8"), Key=key) print(f"✓ 超长待重跑列表已上传 -> cos://{bkt}/{key}") if dcfg.get("long_list_clear_after_upload", True): try: with open(local_long, "r", encoding="utf-8") as f: cur = f.readlines() with open(local_long, "w", encoding="utf-8") as f: f.writelines(cur[len(snapshot):]) print(f" 本地已清空已传的 {len(snapshot)} 行, 保留上传期间新增的 {len(cur) - len(snapshot)} 行") except OSError as e: print(f" ⚠ 清空本地文件失败(不影响已上传的部分): {e}") def main(): cfg = load_config() cos_cfg = cfg.get("tencentcloud", {}).get("cos", {}) dcfg = cfg.get("dispatch", {}) or {} sub_cfg = cfg.get("subtitle", {}) or {} sf_cfg = cfg.get("splitframe", {}) or {} # coscli 路径/配置在这段 client = CosS3Client(CosConfig( Region=cos_cfg["region"], SecretId=cos_cfg["secret_id"], SecretKey=cos_cfg["secret_key"], )) default_bucket = cos_cfg["bucket"] machine_id = get_machine_id(dcfg) print(f"机器ID: {machine_id}") upload_jsonl(client, default_bucket, dcfg, sf_cfg, machine_id) upload_long_list(client, default_bucket, dcfg, sub_cfg, machine_id) if __name__ == "__main__": main()