Fix watermark detection + S3 upload + NVENC color + task download on S3 failure

- smart_detect.py: rewrite detect_static_regions using saturation-signature method
  (high-sat logo frames vs no-logo frames diff) to handle animated/periodic watermarks
  that break temporal-std approach; merge top-region boxes; add print debugging
- web_app.py: fix S3 upload (async chunked streaming, error visible in task log);
  s3_failed flag keeps task downloadable even when S3 upload fails; add s3_failed
  column to DB; download/stream endpoints allow s3_failed tasks
- web/index.html: show play/download buttons for s3_failed tasks; task sort by
  created_at desc; watermark mode label update; isLocalDone() helper
- video_io.py: add -vf format=yuv420p before h264_nvenc to fix green video

Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>

backend/tools/smart_detect.py

@@ -332,15 +332,13 @@ def detect_color_logo(video_path: str, sample_n: int = 40):
     return merged
 
 
-def detect_watermark_per_frame(video_path: str, sample_step: int = 2, pad: int = 6):
+def detect_watermark_per_frame(video_path: str, sample_step: int = None, pad: int = 6):
     """逐帧检测半透明/移动文字水印，返回 {frame_no: [(xmin,xmax,ymin,ymax), ...]}。
 
     适用于 OCR 检测不到的淡色水印（白字、半透明）以及垂直滚动水印。
-    原理：
-      1. 每帧 white top-hat 提取亮文字笔画 → 候选框
-      2. 粗网格投票：水印（固定或缓慢移动）会让相邻网格被大量帧命中，
-         瞬时亮点（皮肤反光等）命中少 → 过滤掉
-      3. 每帧只保留落在「水印网格」内的候选框，并补全相邻帧（插值）
+    采样：每秒一帧（cap.set 跳帧，长视频也快）。
+    判据：基于轨迹自身（绝对出现秒数 + 出现时段内密度），不依赖视频总长，
+          因此「只在部分时段出现的滚动水印」「贯穿全程偶有漏检的固定水印」都能被保留。
     """
     cap = cv2.VideoCapture(video_path)
     total = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
@@ -351,88 +349,111 @@ def detect_watermark_per_frame(video_path: str, sample_step: int = 2, pad: int =
     tophat_kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (15, 15))
     conn_kernel = cv2.getStructuringElement(cv2.MORPH_RECT, (25, 9))
 
-    # 阶段1：逐帧候选（1-based 帧号，与 find_subtitle_frame_no/inpaint 管线一致）
-    frame_boxes = {}
-    fno = 0
-    while True:
+    # 阶段1：每秒一帧采样（cap.set 跳帧；frame_no 用 1-based 与 inpaint 管线一致）
+    step = max(1, int(round(fps)))      # 每秒对应的帧步长
+    sample_frames = list(range(0, total, step))
+    frame_boxes = {}                    # sample_idx -> [box]
+    sidx_to_fno = {}                    # sample_idx -> 1-based 帧号
+    for si, pos in enumerate(sample_frames):
+        cap.set(cv2.CAP_PROP_POS_FRAMES, pos)
         ret, frame = cap.read()
         if not ret:
-            break
-        fno += 1
-        if (fno - 1) % sample_step == 0:
+            continue
         gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
-            frame_boxes[fno] = _frame_text_candidates(gray, tophat_kernel, conn_kernel)
+        frame_boxes[si] = _frame_text_candidates(gray, tophat_kernel, conn_kernel)
+        sidx_to_fno[si] = pos + 1
     cap.release()
 
     n_sampled = len(frame_boxes)
     if n_sampled == 0:
         return {}
 
-    # 阶段2：轨迹连接
-    # 真水印（固定/垂直滚动）跨帧形成连续轨迹：宽度稳定、X几乎不变、Y平滑变化。
-    # 人体亮区(皮肤/内衣)断续出现、形状多变 → 轨迹短或宽度抖动 → 被过滤。
-    sampled_nos = sorted(frame_boxes.keys())
+    # 阶段2：轨迹连接（按「采样序号」相邻，间断容忍 3 个采样点=3秒）
+    sampled_sidx = sorted(frame_boxes.keys())
 
     def cx_of(b): return (b[0] + b[1]) / 2
     def cy_of(b): return (b[2] + b[3]) / 2
     def wid_of(b): return b[1] - b[0]
 
