815 lines
38 KiB
Python
815 lines
38 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""COS 任务分发 / 成功记录采集。
|
||
|
||
核心原语: COS PutObject 带 ForbidOverwrite(x-cos-forbid-overwrite:true) —— 同名对象
|
||
只能被创建一次, 并发写同一个锁文件只有第一个成功(其余 409)。以此实现跨机器原子认领,
|
||
无需任何额外服务(Redis/DB)。
|
||
|
||
策略: 粘性认领(sticky)。认领某分片的机器是它的唯一所有者;机器重启后只续做自己名下
|
||
未完成的分片, 别的机器永不抢占 -> 零重复。机器永久报废时, 人工跑 tools/release_machine.py
|
||
把它名下的任务放回队列。
|
||
|
||
COS 布局(jobs_root 下):
|
||
todo/<shard>.txt 待处理分片(每行一个视频 COS 路径)。完成才删, 是"任务存在性"的唯一凭证。
|
||
claimed/<shard>.lock 认领锁, 内容 JSON {machine_id, host, claim_ts, hb_ts}。
|
||
done/<shard>.done 完成标记, 内容 JSON {machine_id, stats, end_ts}。
|
||
独立位置(各自 config 字段, 可指向别处):
|
||
daily_backup_path/<日期>/数据_<机器号>.jsonl 每日 数据.jsonl 全量快照备份。
|
||
daily_backup_path/<日期>/已处理_<机器号>.txt 上面jsonl的轻量版(只有video_cos_path,
|
||
一行一个), 供 global_processed_set() 全局去重用。
|
||
long_list_backup_path/todo/超长待重跑_<机器号>_<日期>_<时分秒>.txt 超长视频重跑队列(本身就是另一套 jobs_root)。
|
||
"""
|
||
import json
|
||
import logging
|
||
import os
|
||
import shutil
|
||
import socket
|
||
import subprocess
|
||
from datetime import datetime, timedelta
|
||
|
||
from qcloud_cos import CosConfig, CosS3Client
|
||
from qcloud_cos.cos_exception import CosServiceError
|
||
|
||
from config.config_utils import config
|
||
|
||
# 老版本 cos-python-sdk-v5 的 put_object 不认识 ForbidOverwrite 参数(其 maplist 没有该映射),
|
||
# 直接传会报 "No Parameter Named ForbidOverwrite"。这里把映射补进去, 使 ForbidOverwrite='true'
|
||
# 注入 x-cos-forbid-overwrite:true 请求头 —— 认领锁的原子性(同名对象只能创建一次)全靠它。
|
||
# 不升级 SDK(线上离线机不便升级), 版本无关, 幂等。
|
||
try:
|
||
from qcloud_cos import cos_comm as _cos_comm
|
||
_cos_comm.maplist.setdefault("ForbidOverwrite", "x-cos-forbid-overwrite")
|
||
except Exception:
|
||
pass
|
||
|
||
# ---------------------------------------------------------------- 客户端 ----
|
||
|
||
_CLIENT = None
|
||
|
||
|
||
def client():
|
||
"""模块级单例 CosS3Client(复用项目现有 COS 凭据)。"""
|
||
global _CLIENT
|
||
if _CLIENT is None:
|
||
c = config.get_cos_config()
|
||
_CLIENT = CosS3Client(CosConfig(
|
||
Region=c["region"], SecretId=c["secret_id"], SecretKey=c["secret_key"],
|
||
))
|
||
return _CLIENT
|
||
|
||
|
||
def parse_cos_url(url: str):
|
||
"""cos://bucket/a/b -> ('bucket', 'a/b')。无 cos:// 前缀时按 bucket 配置补全。"""
|
||
url = url.strip()
|
||
if url.startswith("cos://"):
|
||
rest = url[len("cos://"):]
|
||
bucket, _, key = rest.partition("/")
|
||
return bucket, key.strip("/")
|
||
return config.get_cos_config()["bucket"], url.strip("/")
|
||
|
||
|
||
# ----------------------------------------------------------------- 配置 ----
|
||
|
||
def _cfg():
|
||
return config.get_dispatch_config() or {}
|
||
|
||
|
||
def _detect_ip() -> str:
|
||
"""取本机对外默认路由的网卡IP:用 UDP socket connect 一个外部地址触发内核选路由
|
||
(不会真的发包), 拿到的本地地址天然跳过 docker 网桥(如172.17.0.1)/回环这类非默认路由
|
||
接口 —— 通常就是 SSH 连进来那张网卡的IP(前提:机器只有一条默认路由出口, 常见的
|
||
'一张主网卡+docker网桥'部署; 若有多张真实网卡且SSH走的不是默认路由那张, 不适用,
|
||
建议改用 `ip route get 8.8.8.8` 在目标机上核实后再决定)。
|
||
取不到再退化到 `hostname -I` 第一个; 都拿不到返回空串, 上层回退主机名。
|
||
仅在 IP 固定(非 DHCP)的机器上用它兜底身份才安全。"""
|
||
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() -> str:
|
||
"""机器ID 取值优先级: config.dispatch.machine_id > 环境变量 MACHINE_ID >
|
||
本机固定IP(默认路由网卡, 见 _detect_ip) > 主机名。
|
||
必须【唯一且重启后稳定】—— 锁靠它判断'这片是不是我的'。克隆机场景下主机名常常相同,
|
||
固定IP各机器天然不同,可以直接兜底;但如果 IP 是 DHCP 会变的,不要依赖这个,
|
||
换IP等于换身份,旧身份名下未完成的锁会变孤儿——那种环境请在 config 里显式填 machine_id。"""
|
||
mid = (_cfg().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()
|
||
|
||
|
||
def is_ocr_colocated() -> bool:
|
||
"""本机是不是 OCR 服务共置机: 本机默认路由IP是否出现在 config.tencentcloud.ocr.endpoints
|
||
列表里。这批端点IP本来就是各台跑OCR服务的机器, 如果本机自己也在这个列表里, 说明这台机器
|
||
除了跑抽帧/OCR客户端之外, 本身还扛着一个OCR服务, 需要多留CPU给它, 不能被抽帧吃满。
|
||
只覆盖"OCR端点=本机IP"这一种共置方式; 其它不出现在这个列表里的共享方式(比如机器上还跑
|
||
了别的无关服务)探测不到, 那种情况还是得手动传小一点的 --max-workers。"""
|
||
try:
|
||
endpoints = set((config.get_tencent_ocr_config() or {}).get("endpoints") or [])
|
||
except Exception:
|
||
return False
|
||
if not endpoints:
|
||
return False
|
||
ip = _detect_ip()
|
||
return bool(ip) and ip in endpoints
|
||
|
||
|
||
MAX_AUTO_WORKERS = 30 # auto_max_workers() 的硬上限, 见函数docstring③
|
||
|
||
|
||
def auto_max_workers(cpu: int, disk_guard_gb: float = 0.0) -> int:
|
||
"""--max-workers 未指定时的自动选择:
|
||
① 默认按CPU核数; 如果 is_ocr_colocated() 判断本机还跑着OCR服务, 打对折让出一半CPU。
|
||
② 再按本地磁盘算一次上限, 取①②里更小的那个 —— 核多盘小的机器(比如高核数但磁盘只有
|
||
100多G)按纯CPU数选出的并发数,同时下载/抽帧/拼合成图会把磁盘写爆;`--disk-guard-gb`只挡
|
||
"新启动"挡不住"已经在跑的", 真正决定磁盘峰值占用的是并发数本身, 所以从源头把并发数卡住:
|
||
磁盘上限 = (可用磁盘 - disk_guard_gb这层安全垫) ÷ 每视频预估峰值占用(GB)
|
||
每视频预估占用读 config.dispatch.est_disk_gb_per_video, 默认 2.0G —— 按 subtitle 配置常见的
|
||
30~2400秒短视频估的(2400秒=3fps≈7200帧, 裁剪后的帧+合成图+源视频下载, 峰值同时存在时
|
||
量级在1~2G上下; 视频普遍更短/分辨率更低可以调小, 更长/更高清要调大, 保守起见默认按上限估)。
|
||
disk_guard_gb 传0(不用磁盘保护)时不再额外扣这层安全垫, 直接用当前可用磁盘算。
|
||
③ 最后再套一个硬上限 MAX_AUTO_WORKERS(30) —— 超高核数机器(如64/128核)按纯CPU数选出的
|
||
并发数会远超实际有意义的范围(单机瓶颈通常先卡在OCR接口/磁盘IO/内存, 而不是CPU), 且并发
|
||
数太高会让 ffmpeg_threads="auto" 摊出来的每进程线程数过低(见 multithread_algo.py)。
|
||
cpu<=0(测不到核数, 比如非Linux)时按10兜底。"""
|
||
cpu = cpu if cpu and cpu > 0 else 10
|
||
workers = cpu
|
||
if is_ocr_colocated():
|
||
workers = max(1, cpu // 2)
|
||
logging.info(f"[资源探测] 本机IP在 tencentcloud.ocr.endpoints 里, 判断为OCR共置机, "
|
||
f"max-workers 自动打对折让资源给OCR服务: {cpu}核 → {workers}")
|
||
|
||
try:
|
||
free_gb = shutil.disk_usage(".").free / 1024 / 1024 / 1024
|
||
budget_gb = free_gb - max(0.0, disk_guard_gb)
|
||
est_gb = float(_cfg().get("est_disk_gb_per_video", 2.0) or 2.0)
|
||
if budget_gb <= 0:
|
||
logging.warning(f"[资源探测] 可用磁盘({free_gb:.1f}G)已经不够留给磁盘保护垫"
|
||
f"({disk_guard_gb}G), max-workers 收紧到1, 建议先清理磁盘")
|
||
workers = 1
|
||
elif est_gb > 0:
|
||
disk_cap = max(1, int(budget_gb / est_gb))
|
||
if disk_cap < workers:
|
||
logging.info(f"[资源探测] 按磁盘算的并发上限更小, max-workers 收紧: "
|
||
f"{workers} → {disk_cap}(可用磁盘{free_gb:.1f}G - 保护垫{disk_guard_gb}G = "
|
||
f"{budget_gb:.1f}G预算 ÷ 每视频预估{est_gb}G)")
|
||
workers = disk_cap
|
||
except OSError:
|
||
pass
|
||
|
||
if workers > MAX_AUTO_WORKERS:
|
||
logging.info(f"[资源探测] max-workers 超过硬上限, 收紧: {workers} → {MAX_AUTO_WORKERS}")
|
||
workers = MAX_AUTO_WORKERS
|
||
return workers
|
||
|
||
|
||
_root_override = {"value": None} # run_worker.py 用: 主队列空闲时临时切去认领长视频队列
|
||
|
||
|
||
def set_active_root_override(root):
|
||
"""临时覆盖 _active_jobs_root() 的返回值(仅当前进程内生效, 不改config/不用重启)。
|
||
传 None 清除覆盖、恢复按 config 判断。给 run_worker.py 做'主队列没活时顺手兼跑长视频队列'用:
|
||
覆盖生效期间, claim/list_todo/download_shard/mark_done/release 等全部原语都会转而操作
|
||
覆盖指向的这个队列根, 跟专职长视频机(run_long_video=true)走的是同一套代码路径, 只是切换是
|
||
动态的、按需的, 而不是启动时定死。"""
|
||
_root_override["value"] = root
|
||
|
||
|
||
def _active_jobs_root():
|
||
"""本 worker 实际认领的队列根, 优先级:
|
||
_root_override(运行时临时切换) > run_long_video=true(专跑超长视频, 用 long_list_backup_path)
|
||
> jobs_root(默认主队列)。"""
|
||
if _root_override["value"]:
|
||
return _root_override["value"]
|
||
c = _cfg()
|
||
if c.get("run_long_video"):
|
||
root = c.get("long_list_backup_path")
|
||
if root:
|
||
return root
|
||
return c.get("jobs_root") or ""
|
||
|
||
|
||
def _bucket_base():
|
||
"""从【当前生效的队列根】解析 (bucket, base_prefix)。base 末尾无 '/'。
|
||
只给 _key()(todo/claimed/done 分片路径)用——这个"当前生效"是有意跟着
|
||
run_long_video/_root_override 变的(分片本来就该落在当前认领的那个队列下)。"""
|
||
root = _active_jobs_root()
|
||
if not root:
|
||
raise RuntimeError("config.dispatch 未配置队列地址(jobs_root / long_list_backup_path)")
|
||
bucket, key = parse_cos_url(root)
|
||
return bucket, key.strip("/")
|
||
|
||
|
||
def _primary_bucket_base():
|
||
"""固定解析自 jobs_root(主队列), 不受 run_long_video / 运行时 _root_override 影响 ——
|
||
daily_backup_path / monitor_root 没显式配置时的兜底基准必须用这个, 不能用 _bucket_base()
|
||
(它会跟 _active_jobs_root() 一起变): 专职长视频机(run_long_video=true)跑起来后
|
||
_active_jobs_root() 长期指向 long_list_backup_path, 如果兜底基准也跟着它变, 算出来的
|
||
monitor_root/daily_backup_path 就会落到 .../long_video/monitor、.../long_video/daily 这种
|
||
跟其它机器不一样的路径下 —— fleet_monitor.py 和 global_processed_set() 只认标准的
|
||
monitor_root/daily_backup_path, 读不到这些"另起炉灶"的路径, 表现为这台机器的小时摘要/
|
||
去重快照对外"消失了"(实际数据都在, 只是传错地方)。"""
|
||
root = _cfg().get("jobs_root") or ""
|
||
if not root:
|
||
raise RuntimeError("config.dispatch.jobs_root 未配置")
|
||
bucket, key = parse_cos_url(root)
|
||
return bucket, key.strip("/")
|
||
|
||
|
||
def _key(kind: str, shard: str = "") -> str:
|
||
"""拼某类对象的 key。kind ∈ {todo, claimed, done}。"""
|
||
_, base = _bucket_base()
|
||
ext = {"todo": ".txt", "claimed": ".lock", "done": ".done"}[kind]
|
||
if shard:
|
||
return f"{base}/{kind}/{shard}{ext}"
|
||
return f"{base}/{kind}/"
|
||
|
||
|
||
# --------------------------------------------------------------- COS 原语 ----
|
||
|
||
def _put(key: str, data: bytes, forbid_overwrite: bool = False):
|
||
bkt, _ = _bucket_base()
|
||
kwargs = {}
|
||
if forbid_overwrite:
|
||
kwargs["ForbidOverwrite"] = "true" # 经 maplist 注入 x-cos-forbid-overwrite 头
|
||
client().put_object(Bucket=bkt, Body=data, Key=key, **kwargs)
|
||
|
||
|
||
def _get_bytes(key: str):
|
||
bkt, _ = _bucket_base()
|
||
resp = client().get_object(Bucket=bkt, Key=key)
|
||
return resp["Body"].get_raw_stream().read()
|
||
|
||
|
||
def _exists(key: str) -> bool:
|
||
bkt, _ = _bucket_base()
|
||
return client().object_exists(Bucket=bkt, Key=key)
|
||
|
||
|
||
def _delete(key: str):
|
||
bkt, _ = _bucket_base()
|
||
try:
|
||
client().delete_object(Bucket=bkt, Key=key)
|
||
except CosServiceError as e:
|
||
if e.get_status_code() != 404:
|
||
raise
|
||
|
||
|
||
def _list_shards(kind: str):
|
||
"""列出某前缀下所有分片名(去掉前缀和扩展名)。"""
|
||
bkt, _ = _bucket_base()
|
||
prefix = _key(kind)
|
||
ext = {"todo": ".txt", "claimed": ".lock", "done": ".done"}[kind]
|
||
out, marker = [], ""
|
||
while True:
|
||
resp = client().list_objects(Bucket=bkt, Prefix=prefix, Marker=marker, MaxKeys=1000)
|
||
for c in resp.get("Contents", []) or []:
|
||
k = c["Key"]
|
||
if k.endswith(ext):
|
||
out.append(k[len(prefix):-len(ext)])
|
||
if resp.get("IsTruncated") in ("true", True):
|
||
marker = resp.get("NextMarker") or (resp["Contents"][-1]["Key"] if resp.get("Contents") else "")
|
||
if not marker:
|
||
break
|
||
else:
|
||
break
|
||
return out
|
||
|
||
|
||
# ----------------------------------------------------------------- 接口 ----
|
||
|
||
def list_todo():
|
||
return _list_shards("todo")
|
||
|
||
|
||
def shard_matches(name: str, patterns) -> bool:
|
||
"""分片名是否匹配 patterns(glob 通配, 如 'cos_urls_8*')。
|
||
patterns 为空/None -> 认领全部(返回 True)。可传单个字符串或列表。
|
||
匹配时对 '<名>' 和 '<名>.txt' 都试一遍, 写不写 .txt 都行。"""
|
||
import fnmatch
|
||
if not patterns:
|
||
return True
|
||
if isinstance(patterns, str):
|
||
patterns = [patterns]
|
||
cands = (name, name + ".txt")
|
||
return any(fnmatch.fnmatch(c, p) for p in patterns for c in cands)
|
||
|
||
|
||
def is_done(shard: str) -> bool:
|
||
return _exists(_key("done", shard))
|
||
|
||
|
||
def read_lock(shard: str):
|
||
"""读认领锁内容(dict)。不存在返回 None。"""
|
||
try:
|
||
return json.loads(_get_bytes(_key("claimed", shard)).decode("utf-8"))
|
||
except CosServiceError as e:
|
||
if e.get_status_code() == 404:
|
||
return None
|
||
raise
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def claim(shard: str, machine_id: str) -> bool:
|
||
"""原子认领: 写 claimed/<shard>.lock(禁止覆盖)。成功=True 表示这片归我;
|
||
409(已被别人锁) -> False。"""
|
||
now = datetime.now()
|
||
body = json.dumps({
|
||
"machine_id": machine_id, "host": socket.gethostname(),
|
||
"claim_ts": now.isoformat(), "hb_ts": now.isoformat(),
|
||
}, ensure_ascii=False).encode("utf-8")
|
||
try:
|
||
_put(_key("claimed", shard), body, forbid_overwrite=True)
|
||
return True
|
||
except CosServiceError as e:
|
||
if e.get_status_code() == 409:
|
||
return False # 已被别的机器抢到
|
||
raise
|
||
|
||
|
||
def heartbeat(shard: str, machine_id: str):
|
||
"""刷新锁的心跳时间(允许覆盖, 仍写回 machine_id 以保持所有权信息)。"""
|
||
lk = read_lock(shard) or {}
|
||
lk.update({"machine_id": machine_id, "host": socket.gethostname(),
|
||
"hb_ts": datetime.now().isoformat()})
|
||
lk.setdefault("claim_ts", lk["hb_ts"])
|
||
_put(_key("claimed", shard), json.dumps(lk, ensure_ascii=False).encode("utf-8"))
|
||
|
||
|
||
def download_shard(shard: str):
|
||
"""读 todo/<shard>.txt, 返回视频 COS 路径列表。"""
|
||
text = _get_bytes(_key("todo", shard)).decode("utf-8")
|
||
return [ln.strip() for ln in text.splitlines() if ln.strip()]
|
||
|
||
|
||
def mark_done(shard: str, machine_id: str, stats: dict):
|
||
_put(_key("done", shard), json.dumps({
|
||
"machine_id": machine_id, "stats": stats,
|
||
"end_ts": datetime.now().isoformat(),
|
||
}, ensure_ascii=False).encode("utf-8"))
|
||
|
||
|
||
def release(shard: str):
|
||
"""跑完后清理: 删 todo 和 lock(保留 done 作为完成账)。"""
|
||
_delete(_key("todo", shard))
|
||
_delete(_key("claimed", shard))
|
||
|
||
|
||
def my_open_claims(machine_id: str):
|
||
"""我名下、尚未完成(无 done)的分片 —— 重启后续做用。"""
|
||
out = []
|
||
for shard in _list_shards("claimed"):
|
||
lk = read_lock(shard)
|
||
if lk and lk.get("machine_id") == machine_id and not is_done(shard):
|
||
out.append(shard)
|
||
return out
|
||
|
||
|
||
# ------------------------------------------------------------- 优雅停机 ----
|
||
|
||
def stop_requested() -> bool:
|
||
"""COS 上存在 stop_flag_key(默认 _jobs/STOP)则请求优雅停机。"""
|
||
flag = _cfg().get("stop_flag_key")
|
||
if not flag:
|
||
_, base = _bucket_base()
|
||
flag = f"{base}/STOP"
|
||
bkt, _ = _bucket_base()
|
||
try:
|
||
return client().object_exists(Bucket=bkt, Key=flag.strip("/"))
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
# ----------------------------------------------------- 每日 数据.jsonl 备份 ----
|
||
|
||
# put_object 是"一次性读进内存 + 单个 HTTP 请求"上传, 底层 send_request 默认超时 30 秒 ——
|
||
# 数据.jsonl 是全量快照, 产能高的机器跑几天就上百 MB/上 GB, 30 秒内传不完就
|
||
# "The write operation timed out", 然后每几分钟重试又超时, 无限循环从没成功(实测 21.13.80.229)。
|
||
# 超过这个大小的文件改用 coscli 上传(它支持分块/断点续传, 几 GB 也没问题); 小文件(如几KB的
|
||
# 小时摘要)仍走 put_object —— 起 coscli 子进程有开销, 小文件没必要。
|
||
_COSCLI_UPLOAD_THRESHOLD = 50 * 1024 * 1024 # 50MB
|
||
|
||
|
||
def _coscli_put(local_path: str, cos_url: str) -> bool:
|
||
"""用 coscli 把 local_path 上传到 cos_url(完整 cos:// 路径)。成功 True, 失败 False。
|
||
coscli 支持分块上传, 大文件不像 put_object 那样受单请求超时限制。
|
||
coscli 路径/配置从 splitframe 配置读(跟主流程上传 SRT 用的是同一个二进制和配置)。"""
|
||
try:
|
||
sf = config.get_splitframe_config() or {}
|
||
coscli = sf.get("coscli_path", "./coscli-linux-amd64")
|
||
coscfg = sf.get("cosconfig_path", ".cos.yaml")
|
||
cmd = [coscli, "-c", coscfg, "cp", local_path, cos_url]
|
||
r = subprocess.run(cmd, capture_output=True, text=True, timeout=1800)
|
||
if r.returncode == 0:
|
||
return True
|
||
logging.warning(f"[备份] coscli 上传失败(将回退 put_object): {(r.stderr or '').strip()[:200]}")
|
||
return False
|
||
except Exception as e:
|
||
logging.warning(f"[备份] coscli 上传异常(将回退 put_object): {type(e).__name__}: {e}")
|
||
return False
|
||
|
||
|
||
def _upload_dated_file(local_path: str, dest_root: str, fname: str) -> str:
|
||
"""把 local_path 上传到 dest_root/<日期>/fname。
|
||
local_path 不存在 / dest_root 为空 -> 返回 ''。返回上传后的 cos:// 路径。
|
||
大文件(> _COSCLI_UPLOAD_THRESHOLD)走 coscli 分块上传, 避免 put_object 30 秒超时;
|
||
coscli 失败(二进制不在/权限等)自动回退 put_object 再试一次 —— 保证引入 coscli 不会
|
||
比原来更差(最坏情况就是退回原行为)。"""
|
||
if not local_path or not dest_root or not os.path.exists(local_path):
|
||
return ""
|
||
bkt, base_key = parse_cos_url(dest_root)
|
||
date = datetime.now().strftime("%Y%m%d") # 按日期建文件夹
|
||
key = f"{base_key.strip('/')}/{date}/{fname}"
|
||
cos_url = f"cos://{bkt}/{key}"
|
||
|
||
size = os.path.getsize(local_path)
|
||
if size > _COSCLI_UPLOAD_THRESHOLD:
|
||
logging.info(f"[备份] {fname} 有 {size / 1024 / 1024:.0f}MB, 超过 "
|
||
f"{_COSCLI_UPLOAD_THRESHOLD // 1024 // 1024}MB, 用 coscli 分块上传")
|
||
if _coscli_put(local_path, cos_url):
|
||
return cos_url
|
||
# coscli 没成: 回退 put_object(大文件很可能还是超时, 但至少不比原来差, 且日志会记清楚)
|
||
|
||
with open(local_path, "rb") as f:
|
||
data = f.read()
|
||
client().put_object(Bucket=bkt, Body=data, Key=key)
|
||
return cos_url
|
||
|
||
|
||
def upload_daily_backup(jsonl_path: str, machine_id: str) -> str:
|
||
"""把本地 数据.jsonl 上传到 dispatch.daily_backup_path/<日期>/数据_<机器号>.jsonl，
|
||
顺带生成并上传一份轻量的"已处理清单"(见 upload_processed_list) 供 global_processed_set()
|
||
用——两者是同一次备份动作的副产物，调这一个函数就够了。"""
|
||
dest_root = _cfg().get("daily_backup_path")
|
||
if not dest_root:
|
||
bucket, base = _bucket_base()
|
||
dest_root = f"cos://{bucket}/{base}/daily"
|
||
dst = _upload_dated_file(jsonl_path, dest_root, f"数据_{machine_id}.jsonl")
|
||
if dst:
|
||
try:
|
||
upload_processed_list(jsonl_path, machine_id)
|
||
except Exception as e:
|
||
logging.warning(f"[每日备份] 已处理清单上传失败(不影响 数据.jsonl 备份): {e}")
|
||
return dst
|
||
|
||
|
||
def upload_longlist_backup(machine_id: str) -> str:
|
||
"""把本地超长待重跑列表(subtitle.long_video_list)上传到
|
||
dispatch.long_list_backup_path/todo/<原文件名>_<机器号>_<日期>_<时分秒>.txt。
|
||
放进 todo/ 子目录, 使其直接成为可认领的任务分片 —— 把别的机器的 jobs_root
|
||
指向 long_list_backup_path, 它们就能自动认领这些超长视频重跑。
|
||
文件名带【机器号+日期+时分秒】: 机器号保证各机器不互相覆盖; 时分秒是因为本函数是
|
||
"传完即清空本地已传部分"的增量式设计(见下), 一天内不管自动备份触发几次、或人工额外
|
||
手动跑几次(如 tools/upload_daily.py / apply_update.sh 更新前那一下), 每次快照的都只是
|
||
"上次传完之后新增的那一小段" —— 如果 key 只精确到天, 同一天第二次上传会用 put_object
|
||
直接覆盖同名对象, 把第一次已传但可能还没被任何机器认领处理的内容连本地带COS一起冲没。
|
||
精确到秒后每次上传各自独立成对象, 不会互相覆盖, 只是分片数量会随调用次数变多(不影响使用,
|
||
worker 一样能各自认领)。"""
|
||
# 专跑长视频的机器(run_long_video)不再把超长列表传回同一队列, 防回环。
|
||
if _cfg().get("run_long_video"):
|
||
return ""
|
||
local = (config.get_subtitle_config() or {}).get("long_video_list", "./超长待重跑.txt")
|
||
if not local or not os.path.exists(local):
|
||
return ""
|
||
# 快照当前内容(文件 append-only); 全空则不传
|
||
with open(local, "r", encoding="utf-8") as f:
|
||
snapshot = f.readlines()
|
||
if not any(ln.strip() for ln in snapshot):
|
||
return ""
|
||
dest_root = _cfg().get("long_list_backup_path")
|
||
if not dest_root:
|
||
bucket, base = _bucket_base()
|
||
dest_root = f"cos://{bucket}/{base}/long_video"
|
||
bkt, base_key = parse_cos_url(dest_root)
|
||
stem = os.path.splitext(os.path.basename(local))[0] # 超长待重跑
|
||
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
key = f"{base_key.strip('/')}/todo/{stem}_{machine_id}_{ts}.txt"
|
||
client().put_object(Bucket=bkt, Body="".join(snapshot).encode("utf-8"), Key=key)
|
||
# 上传成功后清空已上传部分: 只删前 len(snapshot) 行, 保留上传期间新追加的(append 在末尾),
|
||
# 这样每天只传增量、跑完即了, 不会和已完成分片重叠。开关 long_list_clear_after_upload(默认开)。
|
||
if _cfg().get("long_list_clear_after_upload", True):
|
||
try:
|
||
with open(local, "r", encoding="utf-8") as f:
|
||
cur = f.readlines()
|
||
with open(local, "w", encoding="utf-8") as f:
|
||
f.writelines(cur[len(snapshot):])
|
||
except OSError:
|
||
pass
|
||
return f"cos://{bkt}/{key}"
|
||
|
||
|
||
def _daily_backup_bucket_base():
|
||
"""daily_backup_path 配置解析为 (bucket, base_key)，缺省时与 upload_daily_backup 保持一致。
|
||
兜底基准用 _primary_bucket_base()(固定挂 jobs_root), 不能用 _bucket_base() —— 见后者docstring
|
||
里的解释, 否则专职长视频机会把每日备份传到 long_video 队列自己底下, 全局去重读不到。"""
|
||
dest_root = _cfg().get("daily_backup_path")
|
||
if not dest_root:
|
||
bucket, base = _primary_bucket_base()
|
||
dest_root = f"cos://{bucket}/{base}/daily"
|
||
bucket, base = parse_cos_url(dest_root)
|
||
return bucket, base.strip("/")
|
||
|
||
|
||
def _extract_processed_paths(jsonl_path: str) -> list:
|
||
"""从本地 数据.jsonl 逐行解析出 video_cos_path 列表(顺序不去重, 上传前用)。"""
|
||
paths = []
|
||
try:
|
||
with open(jsonl_path, "r", encoding="utf-8") as f:
|
||
for line in f:
|
||
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
|
||
except OSError:
|
||
pass
|
||
return paths
|
||
|
||
|
||
def upload_processed_list(jsonl_path: str, machine_id: str) -> str:
|
||
"""从本地 数据.jsonl 提取 video_cos_path, 生成一份纯文本清单(一行一个路径, 不含状态/耗时等
|
||
其它字段)上传到 daily_backup_path/<日期>/已处理_<机器号>.txt —— 跟 upload_daily_backup 上传的
|
||
完整 jsonl 是同一份数据的两种形态, 前者给 global_processed_set() 用: 只关心"这个视频是否
|
||
处理过"时, 拉这个纯清单比拉完整jsonl再挨行解析JSON 小得多、快得多(jsonl里每条记录还带
|
||
timestamp/duration/image_counts/stage_durations 等一堆跟去重无关的字段)。"""
|
||
dest_root = _cfg().get("daily_backup_path")
|
||
if not dest_root:
|
||
bucket, base = _bucket_base()
|
||
dest_root = f"cos://{bucket}/{base}/daily"
|
||
paths = _extract_processed_paths(jsonl_path)
|
||
if not paths:
|
||
return ""
|
||
bkt, base_key = parse_cos_url(dest_root)
|
||
date = datetime.now().strftime("%Y%m%d")
|
||
key = f"{base_key.strip('/')}/{date}/已处理_{machine_id}.txt"
|
||
client().put_object(Bucket=bkt, Body=("\n".join(paths) + "\n").encode("utf-8"), Key=key)
|
||
return f"cos://{bkt}/{key}"
|
||
|
||
|
||
def list_processed_list_files():
|
||
"""列出 daily_backup_path 下所有 <日期>/已处理_<机器号>.txt，返回 [(日期, 机器号, key), ...]。"""
|
||
bkt, base = _daily_backup_bucket_base()
|
||
prefix = base + "/"
|
||
out, marker = [], ""
|
||
while True:
|
||
resp = client().list_objects(Bucket=bkt, Prefix=prefix, Marker=marker, MaxKeys=1000)
|
||
for c in resp.get("Contents", []) or []:
|
||
rel = c["Key"][len(prefix):] # <日期>/已处理_<机器号>.txt
|
||
parts = rel.split("/")
|
||
if len(parts) == 2 and parts[1].startswith("已处理_") and parts[1].endswith(".txt"):
|
||
out.append((parts[0], parts[1][len("已处理_"):-len(".txt")], c["Key"]))
|
||
if resp.get("IsTruncated") in ("true", True):
|
||
marker = resp.get("NextMarker") or (resp["Contents"][-1]["Key"] if resp.get("Contents") else "")
|
||
if not marker:
|
||
break
|
||
else:
|
||
break
|
||
return out
|
||
|
||
|
||
def global_processed_set(verbose: bool = False, all_dates: bool = False) -> set:
|
||
"""聚合"已处理清单"(已处理_<机器号>.txt, 由 upload_processed_list 在每次备份时跟 数据.jsonl
|
||
一起生成), 合并成全局已处理 video_cos_path 集合。只拉轻量txt清单、不下载/解析完整jsonl,
|
||
省流量也省CPU。语义与本机 load_processed() 一致：不分 success/no_subtitle/error，只要出现过
|
||
就算处理过。用于灌任务前过滤，避免同一视频被 30 台机器重复处理。
|
||
|
||
all_dates=False(默认, 日常灌任务用): 每台机器只取【最新一天】——数据.jsonl 是全量快照(本地
|
||
从不清空), 取最新一份即覆盖该机器全部历史, 省得把每天的备份都下载一遍; 不同机器最新日期
|
||
可能不同(比如有机器当天没开机备份), 各自独立取最新, 不要求同一天。
|
||
|
||
all_dates=True(机群已停机/一次性核对时用): 不做"每机器只取最新"这个假设, 把【所有日期】
|
||
的清单全部下载合并——机群都停了就没有"之后还会补传更晚日期"这回事, 更保险的做法是把能拿到
|
||
的所有历史清单全并起来判断"处理过没有", 而不是赌某一天的快照就是最全的(万一那台机器最新
|
||
一天的快照因为某种原因不完整/被截断过, 只取最新反而会漏)。代价是下载的文件数从"机器数"
|
||
变成"机器数×历史天数", 视频清单量小(轻量txt)时这个代价可以接受。"""
|
||
files = list_processed_list_files()
|
||
if not files:
|
||
return set()
|
||
|
||
if all_dates:
|
||
targets = [(date_str, machine_id, key) for date_str, machine_id, key in files]
|
||
else:
|
||
latest = {} # machine_id -> (日期, key)
|
||
for date_str, machine_id, key in files:
|
||
if machine_id not in latest or date_str > latest[machine_id][0]:
|
||
latest[machine_id] = (date_str, key)
|
||
targets = [(date_str, machine_id, key) for machine_id, (date_str, key) in latest.items()]
|
||
|
||
bkt, _ = _daily_backup_bucket_base()
|
||
processed = set()
|
||
for date_str, machine_id, key in targets:
|
||
try:
|
||
data = client().get_object(Bucket=bkt, Key=key)["Body"].get_raw_stream().read()
|
||
except Exception as e:
|
||
logging.warning(f"[全局去重] 读取 {key} 失败, 跳过: {e}")
|
||
continue
|
||
n = 0
|
||
for line in data.decode("utf-8", errors="ignore").splitlines():
|
||
line = line.strip()
|
||
if line:
|
||
processed.add(line)
|
||
n += 1
|
||
if verbose:
|
||
logging.info(f"[全局去重] 机器 {machine_id}({date_str}): {n} 条")
|
||
return processed
|
||
|
||
|
||
def run_daily_backup_loop(machine_id: str, hour: int, stop, jsonl_path: str = "数据.jsonl",
|
||
check_interval: int = 120):
|
||
"""后台线程: 每天到 hour 点把 数据.jsonl + 超长列表上传 COS, 每天只传一次。
|
||
main.py(本地模式) 和 run_worker.py(COS模式) 通用。stop 为 threading.Event, set() 即退出。"""
|
||
last_date = None
|
||
while not stop.wait(check_interval): # 每 check_interval 秒检查一次
|
||
now = datetime.now()
|
||
today = now.strftime("%Y%m%d")
|
||
if now.hour >= hour and last_date != today:
|
||
try:
|
||
dst = upload_daily_backup(jsonl_path, machine_id)
|
||
if dst:
|
||
logging.info(f"[每日备份] 数据.jsonl 已上传 -> {dst}")
|
||
else:
|
||
logging.info("[每日备份] 本地无 数据.jsonl, 跳过")
|
||
ll = upload_longlist_backup(machine_id)
|
||
if ll:
|
||
logging.info(f"[每日备份] 超长待重跑列表已上传 -> {ll}")
|
||
last_date = today # 当天标记完成, 避免反复触发
|
||
except Exception as e:
|
||
logging.error(f"[每日备份] 上传失败(稍后重试): {e}")
|
||
|
||
|
||
# ----------------------------------------------------- 每小时监控摘要 ----
|
||
|
||
def _monitor_bucket_base():
|
||
"""monitor_root 配置解析为 (bucket, base_key)，缺省时 = jobs_root 的同级 monitor 目录
|
||
(跟 _daily_backup_bucket_base 同一个派生模式)。兜底基准同样必须用 _primary_bucket_base()
|
||
而不是 _bucket_base() —— 这就是本函数曾经的实际bug: 专职长视频机(run_long_video=true)
|
||
没显式配 monitor_root 时, 旧版用 _bucket_base() 兜底会解析到【当前生效队列根】
|
||
(_active_jobs_root(), 长视频机上长期等于 long_list_backup_path), 算出
|
||
.../long_video/monitor/... 这种独立于其它机器的路径, fleet_monitor.py 只读标准
|
||
monitor_root, 找不到这些机器的小时摘要, 表现为"全队机器数对不上"(实际数据都传了,
|
||
只是传错地方, 数据没丢, 改这个函数后新上传的会去对的地方, 旧的仍留在错的路径下)。"""
|
||
dest_root = _cfg().get("monitor_root")
|
||
if not dest_root:
|
||
bucket, base = _primary_bucket_base()
|
||
dest_root = f"cos://{bucket}/{base}/monitor"
|
||
bucket, base = parse_cos_url(dest_root)
|
||
return bucket, base.strip("/")
|
||
|
||
|
||
def compute_hourly_digest(jsonl_path: str, hour_dt: datetime):
|
||
"""从本地 数据.jsonl 里筛出 process_start(缺失回退timestamp) 落在 hour_dt 所在整点小时的记录,
|
||
算出这个小时的摘要。hour_dt 的分钟/秒会被忽略, 只取年月日时。
|
||
本地jsonl不存在/读取失败 -> 返回 None(调用方应跳过上传)。
|
||
这一小时确实一条记录都没处理(count==0, 比如COS任务队列暂时被抢空、worker在idle轮询)
|
||
-> 仍然返回一份全0的摘要, 不当 None 处理 —— 队列见底idle是这套分布式调度系统的正常状态
|
||
(见 run_worker.py 的 keep_alive/idle_poll_seconds), 如果这种情况也不传摘要, fleet_monitor.py
|
||
的"机器掉线"检测会把"确认活着只是没活干"误判成"可能停机", 制造假警报。"""
|
||
if not os.path.exists(jsonl_path):
|
||
return None
|
||
hour_prefix = hour_dt.strftime("%Y-%m-%dT%H") # 匹配 ISO 格式 process_start 的前13位
|
||
count = success = no_subtitle = error = 0
|
||
success_hours = 0.0
|
||
try:
|
||
with open(jsonl_path, "r", encoding="utf-8") as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
rec = json.loads(line)
|
||
except json.JSONDecodeError:
|
||
continue
|
||
ts = rec.get("process_start") or rec.get("timestamp") or ""
|
||
if ts[:13] != hour_prefix:
|
||
continue
|
||
count += 1
|
||
status = rec.get("status")
|
||
if status == "success":
|
||
success += 1
|
||
try:
|
||
success_hours += float(rec.get("duration") or 0) / 3600.0
|
||
except (TypeError, ValueError):
|
||
pass
|
||
elif status == "no_subtitle":
|
||
no_subtitle += 1
|
||
else:
|
||
error += 1
|
||
except OSError:
|
||
return None
|
||
return {
|
||
"machine_id": None, # upload_hourly_digest 里填
|
||
"hour": hour_dt.strftime("%Y-%m-%dT%H"),
|
||
"count": count, "success": success, "no_subtitle": no_subtitle, "error": error,
|
||
"success_hours": round(success_hours, 2),
|
||
"uploaded_at": datetime.now().isoformat(),
|
||
}
|
||
|
||
|
||
def upload_hourly_digest(machine_id: str, jsonl_path: str, hour_dt: datetime) -> str:
|
||
"""算好 hour_dt 那个整点小时的摘要并上传到
|
||
monitor_root/hourly/<日期YYYYMMDD>/<小时HH>/<机器号>.json。
|
||
即使这一小时count==0(队列见底idle)也会正常上传一份全0摘要——只有本地 数据.jsonl
|
||
根本不存在/读取失败(digest为None, 比如worker刚起来还没写过jsonl)才跳过, 返回''。"""
|
||
digest = compute_hourly_digest(jsonl_path, hour_dt)
|
||
if digest is None:
|
||
return ""
|
||
digest["machine_id"] = machine_id
|
||
bkt, base = _monitor_bucket_base()
|
||
date_str = hour_dt.strftime("%Y%m%d")
|
||
hour_str = hour_dt.strftime("%H")
|
||
key = f"{base}/hourly/{date_str}/{hour_str}/{machine_id}.json"
|
||
client().put_object(Bucket=bkt, Body=json.dumps(digest, ensure_ascii=False).encode("utf-8"), Key=key)
|
||
return f"cos://{bkt}/{key}"
|
||
|
||
|
||
def list_hourly_digest_files_for_hour(hour_dt: datetime) -> list:
|
||
"""列出 monitor_root/hourly/<日期>/<小时>/ 下所有机器的摘要文件, 返回 [(机器号, key), ...]。"""
|
||
bkt, base = _monitor_bucket_base()
|
||
date_str = hour_dt.strftime("%Y%m%d")
|
||
hour_str = hour_dt.strftime("%H")
|
||
prefix = f"{base}/hourly/{date_str}/{hour_str}/"
|
||
out, marker = [], ""
|
||
while True:
|
||
resp = client().list_objects(Bucket=bkt, Prefix=prefix, Marker=marker, MaxKeys=1000)
|
||
for c in resp.get("Contents", []) or []:
|
||
rel = c["Key"][len(prefix):]
|
||
if rel.endswith(".json"):
|
||
out.append((rel[:-len(".json")], c["Key"]))
|
||
if resp.get("IsTruncated") in ("true", True):
|
||
marker = resp.get("NextMarker") or (resp["Contents"][-1]["Key"] if resp.get("Contents") else "")
|
||
if not marker:
|
||
break
|
||
else:
|
||
break
|
||
return out
|
||
|
||
|
||
def download_hourly_digest(key: str) -> dict:
|
||
"""下载解析单份小时摘要json, 返回dict。失败抛异常, 调用方负责捕获。"""
|
||
bkt, _ = _monitor_bucket_base()
|
||
data = client().get_object(Bucket=bkt, Key=key)["Body"].get_raw_stream().read()
|
||
return json.loads(data.decode("utf-8"))
|
||
|
||
|
||
def upload_monitor_report(local_path: str, hour_dt: datetime) -> str:
|
||
"""把生成好的监控报告文件上传到 monitor_root/reports/<日期YYYYMMDD>_<小时HH>.md。"""
|
||
if not os.path.exists(local_path):
|
||
return ""
|
||
bkt, base = _monitor_bucket_base()
|
||
date_str = hour_dt.strftime("%Y%m%d")
|
||
hour_str = hour_dt.strftime("%H")
|
||
key = f"{base}/reports/{date_str}_{hour_str}.md"
|
||
with open(local_path, "rb") as f:
|
||
data = f.read()
|
||
client().put_object(Bucket=bkt, Body=data, Key=key)
|
||
return f"cos://{bkt}/{key}"
|
||
|
||
|
||
def run_hourly_digest_loop(machine_id: str, stop, jsonl_path: str = "数据.jsonl",
|
||
check_interval: int = 120):
|
||
"""后台线程: 每小时把刚结束的那个整点小时的处理摘要上传COS, 每小时只传一次。
|
||
main.py(本地模式) 不需要接入(一次性批处理没有"常驻小时"的概念, 见设计文档)。
|
||
stop 为 threading.Event, set() 即退出。"""
|
||
last_uploaded_hour = None
|
||
while not stop.wait(check_interval):
|
||
now = datetime.now()
|
||
prev_hour_dt = now.replace(minute=0, second=0, microsecond=0) - timedelta(hours=1)
|
||
prev_hour_key = prev_hour_dt.strftime("%Y%m%d%H")
|
||
if last_uploaded_hour == prev_hour_key:
|
||
continue
|
||
try:
|
||
dst = upload_hourly_digest(machine_id, jsonl_path, prev_hour_dt)
|
||
if dst:
|
||
logging.info(f"[小时摘要] {prev_hour_key} 已上传 -> {dst}")
|
||
else:
|
||
logging.info(f"[小时摘要] {prev_hour_key} 本地 数据.jsonl 尚不存在, 跳过上传")
|
||
last_uploaded_hour = prev_hour_key
|
||
except Exception as e:
|
||
logging.error(f"[小时摘要] {prev_hour_key} 上传失败(下次继续重试): {e}")
|