diff --git a/fam-core/src/fam_core/dispatcher/dispatcher.py b/fam-core/src/fam_core/dispatcher/dispatcher.py index 67849de..81562ab 100644 --- a/fam-core/src/fam_core/dispatcher/dispatcher.py +++ b/fam-core/src/fam_core/dispatcher/dispatcher.py @@ -46,6 +46,15 @@ FFMPEG_CANDIDATES = [ ] +def _safe_remove(path): + """忽略不存在/清理失败的删除""" + try: + if os.path.isfile(path): + os.remove(path) + except OSError: + pass + + class Dispatcher: """任务下发器,30s 轮询(异步队列 + 分块断点续传)""" @@ -156,6 +165,11 @@ class Dispatcher: src_size = os.path.getsize(video_path) start = time.time() + # 写临时文件(带 PID 防多进程冲突),成功后原子 rename — + # 服务被 kill 时 ffmpeg 成为孤儿继续写 tmp,缓存目录中 + # 只会出现完整产物,杜绝半成品被复用(曾导致上传损坏视频) + tmp_path = f"{out_path}.{os.getpid()}.tmp" + cmd = [ self.ffmpeg, '-y', '-i', video_path, @@ -164,7 +178,7 @@ class Dispatcher: 'scale=trunc(iw/2)*2:trunc(ih/2)*2'), '-c:v', 'libx264', '-preset', 'veryfast', '-crf', '28', '-an', - out_path, + tmp_path, ] try: result = subprocess.run( @@ -172,17 +186,21 @@ class Dispatcher: timeout=self.compress_timeout, ) except subprocess.TimeoutExpired: + _safe_remove(tmp_path) logger.error(f"[task_id={task_id}] 压缩超时 ({self.compress_timeout}s),回退原始上传") return None except OSError as e: logger.error(f"[task_id={task_id}] 启动 ffmpeg 失败: {e}") return None - if result.returncode != 0 or not os.path.isfile(out_path) or os.path.getsize(out_path) == 0: + if result.returncode != 0 or not os.path.isfile(tmp_path) or os.path.getsize(tmp_path) == 0: + _safe_remove(tmp_path) stderr_tail = (result.stderr or '')[-300:] logger.error(f"[task_id={task_id}] 压缩失败 (rc={result.returncode}): {stderr_tail}") return None + os.replace(tmp_path, out_path) + dst_size = os.path.getsize(out_path) elapsed = time.time() - start logger.info(f"[task_id={task_id}] 预压缩完成: {src_size/1048576:.1f}MB → " @@ -191,7 +209,7 @@ class Dispatcher: @staticmethod def _cleanup_compress_cache(): - """清理超过 TTL 的压缩缓存目录(tmpfs 空间有限)""" + """清理超过 TTL 的压缩缓存目录 + 孤儿 .tmp 残留(tmpfs 空间有限)""" try: if not os.path.isdir(COMPRESS_DIR): return @@ -199,8 +217,15 @@ class Dispatcher: for entry in os.listdir(COMPRESS_DIR): path = os.path.join(COMPRESS_DIR, entry) try: - if os.path.isdir(path) and os.path.getmtime(path) < cutoff: - shutil.rmtree(path, ignore_errors=True) + if os.path.isdir(path): + if os.path.getmtime(path) < cutoff: + shutil.rmtree(path, ignore_errors=True) + continue + # 清理超过 1h 的 .tmp 残留(孤儿 ffmpeg 产物) + for fn in os.listdir(path): + if fn.endswith('.tmp') and \ + os.path.getmtime(os.path.join(path, fn)) < time.time() - 3600: + _safe_remove(os.path.join(path, fn)) except OSError: continue except OSError: