208 lines
8.7 KiB
Python
208 lines
8.7 KiB
Python
# -*- 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()
|