-    tracks = []  # 每条: {"boxes": {fno:box}, "last_f":, "last_box":}
-    for f in sampled_nos:
-        for b in frame_boxes[f]:
+    GAP_TOL = 5          # 允许漏检 5 个采样点（5秒，滚动水印经过浅背景会暂时漏检）
+    tracks = []          # {"boxes": {sidx:box}, "last_si":, "last_box":}
+    for si in sampled_sidx:
+        for b in frame_boxes[si]:
             best, best_d = None, 1e9
             for tr in tracks:
-                lb = tr["last_box"]
-                # 已结束的轨迹（间隔太久）跳过
-                if f - tr["last_f"] > sample_step * 6:
+                if si - tr["last_si"] > GAP_TOL:
                     continue
+                lb = tr["last_box"]
+                gap = si - tr["last_si"]
                 dx = abs(cx_of(b) - cx_of(lb))
-                dy = abs(cy_of(b) - cy_of(lb))
                 dw = abs(wid_of(b) - wid_of(lb))
-                # 同一水印：X几乎不变(<40)、宽度接近(<25)、Y可移动(滚动)
-                if dx < 40 and dw < 25 and dy < 120:
-                    d = dx * 2 + dw * 2 + dy * 0.3
+                dy = abs(cy_of(b) - cy_of(lb))
+                # 同一水印：X几乎不变、宽度接近；Y 可滚动（按间隔放宽，每秒最多移动~60px）
+                if dx < 45 and dw < 28 and dy < 60 * max(1, gap):
+                    d = dx * 2 + dw * 2 + dy * 0.2
                     if d < best_d:
                         best, best_d = tr, d
             if best is None:
-                tracks.append({"boxes": {f: b}, "last_f": f, "last_box": b})
+                tracks.append({"boxes": {si: b}, "last_si": si, "last_box": b})
             else:
-                best["boxes"][f] = b
-                best["last_f"] = f
+                best["boxes"][si] = b
+                best["last_si"] = si
                 best["last_box"] = b
 
-    # 阶段3：筛选有效轨迹
-    # 出现帧数 >= 35% 且 宽度稳定（变异系数<0.35）
-    min_frames = max(3, int(n_sampled * 0.35))
-    per_frame = {}
-    valid_tracks = 0
+    # 阶段3：筛选有效轨迹 —— 不依赖视频总长
+    #   ① 至少出现 MIN_SECONDS 秒（绝对阈值，滤掉一闪而过的噪点）
+    #   ② 在其出现时段 [first,last] 内密度高（持续存在，不是零散巧合）
+    #   ③ 宽度稳定（文字水印宽度恒定，人体亮区会抖）
+    MIN_SECONDS = 4
+    valid = []   # 有效轨迹的 boxes dict 列表
     for tr in tracks:
         boxes = tr["boxes"]
-        if len(boxes) < min_frames:
+        n_pts = len(boxes)
+        if n_pts < MIN_SECONDS:
+            continue
+        sids = sorted(boxes.keys())
+        span = sids[-1] - sids[0] + 1
+        density = n_pts / span
+        if density < 0.5:
             continue
         widths = [wid_of(b) for b in boxes.values()]
         wmean = sum(widths) / len(widths)
         wstd = (sum((x - wmean) ** 2 for x in widths) / len(widths)) ** 0.5
-        if wmean > 0 and wstd / wmean > 0.35:
-            continue   # 宽度抖动大 → 不是文字水印
-        valid_tracks += 1
-        for f, b in boxes.items():
-            x1, x2, y1, y2 = b
-            per_frame.setdefault(f, []).append(
-                (max(0, x1 - pad), min(w, x2 + pad), max(0, y1 - pad), min(h, y2 + pad)))
+        if wmean > 0 and wstd / wmean > 0.4:
+            continue
+        valid.append(boxes)
 
