From 1d9cd1f9c70ad5bc198ef2a281c8edbd6f44cdf2 Mon Sep 17 00:00:00 2001
From: lz_db <lz_db@lz.com>
Date: Sun, 25 Oct 2026 16:14:26 +0800
Subject: [PATCH] 1

---
 admin/src/views/CollectSources.vue            | 43 +++++++++-
 app/lib/core/player/feed_player_pool.dart     | 17 ++++
 app/lib/core/storage/feed_progress_store.dart | 79 +++++++++++++++++++
 app/lib/main.dart                             |  3 +
 app/lib/providers/feed_provider.dart          | 43 +++++++++-
 app/lib/screens/feed/feed_screen.dart         | 50 ++++++++++++
 app/lib/screens/home/home_shell.dart          |  5 +-
 app/pubspec.yaml                              |  2 +-
 server/app/api/v1/admin.py                    | 47 ++++++++++-
 server/app/services/aidrama91_service.py      | 12 ++-
 10 files changed, 288 insertions(+), 13 deletions(-)
 create mode 100644 app/lib/core/storage/feed_progress_store.dart

diff --git a/admin/src/views/CollectSources.vue b/admin/src/views/CollectSources.vue
index b239e33..ee92f6f 100644
--- a/admin/src/views/CollectSources.vue
+++ b/admin/src/views/CollectSources.vue
@@ -47,6 +47,28 @@
         <el-table-column prop="last_sync_at" label="上次同步" width="160">
           <template #default="{ row }">{{ row.last_sync_at || '从未' }}</template>
         </el-table-column>
+        <el-table-column label="同步状态" width="220">
+          <template #default="{ row }">
+            <template v-if="row.syncing">
+              <el-tag type="primary" size="small" effect="dark">
+                <el-icon class="is-loading" style="margin-right: 4px"><Loading /></el-icon>同步中
+              </el-tag>
+              <div style="font-size: 12px; color: #606266; margin-top: 4px">
+                <span v-if="row.sync_progress">第 {{ row.sync_progress.pages }} 页</span>
+                <template v-if="row.sync_progress && row.kind === 'aidrama91_vlog'">
+                  · 新增 {{ row.sync_progress.videos_new ?? 0 }} / 跳过 {{ row.sync_progress.videos_exist ?? 0 }}
+                </template>
+                <template v-else-if="row.sync_progress && row.kind === 'aidrama91'">
+                  · 新剧 {{ row.sync_progress.dramas_new ?? 0 }} / 集数 +{{ row.sync_progress.episodes_created ?? 0 }}
+                </template>
+                <div v-if="row.sync_progress && row.sync_progress.started_at" style="color: #909399">
+                  开始于 {{ row.sync_progress.started_at }}
+                </div>
+              </div>
+            </template>
+            <span v-else style="color: #c0c4cc">-</span>
+          </template>
+        </el-table-column>
         <el-table-column label="操作" width="290" fixed="right">
           <template #default="{ row }">
             <el-button link type="warning" :loading="syncingId === row.id" @click="syncOne(row)">同步</el-button>
@@ -144,8 +166,9 @@
 </template>
 
 <script setup>
-import { reactive, ref, onMounted } from 'vue'
+import { reactive, ref, onMounted, onBeforeUnmount } from 'vue'
 import { ElMessage, ElMessageBox } from 'element-plus'
+import { Loading } from '@element-plus/icons-vue'
 import request from '@/api/request'
 
 const list = ref([])
@@ -203,11 +226,23 @@ async function load() {
   try {
     const r = await request({ url: '/admin/collection/sources', method: 'get' })
     list.value = r.list
+    schedulePoll()
   } catch (e) {} finally {
     loading.value = false
   }
 }
 
