refactor: 代码简化 - TTL 共享函数 + 参数收窄 + source 常量化 + 去重复 UI

Author: Astro-Han

quantclass_sync_internal/gui/assets/index.html

@@ -338,10 +338,7 @@ <h2 class="setup-title">QuantClass Sync 初始设置</h2>
             </div>
             <span class="qs-progress-text" x-text="total > 0 ? completed + ' / ' + total : '准备中...'"></span>
           </div>
-          <!-- 进度指示行 -->
-          <div x-show="total === 0" style="font-size:0.92em; color:#6b7280; margin:6px 0 2px;">
-            正在准备同步计划...
-          </div>
+          <!-- 进度指示行（仅在有产品时显示详情，准备中状态由进度条文字承担） -->
           <div x-show="total > 0" style="font-size:0.92em; color:#6b7280; margin:6px 0 2px;">
             <span x-text="'已完成 ' + completed + '/' + total"></span>
             <span x-show="currentProduct"> · 最新: <span style="font-weight:600" x-text="currentProduct"></span></span>

quantclass_sync_internal/orchestrator.py

@@ -99,6 +99,18 @@
     write_local_timestamp,
 )
 
+
+def _is_cache_fresh(checked_at_str: str) -> bool:
+    """检查缓存时间戳是否在 TTL 内。旧格式（无 T）视为过期。"""
+    if "T" not in checked_at_str:
+        return False
+    try:
+        checked_at = datetime.strptime(checked_at_str, "%Y-%m-%dT%H:%M:%S")
+        return (datetime.now() - checked_at).total_seconds() < API_DATE_CACHE_TTL_SECONDS
+    except ValueError:
+        return False
+
+
 def process_product(
     plan: ProductPlan,
     date_time: Optional[str],
@@ -510,19 +522,19 @@ def _resolve_requested_dates_for_plan(
         cached = api_date_cache.get(product_name) or api_date_cache.get(plan.name)
         if cached:
             cached_date, checked_at_str = cached
-            if "T" in checked_at_str:  # 旧日期格式无法比较，视为过期
+            if _is_cache_fresh(checked_at_str):
+                # 计算缓存年龄用于日志
                 try:
                     checked_at = datetime.strptime(checked_at_str, "%Y-%m-%dT%H:%M:%S")
                     age_seconds = (datetime.now() - checked_at).total_seconds()
-                    if age_seconds < API_DATE_CACHE_TTL_SECONDS:
-                        log_info(
-                            f"[{plan.name}] 使用缓存 API 日期 {cached_date}（{int(age_seconds)}s 前查询）",
-                            event="PRODUCT_PLAN", decision="cache_hit",
-                        )
-                        api_latest_candidates = [cached_date]
-                        cache_hit = True
                 except ValueError:
-                    pass  # 解析失败，回退 HTTP
+                    age_seconds = 0.0
+                log_info(
+                    f"[{plan.name}] 使用缓存 API 日期 {cached_date}（{int(age_seconds)}s 前查询）",
+                    event="PRODUCT_PLAN", decision="cache_hit",
+                )
+                api_latest_candidates = [cached_date]
+                cache_hit = True
 
     if not cache_hit:
         try:
@@ -892,32 +904,27 @@ def _maybe_run_coin_preprocess(
 
 def _prefetch_api_dates(
     products: List[str],
-    api_base: str,
+    command_ctx: "CommandContext",
     hid: str,
     headers: Dict[str, str],
-    log_dir: Path,
     max_workers: int = 8,
 ) -> Dict[str, Tuple[str, str]]:
     """并发预取产品的 API 最新日期，写入缓存并返回。
 
     已在缓存中且未过期的产品跳过。失败的产品静默跳过，
     Plan 阶段会回退到逐产品 HTTP 查询。
     """
+    api_base = command_ctx.api_base.rstrip("/")
+    log_dir = report_dir_path(command_ctx.data_root)
     # 1. 读现有缓存，筛出需要查询的产品
     existing_cache = load_api_latest_dates(log_dir)
-    now = datetime.now()
     uncached = []
     for product in products:
         cached = existing_cache.get(product)
         if cached:
             _, checked_at_str = cached
-            if "T" in checked_at_str:
-                try:
-                    checked_at = datetime.strptime(checked_at_str, "%Y-%m-%dT%H:%M:%S")
-                    if (now - checked_at).total_seconds() < API_DATE_CACHE_TTL_SECONDS:
-                        continue  # 缓存新鲜，跳过
-                except ValueError:
-                    pass
+            if _is_cache_fresh(checked_at_str):
+                continue  # 缓存新鲜，跳过
         uncached.append(product)
 
     if not uncached:
@@ -1027,10 +1034,9 @@ def _execute_plans(
     product_names = [normalize_product_name(p.name) for p in plans]
     _api_date_cache = _prefetch_api_dates(
         products=product_names,
-        api_base=command_ctx.api_base.rstrip("/"),
+        command_ctx=command_ctx,
         hid=hid,
         headers=headers,
-        log_dir=report_dir_path(command_ctx.data_root),
     )
 
     # stop-on-error 要求严格顺序控制，强制串行

quantclass_sync_internal/status_store.py

@@ -32,6 +32,10 @@
 
 _RUN_SCOPE_PATTERN = re.compile(r"^\d{8}-\d{6}(?:[-_].+)?$")
 
+# 产品最后状态写入来源标识常量
+_SOURCE_API_CHECK = "api_check"
+_SOURCE_SYNC = "sync"
+
 
 def _status_db_has_rows(path: Path) -> bool:
     """判断状态库是否可用（存在 product_status 且至少 1 行）。"""
@@ -472,7 +476,7 @@ def _update_product_last_status(log_dir: Path, report: RunReport) -> None:
                 "error": item.error,
                 "date_time": item.date_time,
                 "checked_at": datetime.now().strftime("%Y-%m-%dT%H:%M:%S"),
-                "source": "sync",
+                "source": _SOURCE_SYNC,
             }
         # 原子写入
         with atomic_temp_path(status_path, tag="last_status") as tmp:
@@ -505,12 +509,12 @@ def update_api_latest_dates(log_dir: Path, api_latest_dates: Dict[str, str]) ->
             if product in existing:
                 existing[product]["date_time"] = date_str
                 existing[product]["checked_at"] = checked_at
-                existing[product]["source"] = "api_check"
+                existing[product]["source"] = _SOURCE_API_CHECK
             else:
                 existing[product] = {
                     "status": "", "reason_code": "", "error": "",
                     "date_time": date_str, "checked_at": checked_at,
-                    "source": "api_check",
+                    "source": _SOURCE_API_CHECK,
                 }
         with atomic_temp_path(status_path, tag="last_status") as tmp:
             tmp.write_text(json.dumps(existing, ensure_ascii=False, indent=2), encoding="utf-8")
@@ -537,7 +541,7 @@ def load_api_latest_dates(log_dir: Path) -> Dict[str, Tuple[str, str]]:
         if not isinstance(info, dict):
             continue
         # 只读取 check_updates 写入的记录，排除同步结果
-        if info.get("source") != "api_check":
+        if info.get("source") != _SOURCE_API_CHECK:
             continue
         dt, ca = info.get("date_time"), info.get("checked_at")
         if isinstance(dt, str) and isinstance(ca, str):

tests/test_status_store.py

@@ -9,6 +9,8 @@
 from quantclass_sync_internal.models import ProductStatus
 from quantclass_sync_internal.reporting import _append_result, _new_report
 from quantclass_sync_internal.status_store import (
+    _SOURCE_API_CHECK,
+    _SOURCE_SYNC,
     _update_product_last_status,
     ensure_status_table,
     export_status_json,
@@ -167,7 +169,7 @@ def test_update_api_latest_dates_writes_iso_datetime_and_source(self):
             entry = status["product-a"]
             self.assertIn("T", entry["checked_at"])
             self.assertRegex(entry["checked_at"], r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}$")
-            self.assertEqual(entry["source"], "api_check")
+            self.assertEqual(entry["source"], _SOURCE_API_CHECK)
 
     def test_product_last_status_writes_iso_datetime_and_sync_source(self):
         """_update_product_last_status 应写入 ISO datetime 和 source="sync"。"""
@@ -179,7 +181,7 @@ def test_product_last_status_writes_iso_datetime_and_sync_source(self):
             status = json.loads((log_dir / "product_last_status.json").read_text())
             entry = status["p1"]
             self.assertIn("T", entry["checked_at"])
-            self.assertEqual(entry["source"], "sync")
+            self.assertEqual(entry["source"], _SOURCE_SYNC)
 
 
 class TestLoadApiLatestDates(unittest.TestCase):
@@ -227,9 +229,9 @@ def test_sync_result_excluded(self):
             status_path = log_dir / "product_last_status.json"
             status_path.write_text(json.dumps({
                 "prod-a": {"date_time": "2026-03-18", "checked_at": "2026-03-18T10:00:00",
-                           "source": "api_check", "status": ""},
+                           "source": _SOURCE_API_CHECK, "status": ""},
                 "prod-b": {"date_time": "2026-03-17", "checked_at": "2026-03-18T10:00:00",
-                           "source": "sync", "status": "ok"},
+                           "source": _SOURCE_SYNC, "status": "ok"},
             }))
             cache = load_api_latest_dates(log_dir)
             self.assertIn("prod-a", cache)

tests/test_update_catchup.py

@@ -10,7 +10,7 @@
 from quantclass_sync_internal.models import CommandContext, EmptyDownloadLinkError, FatalRequestError, ProductPlan, ProductSyncError, RunReport, SyncStats
 from quantclass_sync_internal.orchestrator import _execute_plans, _resolve_requested_dates_for_plan
 from quantclass_sync_internal.reporting import _new_report
-from quantclass_sync_internal.status_store import connect_status_db
+from quantclass_sync_internal.status_store import _SOURCE_API_CHECK, connect_status_db
 
 
 class UpdateCatchUpTests(unittest.TestCase):
@@ -705,21 +705,31 @@ class TestPrefetchApiDates(unittest.TestCase):
     def setUp(self):
         self._tmpdir = tempfile.TemporaryDirectory()
         self.root = Path(self._tmpdir.name)
+        # log_dir 路径与 report_dir_path(data_root) 保持一致：data_root/.quantclass_sync/log
         self.log_dir = self.root / ".quantclass_sync" / "log"
         self.log_dir.mkdir(parents=True)
 
     def tearDown(self):
         self._tmpdir.cleanup()
 
+    def _ctx(self) -> CommandContext:
+        """构造测试用 CommandContext，api_base 设为 http://fake。"""
+        return CommandContext(
+            run_id="test",
+            data_root=self.root,
+            dry_run=False,
+            stop_on_error=False,
+            api_base="http://fake",
+        )
+
     @patch("quantclass_sync_internal.orchestrator.get_latest_time")
     def test_fetches_uncached_products(self, mock_get):
         """无缓存时并发调用 get_latest_time。"""
         mock_get.return_value = "2026-03-18"
         from quantclass_sync_internal.orchestrator import _prefetch_api_dates
         cache = _prefetch_api_dates(
             products=["prod-a", "prod-b"],
-            api_base="http://fake", hid="hid", headers={},
-            log_dir=self.log_dir,
+            command_ctx=self._ctx(), hid="hid", headers={},
         )
         self.assertEqual(mock_get.call_count, 2)
         self.assertIn("prod-a", cache)
@@ -734,8 +744,7 @@ def test_skips_fresh_cache(self, mock_get):
         update_api_latest_dates(self.log_dir, {"prod-a": "2026-03-18"})
         cache = _prefetch_api_dates(
             products=["prod-a"],
-            api_base="http://fake", hid="hid", headers={},
-            log_dir=self.log_dir,
+            command_ctx=self._ctx(), hid="hid", headers={},
         )
         mock_get.assert_not_called()
         self.assertIn("prod-a", cache)
@@ -749,8 +758,7 @@ def test_partial_cache_only_fetches_missing(self, mock_get):
         update_api_latest_dates(self.log_dir, {"prod-a": "2026-03-18"})
         cache = _prefetch_api_dates(
             products=["prod-a", "prod-b"],
-            api_base="http://fake", hid="hid", headers={},
-            log_dir=self.log_dir,
+            command_ctx=self._ctx(), hid="hid", headers={},
         )
         # 只查了 prod-b
         self.assertEqual(mock_get.call_count, 1)
@@ -763,8 +771,7 @@ def test_failure_returns_partial_cache(self, mock_get):
         from quantclass_sync_internal.orchestrator import _prefetch_api_dates
         cache = _prefetch_api_dates(
             products=["prod-a"],
-            api_base="http://fake", hid="hid", headers={},
-            log_dir=self.log_dir,
+            command_ctx=self._ctx(), hid="hid", headers={},
         )
         # 失败但不抛异常，返回空缓存
         self.assertEqual(cache, {})
@@ -783,14 +790,13 @@ def test_expired_cache_refetches(self, mock_get):
             "prod-a": {
                 "date_time": "2026-03-17",
                 "checked_at": expired_time.strftime("%Y-%m-%dT%H:%M:%S"),
-                "source": "api_check",
+                "source": _SOURCE_API_CHECK,
             }
         }))
         mock_get.return_value = "2026-03-18"
         cache = _prefetch_api_dates(
             products=["prod-a"],
-            api_base="http://fake", hid="hid", headers={},
-            log_dir=self.log_dir,
+            command_ctx=self._ctx(), hid="hid", headers={},
         )
         mock_get.assert_called_once()  # 过期缓存触发重查
         self.assertEqual(cache["prod-a"][0], "2026-03-18")  # 返回新日期