fix: defer source deletion confirmation

docs/腾讯文档招聘信息同步到 GoClub-feature-qiuzhao/implementation.md

@@ -126,7 +126,7 @@ data/jobs 目录当前不存在。
 
 **位置**：merge_history、render_markdown、atomic_replace
 
-将完整新快照与上次成功快照按 record_id 合并。新增记录追加，现有记录覆盖字段更新，缺失记录保留历史；计划任务检测到源记录缺失时先失败并告警，只有手动触发显式确认删除后才标记为源表已移除。所有 JSON 和 Markdown 在临时目录完成并通过回读校验后统一替换。
+将完整新快照与上次成功快照按 record_id 合并。新增记录追加，现有记录覆盖字段更新，缺失记录首次出现时保留历史并写入待确认 ID，下一次独立同步仍缺失才标记为源表已移除；记录恢复时清除待确认状态。所有 JSON 和 Markdown 在临时目录完成并通过回读校验后统一替换。
 
 **当前代码**：
 ```python
@@ -225,7 +225,7 @@ scripts/sync_qq_jobs.py 当前不存在。
 
 - 文件：`GitHub Repository Secret`；位置：Settings > Secrets and variables > Actions；调整：创建 QQ_DOCS_STORAGE_STATE_B64，保存专用账号 Playwright storage state 的 Base64；会话过期后重复一次可见登录并覆盖该 Secret。；检查：手动运行 probe-only workflow，确认两个数据集名称、记录数和哈希摘要通过。
 - 文件：`.github/workflows/qq-jobs-sync.yml`；位置：on.schedule；调整：使用 0 */3 * * *，按 UTC 每 3 小时运行一次。
-- 文件：`.github/workflows/qq-jobs-sync.yml`；位置：workflow_dispatch.inputs.accept_source_deletions；调整：默认 false；人工核对源表后才允许将缺失记录标记为源表已移除。
+- 文件：`scripts/sync_qq_jobs.py`；调整：源记录首次缺失时保留旧记录并写入 `pending_source_deletions`，下一次独立同步仍缺失才标记为 `source_removed`；记录恢复时自动清除待确认状态。保留 `workflow_dispatch.inputs.accept_source_deletions` 作为需要立即确认时的人工兜底入口。
 
 ---
 

scripts/sync_qq_jobs.py

@@ -29,10 +29,6 @@ class IntegrityError(ValueError):
     """Raised when a source snapshot cannot be proven complete."""
 
 
-class SourceDeletionError(IntegrityError):
-    """Raised when records disappear without explicit operator approval."""
-
-
 @dataclass(frozen=True)
 class DatasetSpec:
     view_name: str
@@ -584,27 +580,44 @@ def merge_history(previous, current, now, accept_source_deletions=False):
         for record in previous.get("records", [])
         if record.get("status", "active") == "active" and record["record_id"] not in current_ids
     )
+    previous_pending = set(previous.get("snapshot", {}).get("pending_source_deletions", []))
+    confirmed_missing_ids = set(missing_ids) & previous_pending
+    pending_missing_ids = set(missing_ids) - confirmed_missing_ids
+    if accept_source_deletions:
+        confirmed_missing_ids = set(missing_ids)
+        pending_missing_ids = set()
     removed_history = [
         deepcopy(record)
         for record in previous.get("records", [])
         if record.get("status") == "source_removed" and record["record_id"] not in current_ids
     ]
-    if missing_ids and not accept_source_deletions:
-        raise SourceDeletionError(f"源记录消失，需人工确认：{', '.join(missing_ids)}")
     merged = deepcopy(current)
+    merged.setdefault("snapshot", {})["pending_source_deletions"] = sorted(pending_missing_ids)
     for record in merged["records"]:
         previous_record = previous_by_id.get(record["record_id"], {})
         record["status"] = "active"
         record["first_seen_at"] = previous_record.get("first_seen_at", now)
         record["last_seen_at"] = now
         record["removed_at"] = None
+    for record_id in sorted(pending_missing_ids):
+        retained = deepcopy(previous_by_id[record_id])
+        retained["status"] = "active"
+        retained["removed_at"] = None
+        merged["records"].append(retained)
     merged["records"].extend(removed_history)
     if accept_source_deletions:
         for record_id in missing_ids:
             removed = deepcopy(previous_by_id[record_id])
             removed["status"] = "source_removed"
             removed["removed_at"] = now
             merged["records"].append(removed)
+        merged["snapshot"]["pending_source_deletions"] = []
+    else:
+        for record_id in sorted(confirmed_missing_ids):
+            removed = deepcopy(previous_by_id[record_id])
+            removed["status"] = "source_removed"
+            removed["removed_at"] = now
+            merged["records"].append(removed)
     return merged
 
 

tests/test_sync_qq_jobs.py

@@ -16,7 +16,6 @@
 from sync_qq_jobs import (
     DatasetSpec,
     IntegrityError,
-    SourceDeletionError,
     assert_matching_scans,
     check_link_accessibility,
     merge_history,
@@ -173,7 +172,7 @@ def test_rejects_field_id_that_is_not_in_view_schema(self):
 
 
 class MergeHistoryTests(unittest.TestCase):
-    def test_scheduled_sync_rejects_missing_source_records(self):
+    def test_first_missing_observation_keeps_record_and_marks_pending(self):
         previous = {
             "source": {"view_id": "sc_daily", "view_name": "每日更新"},
             "schema": [],
@@ -196,8 +195,78 @@ def test_scheduled_sync_rejects_missing_source_records(self):
             "records": [],
         }
 
-        with self.assertRaisesRegex(SourceDeletionError, "rec_1"):
-            merge_history(previous, current, now="2026-08-07T00:00:00Z")
+        merged = merge_history(previous, current, now="2026-08-07T00:00:00Z")
+
+        self.assertEqual(["rec_1"], [record["record_id"] for record in merged["records"]])
+        self.assertEqual("active", merged["records"][0]["status"])
+        self.assertEqual(["rec_1"], merged["snapshot"]["pending_source_deletions"])
+
+    def test_second_consecutive_missing_observation_marks_record_removed(self):
+        previous = {
+            "source": {"view_id": "sc_daily", "view_name": "每日更新"},
+            "schema": [],
+            "snapshot": {
+                "source_total": 1,
+                "fetched_count": 1,
+                "pagination_complete": True,
+                "pending_source_deletions": ["rec_1"],
+            },
+            "records": [
+                {
+                    "record_id": "rec_1",
+                    "status": "active",
+                    "first_seen_at": "2026-08-01T00:00:00Z",
+                    "last_seen_at": "2026-08-06T00:00:00Z",
+                    "removed_at": None,
+                    "fields": [],
+                }
+            ],
+        }
+        current = {
+            "source": previous["source"],
+            "schema": [],
+            "snapshot": {"source_total": 0, "fetched_count": 0, "pagination_complete": True},
+            "records": [],
+        }
+
+        merged = merge_history(previous, current, now="2026-08-08T00:00:00Z")
+
+        self.assertEqual("source_removed", merged["records"][0]["status"])
+        self.assertEqual("2026-08-08T00:00:00Z", merged["records"][0]["removed_at"])
+        self.assertEqual([], merged["snapshot"]["pending_source_deletions"])
+
+    def test_returned_record_clears_pending_deletion(self):
+        previous = {
+            "source": {"view_id": "sc_daily", "view_name": "每日更新"},
+            "schema": [],
+            "snapshot": {
+                "source_total": 1,
+                "fetched_count": 1,
+                "pagination_complete": True,
+                "pending_source_deletions": ["rec_1"],
+            },
+            "records": [
+                {
+                    "record_id": "rec_1",
+                    "status": "active",
+                    "first_seen_at": "2026-08-01T00:00:00Z",
+                    "last_seen_at": "2026-08-06T00:00:00Z",
+                    "removed_at": None,
+                    "fields": [],
+                }
+            ],
+        }
+        current = {
+            "source": previous["source"],
+            "schema": [],
+            "snapshot": {"source_total": 1, "fetched_count": 1, "pagination_complete": True},
+            "records": [{"record_id": "rec_1", "fields": []}],
+        }
+
+        merged = merge_history(previous, current, now="2026-08-08T00:00:00Z")
+
+        self.assertEqual("active", merged["records"][0]["status"])
+        self.assertEqual([], merged["snapshot"]["pending_source_deletions"])
 
     def test_manual_confirmation_preserves_and_marks_removed_record(self):
         previous = {