From 4dcbf8ecb3499a534fce7066a9d7a20de61e9414 Mon Sep 17 00:00:00 2001 From: Your Name Date: Sun, 19 Apr 2026 02:53:43 +0800 Subject: [PATCH] feat: ingest mainland detect result events --- .../HANDOFF_20260419_0251.md | 97 +++++++++ .../IMPLEMENTATION_STATUS.md | 67 +++--- docs/ops_center_runtime/TASK_BOARD.md | 69 +++--- domain-api/app/services/sync_push_service.py | 204 +++++++++++++++++- .../app/services/sync_record_service.py | 32 +++ domain-api/tests/test_sync_push_service.py | 147 +++++++++++++ domain-api/tests/test_sync_record_service.py | 55 +++++ 7 files changed, 600 insertions(+), 71 deletions(-) create mode 100644 docs/ops_center_runtime/HANDOFF_20260419_0251.md create mode 100644 domain-api/tests/test_sync_push_service.py create mode 100644 domain-api/tests/test_sync_record_service.py diff --git a/docs/ops_center_runtime/HANDOFF_20260419_0251.md b/docs/ops_center_runtime/HANDOFF_20260419_0251.md new file mode 100644 index 0000000..e3d10f1 --- /dev/null +++ b/docs/ops_center_runtime/HANDOFF_20260419_0251.md @@ -0,0 +1,97 @@ +# HANDOFF 2026-04-19 02:51 + +## 本轮完成了什么 + +本轮只做了 `J1-大陆逐条结果同步前置批` 的最小实现，不进入新页面、不扩控制面： + +- `detect_result_projection` 新增 `recent_domain_events` +- 中央 ingest `detect_result_projection` 时开始尝试把逐条事件写入 `detect_run_events` +- 增加了 `import_fingerprint` 去重 +- 新增最小单测 +- 已重启中央 `domaincheck-api` + +## 已完成验证 + +已完成： + +- `python3 -m py_compile` +- `python -m unittest tests.test_sync_record_service tests.test_sync_push_service` +- 中央 `domaincheck-api` 重启成功 + +## 当前真实状态 + +要区分两层： + +### 1. 代码闭环 + +这一层已经补上： + +- 大陆侧投影格式支持逐条事件 +- 中央视角 ingest 支持把逐条事件落入 `detect_run_events` + +### 2. 真实运行闭环 + +这一层还差最后一步： + +- 大陆节点必须拉到最新代码 +- 然后继续运行同步 +- 中央才能真正看到 mainland `domain_*` + +## 当前最高优先级 + +唯一最高优先级： + +- 让大陆节点拉最新代码后，验证 `recent_domain_events -> central detect_run_events` + +## 下一轮只做什么 + +下一轮只做以下最小动作： + +1. 在大陆节点确认当前代码已是最新 +2. 触发或等待新的 `detect_result_projection` +3. 检查中央最新 `detect_result_ingest.payload.projection.recent_domain_events` +4. 检查中央 `detect_run_events` 是否出现： + - `mainland-controller-01` + - `mainland-worker-01` + - 对应 `domain_started/domain_completed/domain_failed/domain_blacklisted` + +## 下一轮不要做什么 + +继续不要做： + +- 新页面 +- 新模块 +- 控制面增强 +- 发布动作 +- item 级最终回写 + +## 推荐模型与推理等级 + +继续推荐： + +- `GPT-5.4` +- `high` + +## 给下一轮的最小任务卡 + +```text +任务名：J1-大陆逐条结果同步前置批（验证轮） + +目标： +1. 确认大陆节点已拉到最新代码 +2. 确认最新 detect_result_projection 已含 recent_domain_events +3. 确认中央 detect_run_events 已出现 mainland domain_* 事件 + +完成判定： +1. detect_result_ingest.payload.projection.recent_domain_events 非空 +2. detect_run_events 中可查询到 mainland-controller-01 / mainland-worker-01 的 domain_* 事件 + +边界： +1. 不做 item 级回写 +2. 不做新页面 +3. 不做控制面扩展 +``` + +## 一句话结论 + +这轮已经把“中央如何接并落大陆逐条事件”的代码闭环补上了；下一轮只差让大陆节点跑到这版代码并验证真实事件进中央。 diff --git a/docs/ops_center_runtime/IMPLEMENTATION_STATUS.md b/docs/ops_center_runtime/IMPLEMENTATION_STATUS.md index 55e37c9..f120c7a 100644 --- a/docs/ops_center_runtime/IMPLEMENTATION_STATUS.md +++ b/docs/ops_center_runtime/IMPLEMENTATION_STATUS.md @@ -1,6 +1,6 @@ # IMPLEMENTATION_STATUS -更新时间：2026-04-19 02:43 CST +更新时间：2026-04-19 02:51 CST ## 当前真实状态 @@ -12,30 +12,41 @@ 本轮最重要的新事实： -- item 级统一当前不能继续直接推进 -- 根本原因已经查清： - - 大陆逐条结果目前没有进入中央 +- 中央“逐条事件接收与落库”能力已经补齐 +- 当前真正剩余阻塞变成： + - 大陆节点还没把这版代码跑起来并把逐条事件推到中央 ## 本轮结论收口 -本轮没有进入新实现扩展，而是完成了 `I1` 的证据闭环： +本轮没有发散做新模块，而是完成了 `J1` 的最小实现： -### 1. `detect_result_projection` 当前只有聚合信息 +### 1. `detect_result_projection` 已扩展为可带逐条事件 -当前 `projection.keys()` 为： +当前新增： -- `job` -- `latest_event` -- `queue` -- `phase` +- `recent_domain_events` -没有逐条结果： +该字段只承载最小逐条事件： -- 没有 `domain_id` -- 没有 `job_item_id` -- 没有终态域名列表 +- `domain_started` +- `domain_completed` +- `domain_failed` +- `domain_blacklisted` -### 2. 中央 `detect_run_events` 逐条结果只有海外节点 +### 2. 中央 ingest 已具备把投影事件写入 `detect_run_events` 的能力 + +当前中央在 ingest `detect_result_projection` 时会： + +- 提取 `recent_domain_events` +- 以当前中央活跃任务作为展示锚点写入 `detect_run_events` +- 在 payload 中保留： + - `import_source_region` + - `import_source_record_id` + - `import_source_job_id` + - `import_source_job_code` + - `import_fingerprint` + +### 3. 真实端到端仍待大陆节点拉新验证 当前中央实际可见： @@ -43,19 +54,17 @@ - `domain_started` - `domain_completed` -当前中央不可见： +当前中央仍不可见： - `mainland-controller-01` - 无 `domain_*` - `mainland-worker-01` - 无 `domain_*` -### 3. 因此当前还做不了大陆 item 级回写 +说明中央代码已补，但真实输入流还没跑到新版本： -这不是简单 SQL 没补的问题，而是输入流还不存在： - -- 中央不知道大陆到底是哪个域名完成了 -- 中央只知道大陆总体“参与了、跑了多少、当前多少 running” +- 中央还没收到带 `recent_domain_events` 的 mainland 投影 +- 所以真实库里还没有新的 mainland `domain_*` ## 当前已闭合的问题 @@ -65,18 +74,18 @@ - Detect 页面误判“大陆没跑” - Detect 主统计口径不一致 - Runtime / Queue / Detect 主摘要不统一 -- item 级统一为什么做不了，这个问题现在已经查清 +- “中央如何接并落大陆逐条事件”的代码缺口 ## 当前剩余问题 当前唯一剩余问题： -- 大陆逐条结果缺少进入中央的最小同步通道 +- 大陆节点尚未完成“拉新代码并继续推投影”的最后验证 换句话说： -- 当前不是“不会回写” -- 而是“没有可回写的逐条输入” +- 当前不是“中央不会接” +- 而是“大陆还没把新格式投影推上来” ## 当前是否可以继续跑检测测试 @@ -84,7 +93,7 @@ - 可以继续跑检测测试 - 主统计和节点参与已经足够可信 -- 只是不要把 `detect_job_items` 当作大陆执行结果的逐条真相来源 +- 逐条账本是否完整，取决于大陆节点是否已经拉到这版代码 ## 当前是否建议直接上线 @@ -98,7 +107,7 @@ - 如果目标是： - 中央账本逐条反映大陆每个域名的终态 - 那还差最后一个前置批次 + 那现在只差最后一个落地验证动作 ## 当前优先级判断 @@ -116,7 +125,7 @@ ## 完成下一轮后的预期 -如果下一轮把大陆逐条结果输入中央这件事补上： +如果下一轮完成大陆拉新并验证贯通： - item 级统一才会真正可落地 - 后续 `detect_job_items` 回写会变成纯工程实现问题 diff --git a/docs/ops_center_runtime/TASK_BOARD.md b/docs/ops_center_runtime/TASK_BOARD.md index 1589f00..c5a0585 100644 --- a/docs/ops_center_runtime/TASK_BOARD.md +++ b/docs/ops_center_runtime/TASK_BOARD.md @@ -1,6 +1,6 @@ # TASK_BOARD -更新时间：2026-04-19 02:43 CST +更新时间：2026-04-19 02:51 CST ## 当前主批次 @@ -11,35 +11,30 @@ - 不进入新页面 - 不扩展控制面 - 只解决 item 级统一前的唯一前置缺口： - - 大陆节点当前没有把逐条检测结果同步回中央 + - 大陆节点逐条检测事件进入中央 ## 本轮已完成 -本轮完成的是结论确认，不是新功能扩展： +本轮完成的是最小输入通道补齐： -- 已核查 `detect_result_projection` -- 已核查 `detect_run_events` -- 已确认当前 item 级回写为什么做不了 +- 已让 `detect_result_projection` 开始携带 `recent_domain_events` +- 已让中央 `detect_result_ingest` 在接收时尝试落库 `detect_run_events` +- 已补幂等去重，避免重复投影重复写事件 +- 已完成最小单测与中央 API 重启 ## 关键证据 -### 证据 1：`detect_result_projection` 只有聚合数据 +### 证据 1：中央接收能力已经补上 -最新 `detect_result_projection` 当前只包含： +当前代码状态： -- `job` -- `latest_event` -- `queue` -- `phase` +- 大陆侧 `detect_result_projection` 将开始带： + - `recent_domain_events` +- 中央视角在 ingest `detect_result_projection` 时会： + - 提取 `domain_started/domain_completed/domain_failed/domain_blacklisted` + - 以幂等方式写入中央 `detect_run_events` -不包含： - -- `domain_id` -- `job_item_id` -- `domain` -- `status` 的逐条结果列表 - -### 证据 2：中央 `detect_run_events` 的逐条事件只有海外 +### 证据 2：当前真实库里大陆逐条事件仍未出现 当前中央库里可见： @@ -47,31 +42,27 @@ - `domain_started` - `domain_completed` -当前不可见： +当前仍不可见： - `mainland-controller-01` - 无逐条 `domain_*` 事件 - `mainland-worker-01` - 无逐条 `domain_*` 事件 -### 证据 3：大陆当前只同步了“聚合参与态”，没有同步“逐条结果” +### 证据 3：当前剩余阻塞已经收缩为“大陆节点尚未跑到这版代码” -所以当前能做到： +所以现在的状态是： -- 主统计统一 -- 节点参与统一 - -当前还做不到： - -- 把大陆执行结果逐条映射回 `detect_job_items` +- 中央接收与落库逻辑已具备 +- 但大陆节点必须拉到最新代码并继续推投影 +- 拉新前，中央库里当然还看不到新的 mainland `domain_*` ## 当前结论 -`I1-检测结果逐条回写设计批` 的结论已经明确： +`J1-大陆逐条结果同步前置批` 已进入实现态，不再只是结论确认： -- 现在还不能直接做 item 级回写 -- 不是逻辑没写 -- 是中央当前缺少大陆逐条结果输入 +- 中央缺口已补 +- 现在转入“大陆节点拉新并验证贯通” ## 下一步唯一主批次 @@ -84,12 +75,8 @@ 本批次唯一目标： -- 让大陆节点把至少这些字段同步回来： - - `job_code` - - `cycle_token` - - `domain_id` 或可稳定映射的 `domain` - - `result_status` - - `finished_at` +- 让大陆节点把 `recent_domain_events` 真正同步到中央 +- 并在中央确认出现 mainland `domain_*` 事件 ## 候选批次 @@ -137,4 +124,6 @@ 下一轮唯一应该继续做的事情： -- 设计并落地“大陆逐条检测结果 -> 中央可消费事件/投影”的最小同步通道 +- 在大陆节点拉最新代码后，验证： + - `detect_result_projection` 已含 `recent_domain_events` + - 中央 `detect_run_events` 已出现 mainland `domain_*` diff --git a/domain-api/app/services/sync_push_service.py b/domain-api/app/services/sync_push_service.py index 6dfe74a..d04e19c 100644 --- a/domain-api/app/services/sync_push_service.py +++ b/domain-api/app/services/sync_push_service.py @@ -1,5 +1,6 @@ from __future__ import annotations +import hashlib import json import socket import urllib.error @@ -14,6 +15,14 @@ from app.services.cluster_runtime_service import cleanup_imported_runtime_nodes, from app.services.sync_record_service import _decode_json, _normalize_region +_DETECT_RESULT_EVENT_TYPES = { + "domain_started", + "domain_completed", + "domain_failed", + "domain_blacklisted", +} + + def _format_time(value: datetime | None) -> str: return value.isoformat(sep=" ", timespec="seconds") if value else "" @@ -664,6 +673,173 @@ def _update_push_attempt(record_id: int, *, status: str, payload: dict | None = conn.commit() +def _build_detect_result_import_fingerprint( + *, + source_region: str, + source_record_id: int, + source_job_code: str, + event: dict, +) -> str: + payload = _decode_json(event.get("payload")) + seed = { + "source_region": str(source_region or "").strip(), + "source_record_id": int(source_record_id or 0), + "source_job_code": str(source_job_code or "").strip(), + "node_code": str(event.get("node_code") or "").strip(), + "event_type": str(event.get("event_type") or "").strip(), + "message": str(event.get("message") or "").strip(), + "created_at": str(event.get("created_at") or "").strip(), + "domain": str(payload.get("domain") or "").strip(), + "domain_id": str(payload.get("domain_id") or "").strip(), + "status": str(payload.get("status") or "").strip(), + "cycle_token": str(payload.get("cycle_token") or "").strip(), + } + return hashlib.sha1(json.dumps(seed, ensure_ascii=False, sort_keys=True).encode("utf-8")).hexdigest() + + +def _extract_detect_result_projection_events( + *, + source_region: str, + source_record_id: int, + projection: dict, +) -> list[dict]: + source_job = projection.get("job") or {} + source_job_code = str(source_job.get("job_code") or "").strip() + source_job_id = int(source_job.get("job_id") or 0) + events: list[dict] = [] + for raw_event in list(projection.get("recent_domain_events") or []): + event_type = str(raw_event.get("event_type") or "").strip() + if event_type not in _DETECT_RESULT_EVENT_TYPES: + continue + payload = _decode_json(raw_event.get("payload")) + fingerprint = _build_detect_result_import_fingerprint( + source_region=source_region, + source_record_id=source_record_id, + source_job_code=source_job_code, + event=raw_event, + ) + payload.update( + { + "imported_from_projection": True, + "import_source_region": source_region, + "import_source_record_id": int(source_record_id or 0), + "import_source_job_id": source_job_id, + "import_source_job_code": source_job_code, + "import_fingerprint": fingerprint, + } + ) + events.append( + { + "node_code": str(raw_event.get("node_code") or "").strip(), + "event_type": event_type, + "level": str(raw_event.get("level") or "info").strip() or "info", + "message": str(raw_event.get("message") or "").strip(), + "created_at": str(raw_event.get("created_at") or "").strip(), + "payload": payload, + "fingerprint": fingerprint, + } + ) + return events + + +def _resolve_detect_result_target_job_id() -> int: + from app.services.detect_job_service import get_active_detect_job_summary + + active_job = get_active_detect_job_summary(event_limit=1) or {} + return int(active_job.get("job_id") or 0) + + +def _parse_event_created_at(value: str) -> datetime | None: + text = str(value or "").strip() + if not text: + return None + try: + return datetime.fromisoformat(text) + except Exception: + return None + + +def _import_detect_result_projection_events( + *, + source_region: str, + source_record_id: int, + projection: dict, +) -> dict: + events = _extract_detect_result_projection_events( + source_region=source_region, + source_record_id=source_record_id, + projection=projection, + ) + if not events: + return {"imported_count": 0, "deduplicated_count": 0, "target_job_id": 0} + + target_job_id = _resolve_detect_result_target_job_id() + if target_job_id <= 0: + return {"imported_count": 0, "deduplicated_count": 0, "target_job_id": 0} + + imported_count = 0 + deduplicated_count = 0 + with get_db() as conn: + with conn.cursor() as cur: + for event in events: + cur.execute( + """ + SELECT id + FROM detect_run_events + WHERE job_id = %s + AND (payload_json->>'import_fingerprint') = %s + ORDER BY id DESC + LIMIT 1 + """, + (target_job_id, event["fingerprint"]), + ) + if cur.fetchone(): + deduplicated_count += 1 + continue + + created_at = _parse_event_created_at(event.get("created_at", "")) + if created_at: + cur.execute( + """ + INSERT INTO detect_run_events ( + job_id, node_code, event_type, level, message, payload_json, created_at + ) VALUES (%s, %s, %s, %s, %s, %s::jsonb, %s) + """, + ( + target_job_id, + event["node_code"], + event["event_type"], + event["level"], + event["message"], + json.dumps(event["payload"], ensure_ascii=False), + created_at, + ), + ) + else: + cur.execute( + """ + INSERT INTO detect_run_events ( + job_id, node_code, event_type, level, message, payload_json + ) VALUES (%s, %s, %s, %s, %s, %s::jsonb) + """, + ( + target_job_id, + event["node_code"], + event["event_type"], + event["level"], + event["message"], + json.dumps(event["payload"], ensure_ascii=False), + ), + ) + imported_count += 1 + conn.commit() + return { + "imported_count": imported_count, + "deduplicated_count": deduplicated_count, + "target_job_id": target_job_id, + } + + def ingest_runtime_projection(payload: dict, *, shared_token: str | None = None) -> tuple[bool, str, dict]: configured_token = str(settings.sync_shared_token or "").strip() incoming_token = str(shared_token or "").strip() @@ -699,6 +875,14 @@ def ingest_runtime_projection(payload: dict, *, shared_token: str | None = None) if existing: if sync_type == "runtime_projection": _refresh_remote_runtime_node(source_region=source_region, projection=projection, received_at=received_at) + if sync_type == "detect_result_projection": + import_result = _import_detect_result_projection_events( + source_region=source_region, + source_record_id=source_record_id, + projection=projection, + ) + else: + import_result = {} cur.execute( """ UPDATE detect_sync_records @@ -708,7 +892,11 @@ def ingest_runtime_projection(payload: dict, *, shared_token: str | None = None) (int(existing[0]),), ) conn.commit() - return True, "同步投影已存在，已按幂等处理", {"record_id": int(existing[0]), "deduplicated": True} + return True, "同步投影已存在，已按幂等处理", { + "record_id": int(existing[0]), + "deduplicated": True, + "event_import": import_result, + } stored_payload = { "sync_type": sync_type, @@ -737,7 +925,19 @@ def ingest_runtime_projection(payload: dict, *, shared_token: str | None = None) conn.commit() if sync_type == "runtime_projection": _refresh_remote_runtime_node(source_region=source_region, projection=projection, received_at=received_at) - return True, "同步投影接收成功", {"record_id": record_id, "deduplicated": False} + if sync_type == "detect_result_projection": + import_result = _import_detect_result_projection_events( + source_region=source_region, + source_record_id=source_record_id, + projection=projection, + ) + else: + import_result = {} + return True, "同步投影接收成功", { + "record_id": record_id, + "deduplicated": False, + "event_import": import_result, + } def _push_projection_now(sync_type: str, ingest_url: str) -> tuple[bool, str, dict]: diff --git a/domain-api/app/services/sync_record_service.py b/domain-api/app/services/sync_record_service.py index aa75c60..836ca7d 100644 --- a/domain-api/app/services/sync_record_service.py +++ b/domain-api/app/services/sync_record_service.py @@ -38,6 +38,37 @@ def _normalize_region(value: str | None, fallback: str) -> str: return text +_DETECT_RESULT_EVENT_TYPES = { + "domain_started", + "domain_completed", + "domain_failed", + "domain_blacklisted", +} + + +def _collect_recent_domain_events(active_job: dict, limit: int = 30) -> list[dict]: + events = list(active_job.get("current_cycle_events") or active_job.get("recent_events") or []) + normalized: list[dict] = [] + for event in reversed(events): + event_type = str(event.get("event_type") or "").strip() + if event_type not in _DETECT_RESULT_EVENT_TYPES: + continue + payload = _decode_json(event.get("payload")) + normalized.append( + { + "node_code": str(event.get("node_code") or "").strip(), + "event_type": event_type, + "level": str(event.get("level") or "info").strip() or "info", + "message": str(event.get("message") or "").strip(), + "created_at": str(event.get("created_at") or "").strip(), + "payload": payload, + } + ) + if limit <= 0: + return normalized + return normalized[-int(limit):] + + def _build_detect_result_batch_digest(batch: dict | None) -> dict: batch = batch or {} projection = (batch.get("projection") or {}).get("payload") or {} @@ -608,6 +639,7 @@ def append_detect_result_projection_if_changed( "label": detect.get("phase_label", ""), "detail": detect.get("phase_detail", ""), }, + "recent_domain_events": _collect_recent_domain_events(active_job, limit=30), } payload = { "projection": projection, diff --git a/domain-api/tests/test_sync_push_service.py b/domain-api/tests/test_sync_push_service.py new file mode 100644 index 0000000..d1b0ae8 --- /dev/null +++ b/domain-api/tests/test_sync_push_service.py @@ -0,0 +1,147 @@ +import unittest +from unittest.mock import patch + +from app.services.sync_push_service import ( + _extract_detect_result_projection_events, + ingest_runtime_projection, +) + + +class _FakeCursor: + def __init__(self, rows): + self._rows = list(rows) + self.executed = [] + + def execute(self, sql, params=None): + self.executed.append((sql, params)) + + def fetchone(self): + if self._rows: + return self._rows.pop(0) + return None + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + +class _FakeConnection: + def __init__(self, rows): + self.cursor_obj = _FakeCursor(rows) + self.committed = False + + def cursor(self): + return self.cursor_obj + + def commit(self): + self.committed = True + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + +class SyncPushServiceTests(unittest.TestCase): + def test_extract_detect_result_projection_events_adds_import_metadata(self) -> None: + projection = { + "job": { + "job_id": 12, + "job_code": "detect-local-001", + }, + "recent_domain_events": [ + { + "node_code": "mainland-worker-01", + "event_type": "domain_completed", + "level": "info", + "message": "完成 a.com", + "created_at": "2026-04-19 01:05:00", + "payload": {"domain": "a.com", "domain_id": 101, "cycle_token": "cycle-1"}, + } + ], + } + + events = _extract_detect_result_projection_events( + source_region="mainland", + source_record_id=88, + projection=projection, + ) + + self.assertEqual(1, len(events)) + event = events[0] + self.assertEqual("domain_completed", event["event_type"]) + self.assertEqual("mainland", event["payload"]["import_source_region"]) + self.assertEqual(88, event["payload"]["import_source_record_id"]) + self.assertEqual("detect-local-001", event["payload"]["import_source_job_code"]) + self.assertTrue(event["payload"]["imported_from_projection"]) + self.assertTrue(event["payload"]["import_fingerprint"]) + + @patch("app.services.sync_push_service._import_detect_result_projection_events") + @patch("app.services.sync_push_service.get_db") + def test_ingest_runtime_projection_imports_detect_result_events_on_new_record( + self, + mock_get_db, + mock_import_events, + ) -> None: + fake_conn = _FakeConnection(rows=[None, (321,)]) + mock_get_db.return_value = fake_conn + mock_import_events.return_value = {"imported_count": 2, "deduplicated_count": 0, "target_job_id": 9} + + ok, message, data = ingest_runtime_projection( + { + "sync_type": "detect_result_projection", + "source_region": "mainland", + "source_record_id": 66, + "projection_hash": "hash-001", + "projection": {"recent_domain_events": []}, + } + ) + + self.assertTrue(ok) + self.assertEqual("同步投影接收成功", message) + self.assertEqual(321, data["record_id"]) + self.assertEqual(2, data["event_import"]["imported_count"]) + mock_import_events.assert_called_once_with( + source_region="mainland", + source_record_id=66, + projection={"recent_domain_events": []}, + ) + self.assertTrue(fake_conn.committed) + + @patch("app.services.sync_push_service._import_detect_result_projection_events") + @patch("app.services.sync_push_service.get_db") + def test_ingest_runtime_projection_imports_detect_result_events_on_deduplicated_record( + self, + mock_get_db, + mock_import_events, + ) -> None: + fake_conn = _FakeConnection(rows=[(777,)]) + mock_get_db.return_value = fake_conn + mock_import_events.return_value = {"imported_count": 0, "deduplicated_count": 3, "target_job_id": 9} + + ok, message, data = ingest_runtime_projection( + { + "sync_type": "detect_result_projection", + "source_region": "mainland", + "source_record_id": 66, + "projection_hash": "hash-001", + "projection": {"recent_domain_events": []}, + } + ) + + self.assertTrue(ok) + self.assertEqual("同步投影已存在，已按幂等处理", message) + self.assertTrue(data["deduplicated"]) + self.assertEqual(3, data["event_import"]["deduplicated_count"]) + mock_import_events.assert_called_once_with( + source_region="mainland", + source_record_id=66, + projection={"recent_domain_events": []}, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/domain-api/tests/test_sync_record_service.py b/domain-api/tests/test_sync_record_service.py new file mode 100644 index 0000000..c1dad47 --- /dev/null +++ b/domain-api/tests/test_sync_record_service.py @@ -0,0 +1,55 @@ +import unittest + +from app.services.sync_record_service import _collect_recent_domain_events + + +class SyncRecordServiceTests(unittest.TestCase): + def test_collect_recent_domain_events_filters_and_keeps_latest_slice(self) -> None: + active_job = { + "current_cycle_events": [ + { + "node_code": "mainland-worker-01", + "event_type": "job_dispatch_sent", + "level": "info", + "message": "dispatch", + "created_at": "2026-04-19 01:00:05", + "payload": {"cycle_token": "cycle-1"}, + }, + { + "node_code": "mainland-worker-01", + "event_type": "domain_completed", + "level": "info", + "message": "完成 a.com", + "created_at": "2026-04-19 01:00:04", + "payload": {"domain": "a.com", "cycle_token": "cycle-1"}, + }, + { + "node_code": "mainland-worker-01", + "event_type": "domain_started", + "level": "info", + "message": "开始 b.com", + "created_at": "2026-04-19 01:00:03", + "payload": {"domain": "b.com", "cycle_token": "cycle-1"}, + }, + { + "node_code": "mainland-worker-01", + "event_type": "domain_failed", + "level": "warning", + "message": "失败 c.com", + "created_at": "2026-04-19 01:00:02", + "payload": {"domain": "c.com", "cycle_token": "cycle-1"}, + }, + ] + } + + events = _collect_recent_domain_events(active_job, limit=2) + + self.assertEqual(2, len(events)) + self.assertEqual("domain_started", events[0]["event_type"]) + self.assertEqual("b.com", events[0]["payload"]["domain"]) + self.assertEqual("domain_completed", events[1]["event_type"]) + self.assertEqual("a.com", events[1]["payload"]["domain"]) + + +if __name__ == "__main__": + unittest.main()