From b2dfbce962358a241d6cd878f954896c1c4e66bc Mon Sep 17 00:00:00 2001 From: liuchen Date: Sat, 27 Jun 2026 14:03:50 +0800 Subject: [PATCH] Refactor VSR web: align UI with df-autotest, add auth and platform worker - Redesign web/index.html to match df-autotest light theme (sidebar layout, same token system, badges, panels, modals, toast notifications) - Add simple cookie-based auth (ADMIN_PASSWORD env via .env) - Add platform Worker (poll external platform for VSR tasks, sign requests with HMAC same as df-autotest) - Add Settings page for platform connection config (DB-stored, no env vars) - Add worker log API with incremental polling - Remove bottom console strip; add per-task log modal with SQLite persistence - Task table: add created_at / started_at columns, move progress to detail modal - Fix modal close buttons (direct onclick, remove unreliable data-close delegation) - Load .env via python-dotenv on startup Co-Authored-By: Claude Haiku 4.5 --- .env.example | 13 + .gitignore | 1 + web/index.html | 2054 +++++++++++++++++++++++++++--------------------- web_app.py | 815 +++++++++++++------ 4 files changed, 1769 insertions(+), 1114 deletions(-) create mode 100644 .env.example diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..1203e51 --- /dev/null +++ b/.env.example @@ -0,0 +1,13 @@ +# VSR Web 配置示例 — 复制为 .env 后修改 +# 参考 df-autotest-platform 的 .env 结构 + +# 管理员密码（留空则不需要登录） +ADMIN_PASSWORD= + +# 存储目录（默认 ./web_cache） +# VSR_STORAGE=./web_cache + +# 服务监听端口（默认 8001） +# PORT=8001 + +# Platform Worker 连接信息在「系统设置」页面配置，无需在此设置 diff --git a/.gitignore b/.gitignore index b05a3d3..41ace06 100644 --- a/.gitignore +++ b/.gitignore @@ -376,3 +376,4 @@ test*_no_sub*.mp4 inference.onnx /config/config.json web_cache/ +.env diff --git a/web/index.html b/web/index.html index da7a842..7238139 100644 --- a/web/index.html +++ b/web/index.html @@ -1,740 +1,926 @@ - - + + - - -视频字幕去除器 + + +视频字幕去除器 · VSR - -
- -

视频字幕去除器

- -
v1.4.0
-
- -
- - -
- - - -
-

新建任务

- - - -
- - - ✓ 已缓存 - +
+ +
+ -
+ +
+ +
+
+

任务队列

+

提交视频链接，去除字幕或水印，实时查看处理进度。

+
+
+ + +
+
- -
-

任务队列

-
-
- 📋 -

没有任务。输入链接后点击开始处理。

+ +
+
任务总数0
+
处理中0
+
已完成0
+
失败0
+
+ + + + + +
+
+
+

任务队列

+ 0 个任务 +
+
+ + + + + + + + + + + + + +
视频状态算法来源创建时间开始处理
+
-
- -
-
- 处理日志 -
- - - + +
+
+
+

平台 Worker 配置

+
未运行
+
+
+
+
+ + +
+
+ + +
+
+ + +
+
+ + + 0 = 关闭平台拉取，仅保留手动提交 +
+
+ + +
+
+
+ +
+
+
+ +
+
+

Worker 运行日志

+ +
+
+
+
尚无日志。Worker 运行后日志会出现在这里。
+
+
-
-
- +
+ + + + + + + + + + + + + +
+ diff --git a/web_app.py b/web_app.py index 1dc686e..20e40d8 100644 --- a/web_app.py +++ b/web_app.py @@ -2,19 +2,23 @@ """ Web interface for video-subtitle-remover. -Usage (from project root): +Usage: + cp .env.example .env # 编辑 .env 设置 ADMIN_PASSWORD python web_app.py - # then open http://localhost:8000 -Or from any directory: - VSR_PATH=/path/to/video-subtitle-remover python web_app.py +Platform worker: 在「系统设置」页面配置，无需环境变量。 """ +from __future__ import annotations + import asyncio +import collections import cv2 import hashlib -import io +import hmac +import itertools import json +import logging import os import queue as q_module import sqlite3 @@ -24,15 +28,21 @@ import time import traceback import uuid from pathlib import Path -from typing import Optional +from typing import Any, Optional + +# 加载 .env（同目录） +try: + from dotenv import load_dotenv + load_dotenv(Path(__file__).parent / ".env") +except ImportError: + pass import httpx from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException, Request -from fastapi.responses import FileResponse, HTMLResponse +from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse, Response from pydantic import BaseModel -# ── project path ──────────────────────────────────────────────────────────── -# Default: the directory containing this file (run from project root) +# ── project path ───────────────────────────────────────────────────────────── VSR_PATH = os.environ.get("VSR_PATH", str(Path(__file__).parent)) if VSR_PATH not in sys.path: sys.path.insert(0, VSR_PATH) @@ -42,7 +52,6 @@ try: from backend.tools.constant import InpaintMode from backend.main import SubtitleRemover import configparser as _cp - # Force Chinese interface for web app _ch_ini = os.path.join(VSR_PATH, 'backend', 'interface', 'ch.ini') tr.read(_ch_ini, encoding='utf-8') VSR_AVAILABLE = True @@ -51,7 +60,7 @@ except ImportError as _import_err: VSR_AVAILABLE = False _import_err_msg = str(_import_err) -# ── storage dirs ──────────────────────────────────────────────────────────── +# ── paths ───────────────────────────────────────────────────────────────────── BASE_DIR = Path(__file__).parent CACHE_DIR = BASE_DIR / "web_cache" / "downloads" OUTPUT_DIR = BASE_DIR / "web_cache" / "outputs" @@ -59,9 +68,88 @@ DB_PATH = BASE_DIR / "web_cache" / "tasks.db" CACHE_DIR.mkdir(parents=True, exist_ok=True) OUTPUT_DIR.mkdir(parents=True, exist_ok=True) -# ── database ───────────────────────────────────────────────────────────── -def init_db(): +# ── auth ────────────────────────────────────────────────────────────────────── + +_COOKIE = "vsr_session" +_COOKIE_MAX_AGE = 86400 * 30 # 30 天 + +_LOGIN_HTML = """ + + + + +登录 · 视频字幕去除器 + + + +
+

