修复腾讯文档子表改名后同步失败

scripts/sync_qq_jobs.py

@@ -171,6 +171,20 @@ def _workbook_from_payload(payload):
     return workbook
 
 
+def _resolve_sheet_id(workbook, view_name, preferred_sheet_id=None):
+    if preferred_sheet_id:
+        id_matches = [item for item in workbook if item.get("id") == preferred_sheet_id]
+        if len(id_matches) == 1:
+            return preferred_sheet_id
+        if len(id_matches) > 1:
+            raise IntegrityError(f"workbook 中 sheet_id 重复：{preferred_sheet_id}")
+
+    name_matches = [item for item in workbook if item.get("name") == view_name]
+    if len(name_matches) != 1 or not name_matches[0].get("id"):
+        raise IntegrityError(f"workbook 中无法唯一定位视图：{view_name}")
+    return name_matches[0]["id"]
+
+
 def _extract_structured_links(value):
     links = []
 
@@ -317,8 +331,7 @@ def parse_sheet_pages(view_name, payloads, source_url, document_id):
         raise IntegrityError("分页范围未覆盖源端总量")
     if not isinstance(workbook, list):
         raise IntegrityError("缺少 workbook 视图清单")
-    matches = [item for item in workbook if item.get("name") == view_name]
-    if len(matches) != 1 or matches[0].get("id") != sheet_id:
+    if _resolve_sheet_id(workbook, view_name, sheet_id) != sheet_id:
         raise IntegrityError(f"workbook 中无法唯一定位视图：{view_name}")
     if field_definitions is None or field_order is None or row_order is None:
         raise IntegrityError("完整分页中缺少 schema 元数据")
@@ -497,10 +510,13 @@ def schedule_capture(response):
                 workbook = await self._wait_for_workbook(page, captures)
                 target_sheets = {}
                 for spec in specs:
-                    matches = [item for item in workbook if item.get("name") == spec.view_name]
-                    if len(matches) != 1 or not matches[0].get("id"):
-                        raise IntegrityError(f"workbook 中无法唯一定位视图：{spec.view_name}")
-                    target_sheets[spec.view_name] = matches[0]["id"]
+                    previous = _load_previous(spec.data_path)
+                    preferred_sheet_id = previous.get("source", {}).get("sheet_id")
+                    target_sheets[spec.view_name] = _resolve_sheet_id(
+                        workbook,
+                        spec.view_name,
+                        preferred_sheet_id,
+                    )
 
                 grace_deadline = time.monotonic() + min(30, self.timeout_seconds / 2)
                 while time.monotonic() < grace_deadline:

tests/test_sync_qq_jobs.py

@@ -16,6 +16,7 @@
 from sync_qq_jobs import (
     DatasetSpec,
     IntegrityError,
+    _resolve_sheet_id,
     assert_matching_scans,
     check_link_accessibility,
     merge_history,
@@ -171,6 +172,37 @@ def test_rejects_field_id_that_is_not_in_view_schema(self):
             validate_snapshot(snapshot)
 
 
+class ResolveSheetTests(unittest.TestCase):
+    def test_prefers_stable_sheet_id_when_source_view_is_renamed(self):
+        workbook = [
+            {"id": "sheet_daily", "name": "每日更新", "type": "smartsheet"},
+            {"id": "sheet_early", "name": "27届秋招提前批（内推）", "type": "smartsheet"},
+        ]
+
+        self.assertEqual(
+            "sheet_early",
+            _resolve_sheet_id(workbook, "秋招提前批（内推）", "sheet_early"),
+        )
+
+    def test_falls_back_to_unique_view_name_without_saved_sheet_id(self):
+        workbook = [
+            {"id": "sheet_daily", "name": "每日更新", "type": "smartsheet"},
+        ]
+
+        self.assertEqual("sheet_daily", _resolve_sheet_id(workbook, "每日更新", None))
+
+    def test_stable_sheet_id_disambiguates_duplicate_view_names(self):
+        workbook = [
+            {"id": "sheet_old", "name": "秋招提前批（内推）", "type": "smartsheet"},
+            {"id": "sheet_current", "name": "秋招提前批（内推）", "type": "smartsheet"},
+        ]
+
+        self.assertEqual(
+            "sheet_current",
+            _resolve_sheet_id(workbook, "秋招提前批（内推）", "sheet_current"),
+        )
+
+
 class MergeHistoryTests(unittest.TestCase):
     def test_first_missing_observation_keeps_record_and_marks_pending(self):
         previous = {
@@ -823,7 +855,7 @@ def test_parses_crdt_pages_with_stable_ids_and_raw_links(self):
             0,
             2,
             [metadata_op, {"t": 3028, "v": 5, "c": {"k1": "sheet_1", "k2": {"k1": {"rec_1": first_record}}}}],
-            workbook=[{"id": "sheet_1", "name": "每日更新", "type": "smartsheet"}],
+            workbook=[{"id": "sheet_1", "name": "已改名的每日更新", "type": "smartsheet"}],
         )
         second_page = self._payload(
             "sheet_1",