+// 有源正在同步时每 5 秒自动刷新进度；无任务时停止轮询
+let pollTimer = null
+function schedulePoll() {
+  if (pollTimer) { clearTimeout(pollTimer); pollTimer = null }
+  if (list.value.some(s => s.syncing)) {
+    pollTimer = setTimeout(load, 5000)
+  }
+}
+
+onBeforeUnmount(() => { if (pollTimer) clearTimeout(pollTimer) })
+
 function showCreate() {
   editing.value = null
   Object.assign(form, {
@@ -299,8 +334,8 @@ async function syncOne(row) {
   syncingId.value = row.id
   try {
     await request({ url: `/admin/collection/sources/${row.id}/sync`, method: 'post', data: { max_pages: 0 } })
-    ElMessage.success(`「${row.name}」已触发后台同步，稍后刷新查看`)
-    setTimeout(load, 20000)
+    ElMessage.success(`「${row.name}」已触发后台同步`)
+    setTimeout(load, 1500)
   } catch (e) {} finally {
     syncingId.value = 0
   }
@@ -328,7 +363,7 @@ async function syncAll() {
   try {
     const r = await request({ url: '/admin/collection/sources/sync-all', method: 'post' })
     ElMessage.success(`已触发 ${r.started.length} 个源的后台同步`)
-    setTimeout(load, 20000)
+    setTimeout(load, 1500)
   } catch (e) {} finally {
     syncAllLoading.value = false
   }
diff --git a/app/lib/core/player/feed_player_pool.dart b/app/lib/core/player/feed_player_pool.dart
index 6d46c92..fc21066 100644
--- a/app/lib/core/player/feed_player_pool.dart
+++ b/app/lib/core/player/feed_player_pool.dart
@@ -66,6 +66,23 @@ class FeedPlayerPool {
   /// open(play:false) 内部的 pause 覆盖导致恢复后不播）
   Future<void>? _pendingBind;
 
+  /// 当前槽正在播放的视频（-1 表示尚未绑定）与实时播放位置
+  int get currentVideoId => currentSlot.videoId;
+
+  Duration get currentPosition {
+    try {
+      return currentSlot.player.state.position;
+    } catch (_) {
+      return Duration.zero;
+    }
+  }
+
+  /// 冷启动续播：指定视频从 position 起播（bindWindow 时经 mpv start 属性应用）
+  void setResumePoint(int videoId, Duration position) {
+    _resumeVideoId = videoId;
+    _resumePosition = position;
+  }
+
   /// 初始化（创建 3 槽位并订阅各自进度流）
   void init() {
     for (var i = 0; i < slotCount; i++) {
diff --git a/app/lib/core/storage/feed_progress_store.dart b/app/lib/core/storage/feed_progress_store.dart
new file mode 100644
index 0000000..3cbda64
--- /dev/null
+++ b/app/lib/core/storage/feed_progress_store.dart
@@ -0,0 +1,79 @@
+import 'dart:convert';
+
+import 'package:shared_preferences/shared_preferences.dart';
+
+/// Feed 播放进度持久化：记录每个策略最后观看的视频与播放位置，
+/// App 冷启动重新拉列表后按 videoId 定位，继续上次的视频。
+class FeedProgress {
+  const FeedProgress({
+    required this.videoId,
+    required this.index,
+    required this.positionMs,
+  });
+
+  final int videoId;
+  final int index;
+  final int positionMs;
+}
+
+class FeedProgressStore {
+  FeedProgressStore._(this._prefs);
+
+  final SharedPreferences _prefs;
+
+  static const String _prefix = 'feed_progress_';
+  static const String _lastStrategyKey = 'feed_last_strategy';
+
+  static FeedProgressStore? _instance;
+
+  /// 初始化（main 启动阶段调用）
+  static Future<void> init() async {
+    _instance = FeedProgressStore._(await SharedPreferences.getInstance());
+  }
+
+  /// 测试注入
+  static void initForTest(SharedPreferences prefs) {
+    _instance = FeedProgressStore._(prefs);
+  }
+
+  static FeedProgressStore get I => _instance!;
+
+  Future<void> save(
+    String strategy, {
+    required int videoId,
+    required int index,
+    required int positionMs,
+  }) async {
+    await _prefs.setString(
+      '$_prefix$strategy',
+      jsonEncode({
+        'videoId': videoId,
+        'index': index,
+        'positionMs': positionMs,
+      }),
+    );
+  }
+
+  FeedProgress? read(String strategy) {
+    final raw = _prefs.getString('$_prefix$strategy');
+    if (raw == null) return null;
+    try {
+      final map = jsonDecode(raw) as Map<String, dynamic>;
+      final videoId = (map['videoId'] as num?)?.toInt() ?? 0;
+      if (videoId <= 0) return null;
+      return FeedProgress(
+        videoId: videoId,
+        index: (map['index'] as num?)?.toInt() ?? 0,
+        positionMs: (map['positionMs'] as num?)?.toInt() ?? 0,
+      );
+    } catch (_) {
+      return null;
+    }
+  }
+
+  Future<void> saveLastStrategy(String strategy) async {
+    await _prefs.setString(_lastStrategyKey, strategy);
+  }
+
+  String? readLastStrategy() => _prefs.getString(_lastStrategyKey);
+}
diff --git a/app/lib/main.dart b/app/lib/main.dart
index b167104..f542d8c 100644
--- a/app/lib/main.dart
+++ b/app/lib/main.dart
@@ -8,6 +8,7 @@ import 'package:media_kit/media_kit.dart';
 import 'core/log/action_logger.dart';
 import 'core/media/download_manager.dart';
 import 'core/media/local_media_server.dart';
+import 'core/storage/feed_progress_store.dart';
 import 'core/storage/local_store.dart';
 import 'core/storage/search_history_store.dart';
 import 'core/storage/watch_dedup_store.dart';
@@ -61,6 +62,8 @@ Future<void> main() async {
     await SearchHistoryStore.init();
     // 观看上报去重（持久化；不初始化则观看计费/游客免费额度链路失效）
     await WatchDedupStore.init();
+    // Feed 播放进度（冷启动续播：上次看到第几个视频/播放位置）
+    await FeedProgressStore.init();
     // 本地媒体服务：HLS 1G 分片缓存（回看免下载）+ 已下载视频离线播放
     await LocalMediaServer.I.init();
 
diff --git a/app/lib/providers/feed_provider.dart b/app/lib/providers/feed_provider.dart
index 769b714..0b57c15 100644
--- a/app/lib/providers/feed_provider.dart
+++ b/app/lib/providers/feed_provider.dart
@@ -5,6 +5,7 @@ import 'package:flutter_riverpod/flutter_riverpod.dart';
 import '../core/config/test_stream.dart';
 import '../core/network/api_exception.dart';
 import '../core/player/feed_player_pool.dart';
+import '../core/storage/feed_progress_store.dart';
 import '../core/storage/watch_dedup_store.dart';
 import '../models/video_item.dart';
 import '../services/video_service.dart';
@@ -134,13 +135,27 @@ class FeedNotifier extends StateNotifier<FeedState> {
     state = state.copyWith(isRefreshing: true, loadFailed: false);
     try {
       final page = await VideoService.fetchFeed(type: strategy.apiValue);
+      // 冷启动续播：按上次观看的 videoId 在新列表中定位（列表内容可能变化）；
+      // 找不到时回退保存的下标（仅当仍在当前页范围内）
+      final saved = FeedProgressStore.I.read(strategy.name);
+      var index = 0;
+      if (saved != null && page.items.isNotEmpty) {
+        final hit = page.items.indexWhere((v) => v.videoId == saved.videoId);
+        if (hit >= 0) {
+          index = hit;
+        } else if (saved.index < page.items.length) {
+          index = saved.index;
+        }
+        pool.setResumePoint(saved.videoId, Duration(milliseconds: saved.positionMs));
+      }
       state = FeedState(
         items: page.items,
         nextCursor: page.nextCursor,
+        currentIndex: index,
         freeRemainingToday: state.freeRemainingToday,
       );
       if (page.items.isNotEmpty) {
-        _bindAndReport(0);
+        _bindAndReport(index);
       }
     } catch (_) {
       state = state.copyWith(isRefreshing: false, loadFailed: true);
@@ -192,15 +207,41 @@ class FeedNotifier extends StateNotifier<FeedState> {
 
   /// 翻页：滑出即暂停、滑入自动播放并上报观看计费（《01-开发文档》5.3）
   void onPageChanged(int index) {
+    if (index < 0 || index >= state.items.length) return;
+    final item = state.items[index];
+    // 同一视频重新绑定（冷启动跳页）时保留播放位置，否则新页从 0 开始
+    final sameVideo = pool.currentVideoId == item.videoId;
     // 立即静音全部非目标槽位，防止翻页动画期间多个视频同时出声
     pool.pauseExcept(slotIndex: pool.slotOf(index));
     state = state.copyWith(currentIndex: index);
     // fire-and-forget：Future 内部会完整 await bindWindow 再 play，调用方不等
     _bindAndReport(index);
+    // 持久化进度（杀进程重进时续播）
+    unawaited(FeedProgressStore.I.save(
+      strategy.name,
+      videoId: item.videoId,
+      index: index,
+      positionMs: sameVideo ? pool.currentPosition.inMilliseconds : 0,
+    ));
     // 接近尾部预取下一页
     if (index >= state.items.length - 3) loadMore();
   }
 
+  /// 持久化当前播放进度（App 退后台 / 切走 Tab / 定时兜底调用）
+  Future<void> persistProgress() async {
+    final index = state.currentIndex;
+    if (index < 0 || index >= state.items.length) return;
+    final item = state.items[index];
+    final positionMs =
+        pool.currentVideoId == item.videoId ? pool.currentPosition.inMilliseconds : 0;
+    await FeedProgressStore.I.save(
+      strategy.name,
+      videoId: item.videoId,
+      index: index,
+      positionMs: positionMs,
+    );
+  }
+
   /// 绑定播放窗口 + 观看计费上报
   Future<void> _bindAndReport(int index) async {
     if (index < 0 || index >= state.items.length) return;
diff --git a/app/lib/screens/feed/feed_screen.dart b/app/lib/screens/feed/feed_screen.dart
index 671603b..5733e65 100644
--- a/app/lib/screens/feed/feed_screen.dart
+++ b/app/lib/screens/feed/feed_screen.dart
@@ -5,6 +5,7 @@ import 'package:flutter_riverpod/flutter_riverpod.dart';
 import 'package:go_router/go_router.dart';
 
 import '../../core/log/action_logger.dart';
+import '../../core/storage/feed_progress_store.dart';
 import '../../core/theme/app_theme.dart';
 import '../../models/video_item.dart';
 import '../../providers/feed_provider.dart';
@@ -30,10 +31,25 @@ class _FeedScreenState extends ConsumerState<FeedScreen>
   final PageController _pageController = PageController();
   StreamSubscription<FeedEvent>? _eventSub;
 
+  /// 已完成首屏位置同步的策略（冷启动续播跳页只需一次）
+  final Set<FeedStrategy> _initialSyncDone = {};
+
+  /// 定时兜底保存播放进度（被杀进程时也能续播）
+  Timer? _progressTimer;
+
+  /// App 是否在前台（退后台后不写进度）
+  bool _appActive = true;
+
   @override
   void initState() {
     super.initState();
     WidgetsBinding.instance.addObserver(this);
+    // 冷启动恢复上次所在的 Tab（口碑/热播等），再按该策略的进度续播
+    final lastName = FeedProgressStore.I.readLastStrategy();
+    _strategy = FeedStrategy.values.firstWhere(
+      (s) => s.name == lastName && s != FeedStrategy.search,
+      orElse: () => FeedStrategy.recommend,
+    );
     // 登记当前策略，HomeShell 切回首页时只恢复这个策略的播放器池
     ref.read(activeFeedStrategyProvider.notifier).state = _strategy;
     // 监听计费/UI 事件（弹窗与 Toast 由 Notifier 事件流驱动）
@@ -46,6 +62,11 @@ class _FeedScreenState extends ConsumerState<FeedScreen>
     WidgetsBinding.instance.addPostFrameCallback((_) {
       if (mounted) ref.read(feedProvider(_strategy).notifier).rebindCurrent();
     });
+    _progressTimer = Timer.periodic(const Duration(seconds: 5), (_) {
+      if (!mounted || !_appActive) return;
+      if (!ref.read(feedTabVisibleProvider)) return;
+      unawaited(ref.read(feedProvider(_strategy).notifier).persistProgress());
+    });
   }
 
   @override
@@ -78,6 +99,7 @@ class _FeedScreenState extends ConsumerState<FeedScreen>
   void dispose() {
     WidgetsBinding.instance.removeObserver(this);
     routeObserver.unsubscribe(this);
+    _progressTimer?.cancel();
     _eventSub?.cancel();
     // 离开 Feed 路由（如跳登录页）：彻底停止播放（断音+断缓存下载）。
     // feedProvider 是全局常驻，不处理则播放器池在后台继续出声
@@ -93,10 +115,15 @@ class _FeedScreenState extends ConsumerState<FeedScreen>
   void didChangeAppLifecycleState(AppLifecycleState state) {
     final notifier = ref.read(feedProvider(_strategy).notifier);
     if (state == AppLifecycleState.paused) {
+      _appActive = false;
+      // 先存进度再 suspend（suspend 会 stop 槽位，位置随即清零）
+      unawaited(notifier.persistProgress());
       notifier.pool.suspend();
     } else if (state == AppLifecycleState.inactive) {
+      unawaited(notifier.persistProgress());
       notifier.pool.pauseCurrent();
     } else if (state == AppLifecycleState.resumed) {
+      _appActive = true;
       // 仅在首页 Tab 可见时恢复（在"我的"等页面回前台不播）
       if (ref.read(feedTabVisibleProvider)) {
         unawaited(() async {
@@ -121,9 +148,16 @@ class _FeedScreenState extends ConsumerState<FeedScreen>
 
   void _switchStrategy(FeedStrategy next) {
     if (next == _strategy) return;
+    // 先保存旧策略进度再挂起（suspend 会 stop 槽位，位置随即清零）
+    unawaited(ref.read(feedProvider(_strategy).notifier).persistProgress());
     // 挂起旧策略池（防在途 open 完成后自动出声）
     ref.read(feedProvider(_strategy).notifier).pool.suspend();
+    unawaited(FeedProgressStore.I.saveLastStrategy(next.name));
     setState(() => _strategy = next);
+    // 会话内已有数据的策略由下方跳页逻辑处理，不再走冷启动位置同步
+    if (ref.read(feedProvider(next)).items.isNotEmpty) {
+      _initialSyncDone.add(next);
+    }
     ref.read(activeFeedStrategyProvider.notifier).state = next;
     _eventSub?.cancel();
     _eventSub = ref
@@ -175,6 +209,22 @@ class _FeedScreenState extends ConsumerState<FeedScreen>
     final state = ref.watch(feedProvider(_strategy));
     final notifier = ref.read(feedProvider(_strategy).notifier);
 
+    // 冷启动续播：列表加载完成后把 PageView 定位到恢复的下标（每个策略一次）。
+    // Notifier 会在 refresh 时按上次 videoId 恢复 currentIndex，此处只负责跳页，
+    // 跳页触发的 onPageChanged 会把新列表与播放器池对齐
+    ref.listen<FeedState>(feedProvider(_strategy), (previous, next) {
+      if (next.items.isEmpty) return;
+      if (!_initialSyncDone.add(_strategy)) return;
+      final target = next.currentIndex;
+      if (target <= 0) return;
+      WidgetsBinding.instance.addPostFrameCallback((_) {
+        if (!mounted || !_pageController.hasClients) return;
+        if ((_pageController.page?.round() ?? 0) != target) {
+          _pageController.jumpToPage(target);
+        }
+      });
+    });
+
     return Scaffold(
       backgroundColor: Colors.black,
       body: Stack(
diff --git a/app/lib/screens/home/home_shell.dart b/app/lib/screens/home/home_shell.dart
index 33a4f4e..45f8d60 100644
--- a/app/lib/screens/home/home_shell.dart
+++ b/app/lib/screens/home/home_shell.dart
@@ -33,7 +33,10 @@ class _HomeShellState extends ConsumerState<HomeShell> {
     // 回首页 rebind 重新 open + seek 断点续播，缓存分片零流量（抖音行为）
     if (i != 0) {
       for (final s in _feedStrategies) {
-        ref.read(feedProvider(s).notifier).pool.suspend();
+        final notifier = ref.read(feedProvider(s).notifier);
+        // 先存进度再挂起（suspend 会 stop 槽位，位置随即清零）
+        unawaited(notifier.persistProgress());
+        notifier.pool.suspend();
       }
     }
     ref.read(feedTabVisibleProvider.notifier).state = i == 0;
diff --git a/app/pubspec.yaml b/app/pubspec.yaml
index 8388ea4..7faf3a9 100644
--- a/app/pubspec.yaml
+++ b/app/pubspec.yaml
@@ -2,7 +2,7 @@ name: video_app
 description: "短视频应用（Flutter，iOS/Android 双端）"
 publish_to: 'none'
 
-version: 1.3.31+47
+version: 1.3.32+48
 
 environment:
   sdk: ^3.9.2
diff --git a/server/app/api/v1/admin.py b/server/app/api/v1/admin.py
index e86de39..ce352eb 100644
--- a/server/app/api/v1/admin.py
+++ b/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):
diff --git a/server/app/services/aidrama91_service.py b/server/app/services/aidrama91_service.py
index 2e37576..24f4f36 100644
--- a/server/app/services/aidrama91_service.py
+++ b/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():