视频字幕去除器

+
+ + + {error} + +
+
+ +""" + + +def _admin_password() -> str: + return (os.environ.get("ADMIN_PASSWORD") or "").strip() + + +def _make_token() -> str: + pw = _admin_password() + ts = str(int(time.time())) + sig = hmac.new(pw.encode(), ts.encode(), hashlib.sha256).hexdigest() + return f"{ts}.{sig}" + + +def _verify_token(token: str) -> bool: + pw = _admin_password() + if not pw: + return True + try: + ts, sig = token.split(".", 1) + expected = hmac.new(pw.encode(), ts.encode(), hashlib.sha256).hexdigest() + return hmac.compare_digest(sig, expected) and int(ts) > time.time() - _COOKIE_MAX_AGE + except Exception: + return False + + +def _auth_ok(request: Request) -> bool: + if not _admin_password(): + return True + return _verify_token(request.cookies.get(_COOKIE, "")) + + +# ── database ────────────────────────────────────────────────────────────────── + +def _db() -> sqlite3.Connection: conn = sqlite3.connect(str(DB_PATH)) + conn.row_factory = sqlite3.Row + return conn + + +def init_db(): + conn = _db() conn.execute(""" CREATE TABLE IF NOT EXISTS tasks ( id TEXT PRIMARY KEY, @@ -78,24 +166,58 @@ def init_db(): cached BOOLEAN, output_path TEXT, error TEXT, + platform_task_id INTEGER, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) - # Try to add new columns if they don't exist - try: - conn.execute("ALTER TABLE tasks ADD COLUMN preview_mode BOOLEAN DEFAULT 0") - except: - pass - try: - conn.execute("ALTER TABLE tasks ADD COLUMN detect_mode TEXT DEFAULT 'text'") - except: - pass + conn.execute(""" + CREATE TABLE IF NOT EXISTS settings ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL + ) + """) + conn.execute(""" + CREATE TABLE IF NOT EXISTS task_logs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_id TEXT NOT NULL, + message TEXT NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """) + conn.execute("CREATE INDEX IF NOT EXISTS idx_task_logs_task_id ON task_logs(task_id)") + for col in ("preview_mode", "detect_mode", "platform_task_id"): + try: + conn.execute(f"ALTER TABLE tasks ADD COLUMN {col} {'BOOLEAN DEFAULT 0' if col == 'preview_mode' else ('TEXT DEFAULT ' + repr('text') if col == 'detect_mode' else 'INTEGER')}") + except Exception: + pass conn.commit() conn.close() + init_db() +WORKER_KEYS = ("platform_base_url", "platform_app_id", "platform_app_secret", + "worker_concurrency", "worker_poll_interval") + + +def _settings_get_all() -> dict[str, str]: + conn = _db() + rows = conn.execute("SELECT key, value FROM settings").fetchall() + conn.close() + return {r["key"]: r["value"] for r in rows} + + +def _settings_set_many(data: dict[str, str]) -> None: + conn = _db() + for k, v in data.items(): + conn.execute("INSERT OR REPLACE INTO settings (key, value) VALUES (?, ?)", (k, v)) + conn.commit() + conn.close() + + +# ── inpaint mode map ────────────────────────────────────────────────────────── + INPAINT_MODE_MAP = { "sttn-auto": lambda: InpaintMode.STTN_AUTO, "sttn-det": lambda: InpaintMode.STTN_DET, @@ -104,14 +226,14 @@ INPAINT_MODE_MAP = { "opencv": lambda: InpaintMode.OPENCV, } -# ── state ──────────────────────────────────────────────────────────────────── +# ── task state ──────────────────────────────────────────────────────────────── + tasks: dict[str, dict] = {} task_lock = threading.Lock() task_queue: q_module.Queue = q_module.Queue() ws_connections: list[WebSocket] = [] _loop: Optional[asyncio.AbstractEventLoop] = None -# ── helpers ────────────────────────────────────────────────────────────────── def url_to_cache_path(url: str) -> Path: h = hashlib.md5(url.encode()).hexdigest()[:12] @@ -151,69 +273,54 @@ def update_task(task_id: str, **kwargs): _save_task_to_db(tasks[task_id]) _push(_broadcast({"type": "task_update", "task": snap})) + def _save_task_to_db(task: dict): - """Save task to database""" try: - conn = sqlite3.connect(str(DB_PATH)) + conn = _db() conn.execute(""" INSERT OR REPLACE INTO tasks (id, url, is_cache_file, cache_path, inpaint_mode, subtitle_areas, - preview_mode, detect_mode, status, progress, download_progress, cached, output_path, error) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + preview_mode, detect_mode, status, progress, download_progress, + cached, output_path, error, platform_task_id) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( - task["id"], - task["url"], - task.get("is_cache_file", False), - task.get("cache_path", ""), - task["inpaint_mode"], - json.dumps(task["subtitle_areas"]), - task.get("preview_mode", False), - task.get("detect_mode", "text"), - task["status"], - task["progress"], - task["download_progress"], - task["cached"], - task["output_path"], - task["error"], + task["id"], task["url"], task.get("is_cache_file", False), + task.get("cache_path", ""), task["inpaint_mode"], + json.dumps(task["subtitle_areas"]), task.get("preview_mode", False), + task.get("detect_mode", "text"), task["status"], task["progress"], + task["download_progress"], task["cached"], task["output_path"], + task["error"], task.get("platform_task_id"), )) conn.commit() conn.close() except Exception as e: print(f"Failed to save task to DB: {e}") + def _load_tasks_from_db(): - """Load tasks from database on startup""" try: - conn = sqlite3.connect(str(DB_PATH)) - conn.row_factory = sqlite3.Row # Use column names - cursor = conn.execute( - "SELECT * FROM tasks ORDER BY created_at DESC LIMIT 100" - ) + conn = _db() + cursor = conn.execute("SELECT * FROM tasks ORDER BY created_at DESC LIMIT 100") + keys = [d[0] for d in cursor.description] for row in cursor.fetchall(): + r = dict(zip(keys, row)) task = { - "id": row["id"], - "url": row["url"], - "is_cache_file": row["is_cache_file"], - "cache_path": row["cache_path"], - "inpaint_mode": row["inpaint_mode"], - "subtitle_areas": json.loads(row["subtitle_areas"]), - "preview_mode": row["preview_mode"] if "preview_mode" in row.keys() else False, - "detect_mode": row["detect_mode"] if "detect_mode" in row.keys() else "text", - "status": row["status"], - "progress": row["progress"], - "download_progress": row["download_progress"], - "cached": row["cached"], - "output_path": row["output_path"], - "error": row["error"], - "phase": "pending", - "detect_pct": 0, - "process_pct": 0, - "merge_pct": 0, + "id": r["id"], "url": r["url"], + "is_cache_file": r.get("is_cache_file", False), + "cache_path": r.get("cache_path", ""), + "inpaint_mode": r["inpaint_mode"], + "subtitle_areas": json.loads(r["subtitle_areas"] or "[]"), + "preview_mode": r.get("preview_mode", False), + "detect_mode": r.get("detect_mode", "text"), + "status": r["status"], "progress": r["progress"], + "phase": "pending", "detect_pct": 0, "process_pct": 0, "merge_pct": 0, + "download_progress": r["download_progress"], + "cached": r["cached"], "output_path": r["output_path"], + "error": r["error"], "platform_task_id": r.get("platform_task_id"), "logs": [], } with task_lock: tasks[task["id"]] = task - # Re-queue pending/downloading/processing tasks (processing might have crashed) if task["status"] in ("pending", "downloading", "processing"): task_queue.put(task["id"]) conn.close() @@ -225,10 +332,18 @@ def push_log(task_id: str, message: str): with task_lock: if task_id in tasks: tasks[task_id]["logs"].append(message) + # 持久化到 DB + try: + conn = _db() + conn.execute("INSERT INTO task_logs (task_id, message) VALUES (?, ?)", (task_id, message)) + conn.commit() + conn.close() + except Exception: + pass _push(_broadcast({"type": "log", "task_id": task_id, "message": message})) -# ── worker ─────────────────────────────────────────────────────────────────── +# ── local worker ────────────────────────────────────────────────────────────── def _download(task_id: str, url: str, dest: Path) -> bool: update_task(task_id, status="downloading", download_progress=0) @@ -247,10 +362,8 @@ def _download(task_id: str, url: str, dest: Path) -> bool: pct = int(done / total * 100) if pct >= last_pct + 2: last_pct = pct - mb_done = done // 1048576 - mb_total = total // 1048576 update_task(task_id, download_progress=pct) - push_log(task_id, f"下载 {pct}%（{mb_done} MB / {mb_total} MB）") + push_log(task_id, f"下载 {pct}%（{done//1048576} MB / {total//1048576} MB）") push_log(task_id, f"下载完成 → {dest.name}") return True except Exception as e: @@ -259,18 +372,20 @@ def _download(task_id: str, url: str, dest: Path) -> bool: return False -def _process(task_id: str, video_path: str, inpaint_mode_str: str, subtitle_areas: list, preview_mode: bool = False, detect_mode: str = "text"): +def _process(task_id: str, video_path: str, inpaint_mode_str: str, subtitle_areas: list, + preview_mode: bool = False, detect_mode: str = "text"): if not VSR_AVAILABLE: push_log(task_id, f"错误：无法加载后端模块。{_import_err_msg}") update_task(task_id, status="error", error="VSR backend not available") return - modeLabels = {'sttn-auto':'STTN 自动','sttn-det':'STTN 检测','lama':'LaMa','propainter':'ProPainter','opencv':'OpenCV'} + modeLabels = {'sttn-auto': 'STTN 自动', 'sttn-det': 'STTN 检测', + 'lama': 'LaMa', 'propainter': 'ProPainter', 'opencv': 'OpenCV'} push_log(task_id, f"启动 — 算法：{modeLabels.get(inpaint_mode_str, inpaint_mode_str)}") update_task(task_id, status="processing", progress=0, started_at=time.time()) class _WebRemover(SubtitleRemover): - _static_boxes = [] # SAM2检测到的静态区域，格式 (xmin, xmax, ymin, ymax) 像素 + _static_boxes = [] def append_output(self_, *args): msg = " ".join(str(a) for a in args) @@ -278,18 +393,16 @@ def _process(task_id: str, video_path: str, inpaint_mode_str: str, subtitle_area push_log(task_id, msg) def update_preview_with_comp(self_, frame_ori, frame_comp): - pass # no-op: no GUI + pass def run(self_): if self_._static_boxes: from backend.tools.subtitle_detect import SubtitleDetect as _SD _orig = _SD.find_subtitle_frame_no _boxes = list(self_._static_boxes) + def _patched(sd_self, sub_remover=None): result = _orig(sd_self, sub_remover=sub_remover) - ocr_frames = len(result) - if ocr_frames > 0: - push_log(task_id, f"OCR 在 {ocr_frames} 帧中检测到文字") cap = cv2.VideoCapture(sd_self.video_path) total = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) cap.release() @@ -314,41 +427,31 @@ def _process(task_id: str, video_path: str, inpaint_mode_str: str, subtitle_area update_task(task_id, detect_pct=100, process_pct=100, merge_pct=100, phase="merging") push_log(task_id, "音频合并完成") - # Preview mode: cut first 10 seconds using FFmpeg BEFORE processing preview_clip_path = None if preview_mode: try: from backend.tools.ffmpeg_cli import FFmpegCLI - preview_clip_path = str(Path(video_path).parent / f"_preview_{task_id}.mp4") - cut_cmd = [ - FFmpegCLI.instance().ffmpeg_path, - "-y", "-i", video_path, - "-t", "10", - "-c", "copy", - "-loglevel", "error", - preview_clip_path - ] - push_log(task_id, "预览模式：截取前10秒...") import subprocess + preview_clip_path = str(Path(video_path).parent / f"_preview_{task_id}.mp4") + cut_cmd = [FFmpegCLI.instance().ffmpeg_path, "-y", "-i", video_path, + "-t", "10", "-c", "copy", "-loglevel", "error", preview_clip_path] + push_log(task_id, "预览模式：截取前10秒...") subprocess.run(cut_cmd, check=True, timeout=60) push_log(task_id, "截取完成，开始处理预览片段") - video_path = preview_clip_path # process the short clip + video_path = preview_clip_path except Exception as e: push_log(task_id, f"截取失败，改为处理完整视频: {e}") preview_clip_path = None try: sr = _WebRemover(video_path, gui_mode=False) - if preview_mode: push_log(task_id, f"预览片段：{sr.frame_count} 帧 / {sr.fps:.1f}fps") mode_factory = INPAINT_MODE_MAP.get(inpaint_mode_str, INPAINT_MODE_MAP["sttn-auto"]) config.set(config.inpaintMode, mode_factory()) - # Auto-detect regions if not provided if not subtitle_areas: - combined_areas = [] import concurrent.futures def _run_ocr(): @@ -357,92 +460,47 @@ def _process(task_id: str, video_path: str, inpaint_mode_str: str, subtitle_area cap = cv2.VideoCapture(video_path) ret, frame = cap.read() cap.release() - if ret: - return detector.detect_subtitle(frame) - return [] + return detector.detect_subtitle(frame) if ret else [] def _run_sam2(): - print(f"[DEBUG] _run_sam2 called, detect_mode={detect_mode}", flush=True) if detect_mode != "smart": - print(f"[DEBUG] detect_mode != smart, skipping SAM2", flush=True) return [] from backend.tools.smart_detect import smart_detect_watermark return smart_detect_watermark(video_path, sample_frames=30) push_log(task_id, "并发检测中（OCR + SAM2）..." if detect_mode == "smart" else "文字检测（OCR）中...") with concurrent.futures.ThreadPoolExecutor(max_workers=2) as ex: - f_ocr = ex.submit(_run_ocr) + f_ocr = ex.submit(_run_ocr) f_sam2 = ex.submit(_run_sam2) - ocr_result = f_ocr.result() + ocr_result = f_ocr.result() sam2_result = f_sam2.result() if ocr_result: push_log(task_id, f"OCR 检测到 {len(ocr_result)} 个文字区域") - combined_areas += [ - ( - int(y1 / sr.frame_height * 100), - int(y2 / sr.frame_height * 100), - int(x1 / sr.frame_width * 100), - int(x2 / sr.frame_width * 100), - ) - for x1, x2, y1, y2 in ocr_result - ] - else: - push_log(task_id, "OCR 未检测到文字区域") - if sam2_result: - push_log(task_id, f"SAM2 检测到 {len(sam2_result)} 个区域，将强制每帧都处理") - for i, (x1, y1, x2, y2) in enumerate(sam2_result): - push_log(task_id, f" 区域{i+1}: 左上({x1},{y1}) 右下({x2},{y2}) 大小{x2-x1}x{y2-y1}px") + push_log(task_id, f"SAM2 检测到 {len(sam2_result)} 个区域") sr._static_boxes = [(x1, x2, y1, y2) for x1, y1, x2, y2 in sam2_result] elif detect_mode == "smart": push_log(task_id, "SAM2 未检测到静态水印区域") - # 单帧初探仅用于日志，不作为全扫描的空间约束 - # 若把 21 个 15px 小框设为 sub_areas，全扫描几乎找不到任何匹配帧 → 报错 - # SubtitleRemover.find_subtitle_frame_no 会自行全帧扫描 - if not combined_areas and not sr._static_boxes: - push_log(task_id, "初探未检测到区域，全扫描将覆盖全屏") - elif combined_areas: - push_log(task_id, f"初探检测到 {len(combined_areas)} 个文字区域（全扫描不受此约束）") - if sr._static_boxes: - push_log(task_id, f"SAM2 静态区域已注入全部帧，无需区域约束") - # subtitle_areas 保持为空，sub_areas 不设置 → 全帧无约束扫描 - if subtitle_areas: - pixel_areas = [ - ( - int(a[0] / 100 * sr.frame_height), - int(a[1] / 100 * sr.frame_height), - int(a[2] / 100 * sr.frame_width), - int(a[3] / 100 * sr.frame_width), - ) - for a in subtitle_areas - ] if detect_mode == "smart": - # 水印模式：用户手动指定区域 + 智能检测 → 强制每帧处理该区域 - # 适合位置变化的动态水印（OCR 无法识别装饰性字体） - force_boxes = [ + sr._static_boxes = (sr._static_boxes or []) + [ (int(a[2]/100*sr.frame_width), int(a[3]/100*sr.frame_width), int(a[0]/100*sr.frame_height), int(a[1]/100*sr.frame_height)) for a in subtitle_areas ] - if sr._static_boxes: - sr._static_boxes = sr._static_boxes + force_boxes - else: - sr._static_boxes = force_boxes - push_log(task_id, f"手动指定 {len(force_boxes)} 个水印区域 → 强制每帧去除") - print(f"[DEBUG] force_boxes={force_boxes}", flush=True) + push_log(task_id, f"手动指定 {len(subtitle_areas)} 个水印区域 → 强制每帧去除") else: - # 字幕模式：限制 OCR 搜索范围 - sr.sub_areas = pixel_areas + sr.sub_areas = [ + (int(a[0]/100*sr.frame_height), int(a[1]/100*sr.frame_height), + int(a[2]/100*sr.frame_width), int(a[3]/100*sr.frame_width)) + for a in subtitle_areas + ] out_stem = Path(video_path).stem sr.video_out_path = str(OUTPUT_DIR / f"{out_stem}_{task_id[:8]}_no_sub.mp4") - # 检测阶段：progress_total = (100*frame/total)//2 → 0-50 - # 处理阶段：update_progress → 0-100（从0重新开始） - # 判断切换：raw>50 OR 前值≥40后值<15（跌落 = 阶段重置） _state = {"phase": "detecting", "processing": False, "prev_raw": 0} def _on_progress(pct, finished): @@ -450,41 +508,30 @@ def _process(task_id: str, video_path: str, inpaint_mode_str: str, subtitle_area raw = int(pct) prev = _state["prev_raw"] _state["prev_raw"] = raw - # 跌落检测：从检测末尾（≥40）掉回处理起始（<15） if not _state["processing"] and prev >= 40 and raw < 15: _state["processing"] = True if _state["processing"] or raw > 50: _state["processing"] = True - _state["phase"] = "processing" - process_pct = min(raw, 98) - update_task(task_id, detect_pct=100, process_pct=process_pct, merge_pct=0, phase="processing") + update_task(task_id, detect_pct=100, process_pct=min(raw, 98), merge_pct=0, phase="processing") else: - detect_pct = min(raw * 2, 99) # 0-50 → 0-99 - update_task(task_id, detect_pct=detect_pct, process_pct=0, merge_pct=0, phase="detecting") - print(f"[PROGRESS] {task_id}: phase={_state['phase']} raw={raw}%", flush=True) + update_task(task_id, detect_pct=min(raw * 2, 99), process_pct=0, merge_pct=0, phase="detecting") except Exception as e: print(f"[ERROR] Progress callback failed: {e}", flush=True) push_log(task_id, f"检测模式：{'智能（SAM2）' if detect_mode == 'smart' else '文字（OCR）'}") - - print(f"[DEBUG] Adding progress listener for {task_id}", flush=True) sr.add_progress_listener(_on_progress) - print(f"[DEBUG] Progress listeners count: {len(sr.progress_listeners)}", flush=True) - print(f"[DEBUG] Starting sr.run()", flush=True) sr.run() - print(f"[DEBUG] sr.run() completed", flush=True) - # Preview mode: skip audio merge, directly use temp file if preview_mode and hasattr(sr, 'video_temp_file') and sr.video_temp_file: - push_log(task_id, "预览模式：跳过音频合并，直接输出") import shutil + push_log(task_id, "预览模式：跳过音频合并，直接输出") try: shutil.copy2(sr.video_temp_file.name, sr.video_out_path) - push_log(task_id, f"预览完成 → {sr.video_out_path}") except Exception as e: push_log(task_id, f"复制文件失败: {e}") - update_task(task_id, status="done", progress=100, output_path=sr.video_out_path, completed_at=time.time()) + update_task(task_id, status="done", progress=100, + output_path=sr.video_out_path, completed_at=time.time()) push_log(task_id, f"完成 → {sr.video_out_path}") except Exception: @@ -492,7 +539,6 @@ def _process(task_id: str, video_path: str, inpaint_mode_str: str, subtitle_area push_log(task_id, f"错误：\n{err}") update_task(task_id, status="error", error=err.splitlines()[-1]) finally: - # Clean up preview clip if preview_clip_path and Path(preview_clip_path).exists(): try: Path(preview_clip_path).unlink() @@ -506,26 +552,21 @@ def _run_task(task_id: str): if not task: return - # Handle cache file directly if task.get("is_cache_file"): cache_path = Path(task["cache_path"]) - if cache_path.exists(): - push_log(task_id, f"使用本地缓存: {cache_path.name}") - update_task(task_id, cached=True) - else: + if not cache_path.exists(): update_task(task_id, status="error", error="缓存文件不存在") return + push_log(task_id, f"使用本地缓存: {cache_path.name}") + update_task(task_id, cached=True) else: - # Handle URL download url = task["url"] cache_path = url_to_cache_path(url) - if cache_path.exists(): push_log(task_id, f"缓存命中: {cache_path.name}") update_task(task_id, cached=True) else: - ok = _download(task_id, url, cache_path) - if not ok: + if not _download(task_id, url, cache_path): update_task(task_id, status="error", error="下载失败") return @@ -546,20 +587,298 @@ def _worker(): task_queue.task_done() -# ── FastAPI ─────────────────────────────────────────────────────────────────── +# ── platform worker ─────────────────────────────────────────────────────────── + +logger = logging.getLogger(__name__) + + +class WorkerLogBuffer: + _seq = itertools.count(1) + + def __init__(self, maxlen: int = 1000) -> None: + self._buf: collections.deque[dict] = collections.deque(maxlen=maxlen) + self._lock = threading.Lock() + + def push(self, level: str, msg: str) -> None: + with self._lock: + self._buf.append({"seq": next(self._seq), "ts": time.time(), "level": level, "msg": msg}) + + def since(self, seq: int) -> tuple[list[dict], int]: + with self._lock: + lines = [e for e in self._buf if e["seq"] > seq] + return lines, (lines[-1]["seq"] if lines else seq) + + +class VSRTaskWorker: + """轮询外部平台认领 VSR 任务，执行后上报结果。""" + + def __init__(self, *, base_url: str, app_id: str, app_secret: str, + slot_id: int, slot_total: int, poll_interval: float = 5.0) -> None: + self._base_url = base_url.rstrip("/") + self._app_id = app_id + self._app_secret = app_secret + self._slot_id = slot_id + self._slot_total = slot_total + self._poll_interval = poll_interval + self._active: set[asyncio.Task] = set() + + async def run(self) -> None: + logger.info(f"[vsr_worker] 启动 slot={self._slot_id} → {self._base_url}") + async with httpx.AsyncClient(base_url=self._base_url, timeout=30.0) as client: + self._client = client + while True: + try: + await self._tick() + except Exception: + logger.exception("[vsr_worker] tick 异常") + await asyncio.sleep(self._poll_interval) + + async def _tick(self) -> None: + if self._active: + return + task = await self._claim() + if task is None: + return + t = asyncio.create_task(self._run_task(task)) + self._active.add(t) + t.add_done_callback(self._active.discard) + + async def _claim(self) -> dict[str, Any] | None: + resp = await self._signed_post( + "/openapi/vsr-task/claim", + {"slot_id": self._slot_id, "slot_total": self._slot_total}, + ) + if resp is None: + return None + task = resp.get("task") + if task is None: + logger.debug(f"[vsr_worker] slot={self._slot_id} → 无待处理任务") + return task + + async def _run_task(self, ptask: dict[str, Any]) -> None: + platform_task_id: int = ptask["id"] + video_url: str = ptask.get("video_url", "") + logger.info(f"[vsr_worker] 认领任务 #{platform_task_id} url={video_url[:60]}") + + task_id = uuid.uuid4().hex[:8] + cache_path = url_to_cache_path(video_url) + cached = cache_path.exists() + + with task_lock: + tasks[task_id] = { + "id": task_id, + "url": video_url, + "is_cache_file": False, + "cache_path": str(cache_path), + "inpaint_mode": ptask.get("inpaint_mode", "sttn-auto"), + "subtitle_areas": ptask.get("subtitle_areas", []), + "preview_mode": ptask.get("preview_mode", False), + "detect_mode": ptask.get("detect_mode", "text"), + "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": platform_task_id, + "logs": [], + "created_at": time.time(), + "started_at": None, + "completed_at": None, + } + + # 在线程中同步执行任务 + def _do(): + _run_task(task_id) + + loop = asyncio.get_event_loop() + + # 进度上报后台 loop + async def _progress_reporter(): + reported_logs = 0 + while True: + await asyncio.sleep(2) + with task_lock: + t = tasks.get(task_id) + if not t: + break + logs = t.get("logs", []) + for msg in logs[reported_logs:]: + await self._report_progress(platform_task_id, msg) + reported_logs = len(logs) + if t["status"] in ("done", "error"): + break + + reporter = asyncio.create_task(_progress_reporter()) + await loop.run_in_executor(None, _do) + reporter.cancel() + + with task_lock: + final = tasks.get(task_id, {}) + + if final.get("status") == "done" and final.get("output_path"): + await self._report_result(platform_task_id, status="passed", + task_id=task_id, output_path=final["output_path"]) + else: + await self._report_result(platform_task_id, status="failed", + task_id=task_id, error=final.get("error", "unknown")) + + async def _report_progress(self, platform_task_id: int, message: str) -> None: + await self._signed_post( + f"/openapi/vsr-task/{platform_task_id}/progress", + {"message": message}, + headers={"X-Worker-Slot": str(self._slot_id)}, + ) + + async def _report_result(self, platform_task_id: int, *, status: str, + task_id: str | None = None, output_path: str | None = None, + error: str = "") -> None: + await self._signed_post( + f"/openapi/vsr-task/{platform_task_id}/result", + {"status": status, "task_id": task_id, "output_path": output_path, "error": error}, + headers={"X-Worker-Slot": str(self._slot_id)}, + ) + logger.info(f"[vsr_worker] 任务 #{platform_task_id} 上报结果 status={status}") + + async def _signed_post(self, path: str, body: dict[str, Any], + headers: dict[str, str] | None = None) -> dict[str, Any] | None: + body_str = json.dumps(body, separators=(",", ":"), ensure_ascii=False) + timestamp = str(int(time.time())) + nonce = uuid.uuid4().hex + payload_str = "\n".join([self._app_id, timestamp, nonce, body_str]) + signature = hmac.new(self._app_secret.encode(), payload_str.encode(), hashlib.sha256).hexdigest() + req_headers = { + "Content-Type": "application/json", + "X-App-Id": self._app_id, + "X-Timestamp": timestamp, + "X-Nonce": nonce, + "X-Signature": signature, + } + if headers: + req_headers.update(headers) + try: + resp = await self._client.post(path, content=body_str, headers=req_headers) + resp.raise_for_status() + return resp.json() + except httpx.HTTPStatusError as exc: + logger.warning(f"[vsr_worker] POST {path} → {exc.response.status_code}: {exc.response.text[:200]}") + except Exception: + logger.exception(f"[vsr_worker] POST {path} 失败") + return None + + +class WorkerManager: + def __init__(self) -> None: + self._tasks: list[asyncio.Task] = [] + self.log_buffer = WorkerLogBuffer() + _handler = logging.StreamHandler() + _handler.setFormatter(logging.Formatter("%(asctime)s [vsr_worker] %(message)s", datefmt="%H:%M:%S")) + _wl = logging.getLogger(__name__) + _wl.setLevel(logging.DEBUG) + _wl.addHandler(_handler) + + def start(self, *, base_url: str, app_id: str, app_secret: str, + concurrency: int, poll_interval: float) -> None: + concurrency = max(1, concurrency) + for slot_id in range(concurrency): + worker = VSRTaskWorker( + base_url=base_url, app_id=app_id, app_secret=app_secret, + slot_id=slot_id, slot_total=concurrency, poll_interval=poll_interval, + ) + self._tasks.append(asyncio.create_task(worker.run())) + logger.info(f"[WorkerManager] 启动 {concurrency} 个 worker slot") + self.log_buffer.push("info", f"启动 {concurrency} 个 worker slot → {base_url}") + + def stop(self) -> None: + for t in self._tasks: + t.cancel() + self._tasks.clear() + + def restart(self, *, base_url: str, app_id: str, app_secret: str, + concurrency: int, poll_interval: float) -> None: + self.stop() + if base_url and app_id and app_secret and concurrency >= 1: + self.start(base_url=base_url, app_id=app_id, app_secret=app_secret, + concurrency=concurrency, poll_interval=poll_interval) + + @property + def running(self) -> bool: + return bool(self._tasks) + + +# ── FastAPI app ─────────────────────────────────────────────────────────────── app = FastAPI(title="VSR Web") +_worker_manager = WorkerManager() _worker_thread = threading.Thread(target=_worker, daemon=True, name="vsr-worker") _worker_thread.start() +@app.middleware("http") +async def _auth_middleware(request: Request, call_next): + path = request.url.path + skip = {"/login", "/auth/login", "/auth/logout", "/health", "/ws"} + if path in skip or path.startswith("/static"): + return await call_next(request) + if not _auth_ok(request): + if path.startswith("/api/"): + return Response(content='{"detail":"Unauthorized"}', + status_code=401, media_type="application/json") + return RedirectResponse(url="/login", status_code=302) + return await call_next(request) + + @app.on_event("startup") async def _startup(): global _loop _loop = asyncio.get_event_loop() _load_tasks_from_db() + saved = _settings_get_all() + base_url = saved.get("platform_base_url", "").strip() + app_id = saved.get("platform_app_id", "").strip() + app_secret = saved.get("platform_app_secret", "").strip() + concurrency = int(saved.get("worker_concurrency", "1")) + poll_interval = float(saved.get("worker_poll_interval", "5")) + _worker_manager.restart(base_url=base_url, app_id=app_id, app_secret=app_secret, + concurrency=concurrency, poll_interval=poll_interval) + + +@app.on_event("shutdown") +async def _shutdown(): + _worker_manager.stop() + + +# ── auth endpoints ──────────────────────────────────────────────────────────── + +@app.get("/login", include_in_schema=False) +async def login_page(): + return HTMLResponse(_LOGIN_HTML.format(error="")) + + +@app.post("/auth/login", include_in_schema=False) +async def do_login(request: Request): + form = await request.form() + pw = (form.get("password") or "").strip() + if hmac.compare_digest(pw.encode(), _admin_password().encode()): + resp = RedirectResponse(url="/", status_code=302) + resp.set_cookie(_COOKIE, _make_token(), max_age=_COOKIE_MAX_AGE, httponly=True, samesite="lax") + return resp + html = _LOGIN_HTML.format(error='

