1

server/app/api/v1/admin.py

@@ -1143,6 +1143,9 @@ def _source_dict(s) -> dict:
         "has_token": bool(cfg.get("token")),
         "last_sync_at": s.last_sync_at.strftime("%Y-%m-%d %H:%M:%S") if s.last_sync_at else "",
         "created_at": s.created_at.strftime("%Y-%m-%d %H:%M:%S"),
+        # 同步运行状态以列表接口合并的 Redis 进度为准（多 worker 下进程内存不可靠）
+        "syncing": False,
+        "sync_progress": None,
     }
 
 
@@ -1161,14 +1164,32 @@ async def _resolve_category(db: AsyncSession, name: str) -> int:
     return c.id
 
 
+# 同步进度 Redis key（多 worker 共享，进程内存不可靠）
+_SYNC_PROGRESS_KEY = "video:admin:sync_progress:{}"
+
+
 @router.get("/collection/sources")
 async def list_sources(db: AsyncSession = Depends(get_db),
+                       redis: aioredis.Redis = Depends(get_redis),
                        admin: AdminIdentity = Depends(require_operator)):
+    import json as _json
+
     from app.models.video import CollectionSource
     from sqlalchemy import select
 
     rows = (await db.execute(select(CollectionSource).order_by(CollectionSource.id.desc()))).scalars().all()
-    return ok({"list": [_source_dict(s) for s in rows]})
+    items = [_source_dict(s) for s in rows]
+    # 从 Redis 合并实时同步进度（同步任务可能跑在另一个 worker 进程上）
+    if rows:
+        vals = await redis.mget([_SYNC_PROGRESS_KEY.format(s.id) for s in rows])
+        for item, raw in zip(items, vals):
+            if raw:
+                try:
+                    item["syncing"] = True
+                    item["sync_progress"] = _json.loads(raw)
+                except Exception:
+                    pass
+    return ok({"list": items})
 
 
 @router.post("/collection/sources")
@@ -1317,10 +1338,27 @@ _sync_cancelled_sources: set[int] = set()
 
 
 async def _run_source_sync(source_id: int, max_pages: int) -> None:
+    import asyncio
+    import json as _json
+    from datetime import datetime
+
     from app.database import SessionFactory
     from app.models.video import CollectionSource
+    from app.redis_client import redis_client
     from app.services import aidrama91_service, collection_service
 
+    # 进度写 Redis（多 worker 共享，列表接口可能落在另一个进程上查询）
+    key = _SYNC_PROGRESS_KEY.format(source_id)
+    progress: dict = {
+        "started_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "pages": 0,
+    }
+    await redis_client.set(key, _json.dumps(progress), ex=6 * 3600)  # TTL 兜底：进程被杀残留 key 自动过期
+
+    def _progress(stat: dict) -> None:
+        progress.update(stat)
+        # fire-and-forget：回调在采集循环内同步触发，不阻塞采集
+        asyncio.create_task(redis_client.set(key, _json.dumps(progress), ex=6 * 3600))
+
     try:
         async with SessionFactory() as s:
             source = await s.get(CollectionSource, source_id)
@@ -1332,11 +1370,13 @@ async def _run_source_sync(source_id: int, max_pages: int) -> None:
                 if source.kind == aidrama91_service.KIND:
                     await aidrama91_service.refresh_source_id(s, source)
                     stat = await aidrama91_service.sync_source(
-                        s, source, max_pages=max_pages, should_cancel=_cancelled)
+                        s, source, max_pages=max_pages, should_cancel=_cancelled,
+                        on_progress=_progress)
                 elif source.kind == aidrama91_service.VLOG_KIND:
                     await aidrama91_service.refresh_source_id(s, source)
                     stat = await aidrama91_service.sync_vlog_source(
-                        s, source, max_pages=max_pages, should_cancel=_cancelled)
+                        s, source, max_pages=max_pages, should_cancel=_cancelled,
+                        on_progress=_progress)
                 else:
                     stat = await collection_service.sync_source(
                         s, source, max_pages=max_pages, should_cancel=_cancelled)
@@ -1349,6 +1389,7 @@ async def _run_source_sync(source_id: int, max_pages: int) -> None:
     finally:
         _sync_running_sources.discard(source_id)
         _sync_cancelled_sources.discard(source_id)
+        await redis_client.delete(key)
 
 
 class _SourceSyncReq(BaseModel):

server/app/services/aidrama91_service.py

@@ -477,11 +477,12 @@ async def import_drama(
 
 async def sync_source(
     db: AsyncSession, source: CollectionSource, max_pages: int = 0, delay: float = 0.3,
-    should_cancel=None,
+    should_cancel=None, on_progress=None,
 ) -> dict:
     """按列表翻页增量采集（已导入的短剧跳过，仍会补同步新集数）
 
     should_cancel: 可选回调，返回 True 时在每页/每条边界退出循环（用于手动取消）
+    on_progress: 可选回调，每页拉到列表后调用 on_progress(dict(stat))（用于后台展示进度）
     """
     cfg = load_config(source)
     # 每条/每页之间的间隔可由采集源 config 的 delay 自定义（秒），未配置时用默认值
@@ -515,6 +516,8 @@ async def sync_source(
                     stat["total"] = data["total"]
                     items = data["list"]
                     logger.info("91 短剧第 %s 页：%s 部（共 %s）", page, len(items), data["total"])
+                    if on_progress:
+                        on_progress(dict(stat))
 
                     for drama in items:
                         if _cancelled():
@@ -747,9 +750,10 @@ async def import_vlog_video(
 
 async def sync_vlog_source(
     db: AsyncSession, source: CollectionSource, max_pages: int = 0, delay: float = 0.3,
-    should_cancel=None,
+    should_cancel=None, on_progress=None,
 ) -> dict:
-    """vlog 短视频翻页增量采集。should_cancel 返回 True 时每页/每条边界退出"""
+    """vlog 短视频翻页增量采集。should_cancel 返回 True 时每页/每条边界退出；
+    on_progress 在每页拉到列表后回调 dict(stat)（用于后台展示进度）"""
     cfg = load_config(source)
     # 每条/每页之间的间隔可由采集源 config 的 delay 自定义（秒），未配置时用默认值
     if cfg.get("delay") is not None:
@@ -781,6 +785,8 @@ async def sync_vlog_source(
                     stat["pages"] = page
                     items = data["list"]
                     logger.info("91 vlog 第 %s 页：%s 条", page, len(items))
+                    if on_progress:
+                        on_progress(dict(stat))
 
                     for item in items:
                         if _cancelled():