-    print(f"[WM-perframe] 采样{n_sampled}帧 轨迹{len(tracks)}条 有效{valid_tracks}条(>={min_frames}帧)", flush=True)
+    print(f"[WM-perframe] 每秒一帧采样{n_sampled}帧 轨迹{len(tracks)}条 有效{len(valid)}条", flush=True)
 
-    # 阶段4：插值填充文字水印框（用最近的已检测帧）
+    # 阶段4：每条有效轨迹在其 [first,last] 时段内逐采样点线性插值，
+    #   补全中间漏检的采样点（滚动水印经过浅背景漏检时也能被覆盖），
+    #   再把每个采样点的框展开到对应的 1 秒帧区间，y 方向外扩覆盖滚动移动。
+    ROLL_PAD = 16
     full = {}
-    pf_nos = sorted(per_frame.keys())
-    if pf_nos:
-        for f, nf in zip(pf_nos, pf_nos[1:]):
-            full[f] = list(per_frame[f])
-            if nf - f <= sample_step * 3:
-                for fill in range(f + 1, nf):
-                    full[fill] = list(per_frame[f])
-        full[pf_nos[-1]] = list(per_frame[pf_nos[-1]])
+
+    def lerp_box(ba, bb, t):
+        return tuple(int(round(ba[i] + (bb[i] - ba[i]) * t)) for i in range(4))
+
+    for boxes in valid:
+        sids = sorted(boxes.keys())
+        first, last = sids[0], sids[-1]
+        for si in range(first, last + 1):
+            if si in boxes:
+                b = boxes[si]
+            else:
+                # 线性插值：前后最近的已知采样点
+                prev = max(s for s in sids if s < si)
+                nxt = min(s for s in sids if s > si)
+                b = lerp_box(boxes[prev], boxes[nxt], (si - prev) / (nxt - prev))
+            x1, x2, y1, y2 = b
+            bx = (max(0, x1 - pad), min(w, x2 + pad),
+                  max(0, y1 - ROLL_PAD), min(h, y2 + ROLL_PAD))
+            # 该采样点覆盖 [本帧, 下一采样帧) 的所有原始帧
+            f_start = si * step + 1
+            f_end = (si + 1) * step + 1
+            for f in range(f_start, min(f_end, total + 1)):
+                full.setdefault(f, [])
+                full[f].append(bx)
 
     # 阶段5：彩色 logo（高饱和+静止）→ 静止框，注入所有帧
     color_logos = detect_color_logo(video_path)

web/index.html