密码错误，请重试

') + return HTMLResponse(html, status_code=401) + + +@app.get("/auth/logout", include_in_schema=False) +async def do_logout(): + resp = RedirectResponse(url="/login", status_code=302) + resp.delete_cookie(_COOKIE) + return resp + + +# ── main page ───────────────────────────────────────────────────────────────── @app.get("/") async def index(): @@ -567,6 +886,13 @@ async def index(): return HTMLResponse(html) +@app.get("/health") +async def health(): + return {"status": "ok"} + + +# ── task API ────────────────────────────────────────────────────────────────── + class TaskCreate(BaseModel): url: str inpaint_mode: str = "sttn-auto" @@ -578,13 +904,9 @@ class TaskCreate(BaseModel): @app.post("/api/tasks") async def create_task(body: TaskCreate): task_id = uuid.uuid4().hex[:8] - - # Check if it's a cache file name (from the recent list) is_cache_file = body.url.startswith("cache_") if is_cache_file: - # Remove "cache_" prefix - filename = body.url[6:] # strip "cache_" - cache_path = CACHE_DIR / filename + cache_path = CACHE_DIR / body.url[6:] cached = cache_path.exists() else: cache_path = url_to_cache_path(body.url) @@ -593,28 +915,15 @@ async def create_task(body: TaskCreate): with task_lock: now = time.time() tasks[task_id] = { - "id": task_id, - "url": body.url, - "is_cache_file": is_cache_file, - "cache_path": str(cache_path), - "inpaint_mode": body.inpaint_mode, - "subtitle_areas": body.subtitle_areas, - "preview_mode": body.preview_mode, - "detect_mode": body.detect_mode, - "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, - "logs": [], - "created_at": now, - "started_at": None, - "completed_at": None, + "id": task_id, "url": body.url, + "is_cache_file": is_cache_file, "cache_path": str(cache_path), + "inpaint_mode": body.inpaint_mode, "subtitle_areas": body.subtitle_areas, + "preview_mode": body.preview_mode, "detect_mode": body.detect_mode, + "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, + "logs": [], "created_at": now, "started_at": None, "completed_at": None, } task_queue.put(task_id) return {"task_id": task_id} @@ -639,17 +948,25 @@ async def get_task(task_id: str): async def delete_task(task_id: str): with task_lock: tasks.pop(task_id, None) - # Also delete from database - try: - conn = sqlite3.connect(str(DB_PATH)) - conn.execute("DELETE FROM tasks WHERE id = ?", (task_id,)) - conn.commit() - conn.close() - except Exception as e: - print(f"Failed to delete task from DB: {e}") + conn = _db() + conn.execute("DELETE FROM tasks WHERE id = ?", (task_id,)) + conn.execute("DELETE FROM task_logs WHERE task_id = ?", (task_id,)) + conn.commit() + conn.close() return {"ok": True} +@app.get("/api/tasks/{task_id}/logs") +async def get_task_logs(task_id: str): + conn = _db() + rows = conn.execute( + "SELECT message, created_at FROM task_logs WHERE task_id = ? ORDER BY id ASC", + (task_id,) + ).fetchall() + conn.close() + return {"logs": [{"message": r[0], "ts": r[1]} for r in rows]} + + @app.get("/api/tasks/{task_id}/download") async def download_result(task_id: str): with task_lock: @@ -671,42 +988,28 @@ async def stream_result(task_id: str, request: Request): out = Path(task["output_path"]) if not out.exists(): raise HTTPException(404, "Output file missing") - file_size = out.stat().st_size range_header = request.headers.get("range") - if range_header: - # Parse Range header: "bytes=start-end" try: range_value = range_header.replace("bytes=", "") start, end = range_value.split("-") start = int(start) if start else 0 end = int(end) if end else file_size - 1 - if start >= file_size or end >= file_size or start > end: raise HTTPException(416, "Range Not Satisfiable") - - headers = { + return FileResponse(str(out), media_type="video/mp4", status_code=206, headers={ "Content-Range": f"bytes {start}-{end}/{file_size}", "Accept-Ranges": "bytes", "Content-Length": str(end - start + 1), - } - return FileResponse( - str(out), - media_type="video/mp4", - headers=headers, - status_code=206, - ) + }) except (ValueError, IndexError): pass + return FileResponse(str(out), filename=out.name, media_type="video/mp4", + headers={"Accept-Ranges": "bytes"}) - return FileResponse( - str(out), - filename=out.name, - media_type="video/mp4", - headers={"Accept-Ranges": "bytes"}, - ) +# ── cache API ───────────────────────────────────────────────────────────────── @app.get("/api/cache-check") async def cache_check(url: str): @@ -719,23 +1022,63 @@ async def cache_list(): if CACHE_DIR.exists(): for f in sorted(CACHE_DIR.iterdir(), key=lambda x: x.stat().st_mtime, reverse=True): if f.is_file(): - files.append({ - "name": f.name, - "size": f.stat().st_size, - "mtime": f.stat().st_mtime, - }) + files.append({"name": f.name, "size": f.stat().st_size, "mtime": f.stat().st_mtime}) return {"files": files} +# ── settings API ────────────────────────────────────────────────────────────── + +@app.get("/api/settings") +async def get_settings_api(): + saved = _settings_get_all() + return { + "platform_base_url": saved.get("platform_base_url", ""), + "platform_app_id": saved.get("platform_app_id", ""), + "platform_app_secret": saved.get("platform_app_secret", ""), + "worker_concurrency": int(saved.get("worker_concurrency", "1")), + "worker_poll_interval": float(saved.get("worker_poll_interval", "5")), + } + + +@app.put("/api/settings") +async def put_settings(body: dict): + allowed = set(WORKER_KEYS) + to_save = {k: str(v) for k, v in body.items() if k in allowed} + _settings_set_many(to_save) + saved = _settings_get_all() + _worker_manager.restart( + base_url=saved.get("platform_base_url", "").strip(), + app_id=saved.get("platform_app_id", "").strip(), + app_secret=saved.get("platform_app_secret", "").strip(), + concurrency=int(saved.get("worker_concurrency", "1")), + poll_interval=float(saved.get("worker_poll_interval", "5")), + ) + return {"ok": True, "worker_running": _worker_manager.running} + + +# ── worker log API ──────────────────────────────────────────────────────────── + +@app.get("/api/worker/logs") +async def get_worker_logs(since: int = 0): + lines, next_seq = _worker_manager.log_buffer.since(since) + return {"lines": lines, "next_seq": next_seq} + + +# ── status API ──────────────────────────────────────────────────────────────── + @app.get("/api/status") async def server_status(): return { "vsr_available": VSR_AVAILABLE, "vsr_path": VSR_PATH, "queue_size": task_queue.qsize(), + "worker_running": _worker_manager.running, + "auth_enabled": bool(_admin_password()), } +# ── websocket ────────────────────────────────────────────────────────────────── + @app.websocket("/ws") async def ws_endpoint(ws: WebSocket): await ws.accept() @@ -763,6 +1106,6 @@ if __name__ == "__main__": print(f"VSR backend available: {VSR_AVAILABLE}") if not VSR_AVAILABLE: print(f" Import error: {_import_err_msg}") - print(f" Set VSR_PATH to the project root and ensure dependencies are installed.") + print(f"Auth enabled: {bool(_admin_password())}") print(f"Starting on http://{host}:{port}") uvicorn.run(app, host=host, port=port)