@@ -944,7 +944,7 @@ function renderTasks() {
       <td style="white-space:nowrap;font-size:12px;color:var(--muted)">${fmtDur(t)}</td>
       <td>
         <div style="display:flex;gap:6px">
-          ${t.status==='done' ? `
+          ${isLocalDone(t) ? `
             <button class="button secondary" style="min-height:28px;padding:0 10px;font-size:12px" onclick="event.stopPropagation();playVideo('${t.id}')">▶ 播放</button>
           ` : ''}
           <button class="button secondary" style="min-height:28px;padding:0 10px;font-size:12px;color:var(--bad)" onclick="event.stopPropagation();deleteTask('${t.id}')">✕</button>
@@ -954,7 +954,13 @@ function renderTasks() {
   }).join('');
 }
 
+function isLocalDone(t) {
+  // 本地处理完成（含 S3 上传失败但文件在本地的情况）
+  return t.status === 'done' || (t.status === 'error' && t.s3_failed && t.output_path);
+}
+
 function taskStatus(t) {
+  if (t.status === 'error' && t.s3_failed) return { cls:'warn', label:'完成(S3失败)' };
   const map = {
     pending:     { cls:'neutral', label:'等待中' },
     downloading: { cls:'warn',    label:'下载中'  },
@@ -994,6 +1000,10 @@ function progressCell(t) {
   if (t.status === 'done') {
     return `<div class="prog-track"><div class="prog-fill good" style="width:100%"></div></div>`;
   }
+  if (t.status === 'error' && t.s3_failed) {
+    return `<div class="prog-track"><div class="prog-fill good" style="width:100%"></div></div>
+            <div style="font-size:10px;color:var(--warn);margin-top:2px">S3上传失败，本地可下载</div>`;
+  }
   if (t.status === 'error') {
     return `<div style="font-size:11px;color:var(--bad);max-width:180px;overflow:hidden;text-overflow:ellipsis;white-space:nowrap" title="${esc(t.error||'')}">${esc((t.error||'').slice(0,60))}</div>`;
   }
@@ -1097,12 +1107,14 @@ function updateDetailDynamic(t) {
   }
 
   const actEl = document.getElementById('detailActions');
-  if (actEl && t.status === 'done') {
+  if (actEl && isLocalDone(t)) {
     // 只在完成时补充按钮，避免重复加
     if (!actEl.querySelector('.btn-output')) {
+      const s3Warn = t.s3_failed ? `<span style="font-size:11px;color:var(--warn);margin-left:8px">⚠ S3上传失败</span>` : '';
       actEl.insertAdjacentHTML('beforeend', `
         <button class="button btn-output" style="min-height:32px;padding:0 14px;font-size:13px" onclick="playVideo('${t.id}', 'output')">▶ 处理结果</button>
         <a href="/api/tasks/${t.id}/download" download class="button secondary" style="min-height:32px;padding:0 14px;font-size:13px;text-decoration:none">↓ 下载</a>
+        ${s3Warn}
       `);
     }
   }

web_app.py

@@ -194,6 +194,7 @@ def init_db():
         ("platform_task_id", "INTEGER"),
         ("started_at", "REAL"),
         ("completed_at", "REAL"),
+        ("s3_failed", "BOOLEAN DEFAULT 0"),
     ]:
         try:
             conn.execute(f"ALTER TABLE tasks ADD COLUMN {col} {default}")
@@ -291,8 +292,8 @@ def _save_task_to_db(task: dict):
             (id, url, is_cache_file, cache_path, inpaint_mode, subtitle_areas,
              preview_mode, detect_mode, status, progress, download_progress,
              cached, output_path, error, platform_task_id,
-             created_at, started_at, completed_at)
-            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+             created_at, started_at, completed_at, s3_failed)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
         """, (
             task["id"], task["url"], task.get("is_cache_file", False),
             task.get("cache_path", ""), task["inpaint_mode"],
@@ -301,6 +302,7 @@ def _save_task_to_db(task: dict):
             task["download_progress"], task["cached"], task["output_path"],
             task["error"], task.get("platform_task_id"),
             task.get("created_at"), task.get("started_at"), task.get("completed_at"),
+            task.get("s3_failed", False),
         ))
         conn.commit()
         conn.close()
@@ -330,6 +332,7 @@ def _load_tasks_from_db():
                 "error": r["error"], "platform_task_id": r.get("platform_task_id"),
                 "created_at": r.get("created_at"), "started_at": r.get("started_at"),
                 "completed_at": r.get("completed_at"),
+                "s3_failed": bool(r.get("s3_failed", False)),
                 "logs": [],
             }
             # 重启时 downloading/processing 说明上次处理被中断，标为 error
@@ -529,16 +532,26 @@ def _process(task_id: str, video_path: str, inpaint_mode_str: str, subtitle_area
 
         if not subtitle_areas:
             if detect_mode == "smart":
-                # 水印模式：逐帧 top-hat 检测半透明/移动文字水印
-                push_log(task_id, "水印检测中（逐帧 top-hat）...")
+                # 水印模式：逐帧检测半透明/移动文字水印 + 彩色 logo（每秒一帧采样）
+                push_log(task_id, "水印检测中（逐帧 top-hat + 彩色 logo）...")
                 from backend.tools.smart_detect import detect_watermark_per_frame
-                per_frame = detect_watermark_per_frame(video_path, sample_step=2)
+                per_frame = detect_watermark_per_frame(video_path)
                 if per_frame:
                     n_boxes = sum(len(v) for v in per_frame.values())
                     sr._per_frame_boxes = per_frame
                     push_log(task_id, f"水印检测：{len(per_frame)} 帧共 {n_boxes} 个水印框")
                 else:
-                    push_log(task_id, "未检测到水印，回退到全帧 OCR")
+                    # 未检测到水印：这类淡水印 OCR 也看不到，跑 OCR 纯浪费 → 原样输出
+                    push_log(task_id, "未检测到水印，原样输出原视频")
+                    out_stem = Path(video_path).stem
+                    sr.video_out_path = str(OUTPUT_DIR / f"{out_stem}_{task_id[:8]}_no_sub.mp4")
+                    import shutil
+                    shutil.copy2(video_path, sr.video_out_path)
+                    update_task(task_id, status="done", progress=100,
+                                detect_pct=100, process_pct=100, merge_pct=100,
+                                output_path=sr.video_out_path, completed_at=time.time())
+                    push_log(task_id, f"完成（未检测到水印，输出原视频）→ {sr.video_out_path}")
+                    return
             else:
                 # 文字模式：标准 OCR 逐帧检测（走原生 find_subtitle_frame_no）
                 push_log(task_id, "文字检测（OCR）将逐帧进行")
@@ -754,6 +767,7 @@ class VSRTaskWorker:
             "cached": cached,
             "output_path": None,
             "error": None,
+            "s3_failed": False,
             "platform_task_id": platform_task_id,
             "logs": [],
             "created_at": time.time(),
@@ -802,8 +816,18 @@ class VSRTaskWorker:
             final = tasks.get(task_id, {})
 
         if final.get("status") == "done" and final.get("output_path"):
-            output_url = await self._upload_result_video(final["output_path"])
+            output_url = await self._upload_result_video(final["output_path"], task_id=task_id)
+            if output_url:
+                # 上传 S3 成功，才算真正完成
                 await self._report_result(platform_task_id, status="done", output_url=output_url)
+                push_log(task_id, f"✓ 已上传 S3：{output_url}")
+            else:
+                # 处理成功但上传失败 → 保留 done 状态（本地文件可播放/下载），仅报错给平台
+                logger.error(f"[vsr_worker] 任务 #{platform_task_id} 处理成功但上传 S3 失败")
+                update_task(task_id, s3_failed=True, error="S3 上传失败，视频已在本地，可直接下载")
+                push_log(task_id, "✗ S3 上传失败，但本地文件完好，可点击下载按钮获取")
+                await self._report_result(platform_task_id, status="failed",
+                                          error_message="处理完成但视频上传 S3 失败")
         else:
             await self._report_result(platform_task_id, status="failed",
                                       error_message=final.get("error", "unknown"))
@@ -825,9 +849,13 @@ class VSRTaskWorker:
         )
         logger.info(f"[vsr_worker] 任务 #{platform_task_id} 上报结果 status={status}")
 
-    async def _upload_result_video(self, local_path: str) -> str | None:
+    async def _upload_result_video(self, local_path: str, task_id: str | None = None) -> str | None:
         """处理完成后将输出视频上传到平台，返回 play_url（S3 地址）。流式上传，不一次性读入内存。"""
         import hashlib as _hl
+        def _log(msg: str):
+            logger.info(msg)
+            if task_id:
+                push_log(task_id, msg)
         try:
             file_path = Path(local_path)
             file_size = file_path.stat().st_size
@@ -839,7 +867,7 @@ class VSRTaskWorker:
                 for chunk in iter(lambda: f.read(8 * 1024 * 1024), b""):
                     h.update(chunk)
             sha256 = h.hexdigest()
-            logger.info(f"[vsr_worker] 文件 {filename} size={file_size//1048576}MB sha256={sha256[:12]}…")
+            _log(f"[S3] 准备上传 {filename}，大小={file_size//1048576}MB")
 
             sign_resp = await self._signed_post(
                 "/openapi/file/upload/get_upload_sign",
@@ -854,29 +882,41 @@ class VSRTaskWorker:
                 },
             )
             if sign_resp is None:
-                logger.warning("[vsr_worker] get_upload_sign 返回空，跳过上传")
+                _log("[S3] get_upload_sign 返回空")
                 return None
             action = sign_resp.get("action", "")
             if action == "skip":
-                logger.info(f"[vsr_worker] 文件已存在，直接返回 play_url")
+                _log("[S3] 文件已存在，直接复用")
                 return sign_resp.get("play_url")
             presigned_url = sign_resp.get("presigned_url", "")
             object_key = sign_resp.get("object_key", "")
             if not presigned_url or not object_key:
-                logger.warning(f"[vsr_worker] upload sign 未返回 presigned_url/object_key: {sign_resp}")
+                _log(f"[S3] upload sign 未返回 presigned_url/object_key: {sign_resp}")
                 return None
 
-            # 流式 PUT 上传，不把文件加载到内存
-            logger.info(f"[vsr_worker] 开始流式上传 {file_size//1048576}MB → S3…")
+            # 用 asyncio executor 做同步分块读取，避免 sync file 阻塞 async event loop
+            _log(f"[S3] 开始上传 {file_size//1048576}MB → S3…")
+            CHUNK = 8 * 1024 * 1024  # 8MB chunks
+
+            async def _file_chunks():
+                loop = asyncio.get_running_loop()
+                with open(local_path, "rb") as fobj:
+                    while True:
+                        chunk = await loop.run_in_executor(None, fobj.read, CHUNK)
+                        if not chunk:
+                            break
+                        yield chunk
+
             async with httpx.AsyncClient(timeout=7200) as upload_client:
-                with open(local_path, "rb") as f:
                 put_resp = await upload_client.put(
                     presigned_url,
-                        content=f,
+                    content=_file_chunks(),
                     headers={"Content-Type": "video/mp4", "Content-Length": str(file_size)},
                 )
-            put_resp.raise_for_status()
-            logger.info(f"[vsr_worker] S3 上传完成，注册资产…")
+            if put_resp.status_code not in (200, 204):
+                _log(f"[S3] PUT 失败 HTTP {put_resp.status_code}: {put_resp.text[:200]}")
+                return None
+            _log("[S3] 上传完成，注册资产…")
 
             reg_resp = await self._signed_post(
                 "/openapi/file/assets/register",
@@ -892,9 +932,11 @@ class VSRTaskWorker:
                 },
             )
             play_url = reg_resp.get("play_url") if reg_resp else None
-            logger.info(f"[vsr_worker] 资产注册完成 play_url={play_url}")
+            _log(f"[S3] 资产注册完成 play_url={play_url}")
             return play_url
         except Exception as e:
+            import traceback as _tb
+            _log(f"[S3] 上传失败: {e}\n{_tb.format_exc()[-400:]}")
             logger.warning(f"[vsr_worker] 上传结果视频失败: {e}")
             return None
 
@@ -1078,7 +1120,7 @@ async def create_task(body: TaskCreate):
             "status": "pending", "progress": 0,
             "phase": "pending", "detect_pct": 0, "process_pct": 0, "merge_pct": 0,
             "download_progress": 0, "cached": cached,
-            "output_path": None, "error": None, "platform_task_id": None,
+            "output_path": None, "error": None, "s3_failed": False, "platform_task_id": None,
             "logs": [], "created_at": now, "started_at": None, "completed_at": None,
         }
     task_queue.put(task_id)
@@ -1127,7 +1169,12 @@ async def get_task_logs(task_id: str):
 async def download_result(task_id: str):
     with task_lock:
         task = tasks.get(task_id)
-    if not task or task["status"] != "done" or not task.get("output_path"):
+    if not task or not task.get("output_path"):
+        raise HTTPException(404, "Result not ready")
+    # 允许 done 或 S3 上传失败（本地文件完好）的任务下载
+    if task["status"] not in ("done", "error") or (
+        task["status"] == "error" and not task.get("s3_failed")
+    ):
         raise HTTPException(404, "Result not ready")
     out = Path(task["output_path"])
     if not out.exists():
@@ -1139,7 +1186,11 @@ async def download_result(task_id: str):
 async def stream_result(task_id: str, request: Request):
     with task_lock:
         task = tasks.get(task_id)
-    if not task or task["status"] != "done" or not task.get("output_path"):
+    if not task or not task.get("output_path"):
+        raise HTTPException(404, "Result not ready")
+    if task["status"] not in ("done", "error") or (
+        task["status"] == "error" and not task.get("s3_failed")
+    ):
         raise HTTPException(404, "Result not ready")
     out = Path(task["output_path"])
     if not out.exists():