From 6c4e995f84ac934ea490291ee5851e099582873c Mon Sep 17 00:00:00 2001 From: Your Name Date: Sun, 19 Apr 2026 02:41:45 +0800 Subject: [PATCH] fix: unify detect effective job metrics --- .../HANDOFF_20260419_0240.md | 127 +++++++++++++++ .../IMPLEMENTATION_STATUS.md | 107 ++++++------ docs/ops_center_runtime/TASK_BOARD.md | 153 +++++++++--------- domain-api/app/services/detect_job_service.py | 145 +++++++++++++++-- domain-api/tests/test_detect_job_service.py | 122 ++++++++++++++ 5 files changed, 520 insertions(+), 134 deletions(-) create mode 100644 docs/ops_center_runtime/HANDOFF_20260419_0240.md create mode 100644 domain-api/tests/test_detect_job_service.py diff --git a/docs/ops_center_runtime/HANDOFF_20260419_0240.md b/docs/ops_center_runtime/HANDOFF_20260419_0240.md new file mode 100644 index 0000000..e78543c --- /dev/null +++ b/docs/ops_center_runtime/HANDOFF_20260419_0240.md @@ -0,0 +1,127 @@ +# HANDOFF 2026-04-19 02:40 + +## 本轮完成了什么 + +本轮只处理了一个问题： + +- Detect 主统计还是“中央原值 + display 附加值”双轨，Runtime / Detect 不够统一 + +实际完成： + +1. 新增有效账本聚合逻辑 +2. `active_job` 主字段切到统一后的有效值 +3. 保留 `raw_*` 原始中央账本字段 +4. 新增针对统一规则的后端单测 +5. 重启 `domaincheck-api` +6. 重新验证线上接口 + +## 本轮关键结果 + +`detect/job/active` 现在主字段直接返回： + +- `items_total = 1000` +- `items_pending = 931` +- `items_claimed = 34` +- `items_running = 14` +- `items_completed = 21` + +同时保留原始中央账本： + +- `raw_items_pending = 950` +- `raw_items_claimed = 25` +- `raw_items_running = 4` + +`node_stats` / `distributed_node_stats` 当前已统一为： + +- `mainland-controller-01` +- `mainland-worker-01` +- `overseas-control-01` +- `unassigned = 680` + +`detect/queue-summary` 当前主字段： + +- `pending = 931` +- `claimed = 34` +- `running = 14` + +## 当前最高优先级 + +唯一最高优先级： + +- 决定是否继续进入 item 级结果回写统一 + +原因： + +- 当前主统计已经基本统一 +- 剩下真正没闭环的是 `detect_result_projection -> detect_job_items` + +## 当前状态结论 + +当前可以明确说： + +- 检测能跑 +- 大陆节点确实参与了 +- Detect 页面和主统计都更接近真实执行态 +- Runtime / Queue / Detect 主摘要开始吃同一套有效账本 + +## 当前唯一剩余问题 + +唯一剩余问题： + +- 目前还是“聚合统一” +- 还不是“逐条 item 最终一致” + +所以： + +- 主观感受和主统计已经收口 +- 底层逐条账本还没完全收口 + +## 下一轮如果继续，只做什么 + +下一轮只做这一件事： + +- 查清是否要把 `detect_result_projection` 最小映射回 `detect_job_items` + +优先原则： + +- 不发散 +- 不扩新页面 +- 不碰发布动作 + +## 推荐模型与推理等级 + +继续推荐： + +- `GPT-5.4` +- `high` + +原因： + +- 下一轮是 item 级账本映射策略 +- 仍然是高复杂度，但还没到必须切超高 + +## 给下一轮的任务卡 + +```text +任务名：I1-detect_result_projection 到 detect_job_items 的最小映射设计批 + +目标： +1. 查清当前 detect_result_projection 能提供哪些聚合字段 +2. 判断是否足以做 item 级状态回写，还是只能做批次级回写 +3. 只做最小一致性设计，不重做执行链 + +范围限制： +1. 不新增页面 +2. 不扩控制面功能 +3. 不进入发布动作 +4. 不并行做别的方向 + +完成判定： +1. 给出 item 级统一是否可行的结论 +2. 如果可行，明确最小落点 +3. 如果暂不可行，明确缺的字段和最小补数方案 +``` + +## 一句话结论 + +这轮已经把 Detect 主统计从“靠 display 补丁看懂”推进到“主字段直接统一可读”；下一轮真正该做的，只剩下是否继续把结果同步推进到 item 级逐条账本。 diff --git a/docs/ops_center_runtime/IMPLEMENTATION_STATUS.md b/docs/ops_center_runtime/IMPLEMENTATION_STATUS.md index 86e4f3e..e3639f6 100644 --- a/docs/ops_center_runtime/IMPLEMENTATION_STATUS.md +++ b/docs/ops_center_runtime/IMPLEMENTATION_STATUS.md @@ -1,20 +1,20 @@ # IMPLEMENTATION_STATUS -更新时间：2026-04-19 02:33 CST +更新时间：2026-04-19 02:40 CST ## 当前真实状态 阶段判断： - 海外单脑接管能力：约 `95%` -- 分布式检测真实执行能力：约 `93%` -- 距离“可稳定上线并放心用后台发起检测”：约 `92%` +- 分布式检测真实执行能力：约 `94%` +- 距离“可稳定上线并放心用后台发起检测”：约 `93%~94%` 本轮最重要的新事实： -- Detect 页面已经不再只显示海外节点 -- 大陆 controller / worker 已进入 Detect 总览显示口径 -- 用户现在能从 Detect 页面直接看见大陆节点参与执行 +- Detect 主统计口径已经从“双轨显示”推进到“统一后的有效账本” +- `detect/job/active` 和 `detect/queue-summary` 主字段已直接体现大陆 + 海外的统一聚合结果 +- 原始中央账本仍完整保留，便于继续排障和审计 ## 本轮代码收口 @@ -22,48 +22,64 @@ - `domain-api/app/services/detect_job_service.py` -本轮前端变更： +本轮测试变更： -- `domain-web/src/views/detect/DetectView.vue` +- `domain-api/tests/test_detect_job_service.py` 本轮修复方式： -- 保留中央账本原值，不硬改 `detect_job_items` -- 额外补出分布式显示口径： - - `distributed_node_stats` - - `display_items_claimed` - - `display_items_running` -- Detect 页面优先显示这个分布式显示口径 +- 保留中央 `detect_job_items` 原始值 +- 引入“有效统一账本”： + - `node_stats` + - `items_pending` + - `items_claimed` + - `items_running` + - `items_completed` + - `items_failed` + - `items_terminal` +- 同时新增 `raw_*` 字段保留中央原始账本 ## 关键验证证据 接口验证： - `detect/job/active` - - `display_items_claimed = 34` - - `display_items_running = 14` - - `distributed_node_stats` 包含： + - 已统一为： + - `items_total = 1000` + - `items_pending = 931` + - `items_claimed = 34` + - `items_running = 14` + - `items_completed = 21` + - 原始账本仍可追溯： + - `raw_items_pending = 950` + - `raw_items_claimed = 25` + - `raw_items_running = 4` + - `node_stats` 已包含： - `mainland-controller-01` - `mainland-worker-01` - `overseas-control-01` + - `unassigned` - `detect/queue-summary` - - `queue.display_claimed = 34` - - `queue.display_running = 14` - - `nodes` 已含大陆 controller / worker + - 已统一为： + - `pending = 931` + - `claimed = 34` + - `running = 14` 编译验证： -- 后端： - - `python3 -m py_compile domain-api/app/services/detect_job_service.py` +- `python3 -m py_compile domain-api/app/services/detect_job_service.py domain-api/tests/test_detect_job_service.py` - 通过 -- 前端： - - `cd domain-web && npm run build` + +单测验证： + +- `/opt/domaincheck/domainCheck/.venv/bin/python -m unittest tests.test_detect_job_service` - 通过 运行验证： -- `domaincheck-api` 已重启成功 -- 活跃接口返回新字段正常 +- `domaincheck-api` + - 已重启成功 +- 线上接口返回统一账本字段正常 ## 当前已闭合的问题 @@ -72,46 +88,43 @@ - 大陆 controller 无法 `pull_tasks` - Detect 页面误判“大陆没跑” - Detect 节点表格只显示海外与 `unassigned` +- Detect 主统计仍依赖 display 附加字段才能看懂 ## 当前剩余问题 当前唯一剩余问题： -- 中央账本与分布式显示口径还不是同一套最终账本 +- 还没有做到 item 级逐条统一回写 具体来说： -- `detect_job_items` - - 仍是中央派发账本 -- `distributed_node_stats` - - 是运行态 / 投影视角 - -所以当前页面已经“可观察、可判断”了，但底层仍属于双账本阶段。 +- 当前已做到： + - 主统计口径一致 + - 节点参与口径一致 + - Runtime / Detect 读取同一套有效聚合值 +- 当前未做到： + - 大陆执行结果逐条回写 `detect_job_items` ## 当前是否可以继续跑检测测试 当前结论： - 可以继续跑检测任务测试 -- 而且现在 Detect 页面已经能看出大陆节点参与 +- 而且现在 Detect 主统计和节点参与都更接近真实执行态 ## 当前是否建议直接上线 当前结论： -- 比上一轮更接近上线 -- 但还不建议把“检测调度账本”宣称为最终完全统一 - -原因： - -- 现在解决的是“显示正确性” -- 下一步要解决的是“账本统一性” +- 已经比上一轮更接近可上线状态 +- 如果目标是“后台可以稳定发起检测并看懂谁在跑”，已经基本可用 +- 如果目标是“中央账本逐条结果完全统一”，还差最后一层 item 级回写 ## 当前优先级判断 最高优先级： -- 设计并收敛中央账本与分布式执行账本的统一规则 +- 决定是否继续进入 `detect_result_projection -> detect_job_items` 的 item 级统一 当前不应继续推进： @@ -119,12 +132,12 @@ - 新模块 - 新页面 - 发布动作 -- 与账本统一无关的工作 +- 与检测账本统一无关的工作 ## 完成下一轮后的预期 -如果下一轮把账本统一策略明确并落地： +如果下一轮把 item 级结果回写规则明确并最小落地： -- Detect 页面将不仅“看起来对” -- 而且底层统计也会真正一致 -- 整体可上线程度预计提升到 `94%~96%` +- Detect 不仅主统计一致 +- 而且底层逐条账本也会接近最终一致 +- 整体可上线程度预计提升到 `95%~96%` diff --git a/docs/ops_center_runtime/TASK_BOARD.md b/docs/ops_center_runtime/TASK_BOARD.md index 5f315da..ef14a83 100644 --- a/docs/ops_center_runtime/TASK_BOARD.md +++ b/docs/ops_center_runtime/TASK_BOARD.md @@ -1,131 +1,138 @@ # TASK_BOARD -更新时间：2026-04-19 02:33 CST +更新时间：2026-04-19 02:40 CST ## 当前主批次 -唯一主批次：`G1-检测总览口径统一批` +唯一主批次：`H1-检测账本统一最小落地批` 目标： - 不进入新功能实现 - 不扩展控制面 -- 只收口 Detect 页面“大陆明明在跑，但页面像没跑”的口径问题 +- 只把 Detect 主账本口径从“双轨显示”收敛到“统一后的有效口径” ## 本轮已完成 -本轮完成了最小可见修复： +本轮完成了最小统一落地： -- `detect/job/active` - - 新增： - - `distributed_node_stats` - - `display_items_claimed` - - `display_items_running` - - `display_items_completed` - - `display_items_failed` - - `display_active_node_codes` -- `detect/queue-summary` - - 新增： - - `queue.display_claimed` - - `queue.display_running` - - `nodes` 已切到分布式显示口径 -- `DetectView.vue` - - 已优先显示分布式口径 - - 节点表格已优先显示 `distributed_node_stats` +- `detect_job_service.py` + - 新增统一账本聚合逻辑： + - `_build_effective_node_stats` + - `_build_effective_summary` + - `active_job` 主字段已切换为统一后的有效值： + - `items_pending` + - `items_claimed` + - `items_running` + - `items_completed` + - `items_failed` + - `items_terminal` + - `node_stats` +- 保留中央原始账本供追溯： + - `raw_items_total` + - `raw_items_pending` + - `raw_items_claimed` + - `raw_items_running` + - `raw_items_completed` + - `raw_items_blacklisted` + - `raw_items_failed` + - `raw_items_terminal` + - `raw_node_stats` +- 新增单测： + - `domain-api/tests/test_detect_job_service.py` ## 当前结果快照 接口验证结果： - `detect/job/active` - - `display_items_claimed = 34` - - `display_items_running = 14` - - `distributed_node_stats` 已包含： + - 主字段已统一为： + - `items_total = 1000` + - `items_pending = 931` + - `items_claimed = 34` + - `items_running = 14` + - `items_completed = 21` + - 原始中央账本仍可追溯： + - `raw_items_pending = 950` + - `raw_items_claimed = 25` + - `raw_items_running = 4` + - `node_stats` / `distributed_node_stats` 已统一为： - `mainland-controller-01` - `mainland-worker-01` - `overseas-control-01` - - `unassigned` + - `unassigned = 680` - `detect/queue-summary` - - `queue.display_claimed = 34` - - `queue.display_running = 14` - - `nodes` 已包含： - - `mainland-controller-01` - - `mainland-worker-01` - - `overseas-control-01` - - `unassigned` + - 主字段已统一为： + - `pending = 931` + - `claimed = 34` + - `running = 14` -运行侧证据： +验证结果： -- `mainland-controller-01` - - `runtime.pull_tasks` - - `job_id = 87` - - `status = success` -- 大陆 controller / worker 仍在执行中 +- `python3 -m py_compile` + - 通过 +- `/opt/domaincheck/domainCheck/.venv/bin/python -m unittest tests.test_detect_job_service` + - 通过 +- `domaincheck-api` + - 已重启并验证通过 ## 当前结论 -这轮已经把“Detect 页面看不出大陆参与”这个核心观感问题修掉了一大半。 +这轮已经把“中央账本原值”和“分布式 display 附加值”进一步收敛成一套主展示账本。 -现在的 Detect 页面口径变成： +现在的口径变成： -- 中央原始账本仍保留： - - `items_total = 1000` - - `pending = 950` - - `claimed = 25` - - `running = 4` -- 页面展示口径新增分布式视角： - - `display_claimed = 34` - - `display_running = 14` - - 节点列表能直接看到大陆 controller / worker +- 主字段： + - 直接反映当前统一后的有效检测账本 +- 原始字段： + - 仍保留中央派发账本，供排障和审计 + +这意味着： + +- Detect 页面不再只是“看起来对” +- Runtime / Queue / Detect 主视图开始吃同一套有效统计 ## 当前唯一剩余问题 唯一剩余问题： -- 中央账本与分布式显示口径仍是“双轨” +- 这 still 是“聚合统一”，还不是“逐条 item 真正回写统一” -这不是 bug，而是当前架构现状： +也就是说： -- 中央 `detect_job_items` 统计的是海外控制面原始派发账本 -- 大陆拉批后的本地执行账本在 runtime / projection 侧 - -当前已经做到： - -- 页面不再误判“大陆没跑” - -但还没做到： - -- 把两套账本真正收敛成一个最终一致的统一任务账本 +- 当前已经解决： + - 主统计口径一致 + - 节点参与显示一致 +- 当前还没解决： + - 大陆执行结果逐条回写 `detect_job_items` ## 候选批次 -### Candidate H1 +### Candidate I1 -名称：检测账本统一设计批 +名称：检测结果逐条回写设计批 进入条件： -- 用户确认下一轮继续深入“中央账本与分布式执行账本统一” +- 用户确认要继续把聚合统一推进到 item 级一致 范围： -- 先设计统一账本规则 -- 不直接大改执行链 -- 明确 pull / ack / result sync 后中央账本应如何回写 +- 只设计 `detect_result_projection -> detect_job_items` 的最小映射方案 +- 不直接重做整条执行链 -### Candidate H2 +### Candidate I2 -名称：Detect 页面文案微收口批 +名称：Detect / Runtime 文案统一批 进入条件： -- 如果用户只想先把页面提示说清楚 +- 如果用户只想进一步把“raw vs effective”提示写清楚 范围： -- 在 Detect 页面明确标注： - - “中央账本视角” - - “分布式执行视角” +- 补统一文案 +- 不再扩逻辑 ## 暂停项 @@ -141,4 +148,4 @@ 下一轮唯一应该继续做的事情： -- 明确中央 `detect_job_items` 与大陆 runtime 投影之间的统一账本策略 +- 明确是否要继续进入 item 级结果回写统一 diff --git a/domain-api/app/services/detect_job_service.py b/domain-api/app/services/detect_job_service.py index 779fde1..594d81e 100644 --- a/domain-api/app/services/detect_job_service.py +++ b/domain-api/app/services/detect_job_service.py @@ -187,6 +187,96 @@ def _build_display_summary(node_stats: list[dict]) -> dict: } +def _normalize_node_bucket(item: dict) -> dict: + node_code = str(item.get("node_code") or "").strip() + return { + "node_code": node_code, + "items_total": _int_value(item.get("items_total")), + "items_pending": _int_value(item.get("items_pending")), + "items_claimed": _int_value(item.get("items_claimed")), + "items_running": _int_value(item.get("items_running")), + "items_completed": _int_value(item.get("items_completed")), + "items_blacklisted": _int_value(item.get("items_blacklisted")), + "items_failed": _int_value(item.get("items_failed")), + "metrics_source": str(item.get("metrics_source") or "").strip(), + "region": str(item.get("region") or "").strip(), + "role": str(item.get("role") or "").strip(), + "status": str(item.get("status") or "").strip(), + "current_load": _int_value(item.get("current_load")), + "last_heartbeat_at": str(item.get("last_heartbeat_at") or "").strip(), + } + + +def _build_effective_node_stats( + *, + distributed_node_stats: list[dict], + raw_items_total: int, +) -> list[dict]: + normalized_total = max(0, _int_value(raw_items_total)) + assigned_buckets: list[dict] = [] + assigned_total = 0 + for item in list(distributed_node_stats or []): + bucket = _normalize_node_bucket(item) + if not bucket["node_code"] or bucket["node_code"] == "unassigned": + continue + assigned_buckets.append(bucket) + assigned_total += bucket["items_total"] + + effective_total = max(normalized_total, assigned_total) + remainder_total = max(effective_total - assigned_total, 0) + if remainder_total > 0: + assigned_buckets.append( + { + "node_code": "unassigned", + "items_total": remainder_total, + "items_pending": remainder_total, + "items_claimed": 0, + "items_running": 0, + "items_completed": 0, + "items_blacklisted": 0, + "items_failed": 0, + "metrics_source": "central_queue", + "region": "", + "role": "", + "status": "", + "current_load": 0, + "last_heartbeat_at": "", + } + ) + + return sorted( + assigned_buckets, + key=lambda item: ( + str(item.get("node_code") or "") == "unassigned", + -_int_value(item.get("items_running")), + -_int_value(item.get("items_claimed")), + -_int_value(item.get("items_total")), + str(item.get("node_code") or ""), + ), + ) + + +def _build_effective_summary(*, node_stats: list[dict], raw_items_total: int, raw_items_blacklisted: int) -> dict: + effective_total = max(0, _int_value(raw_items_total)) + effective_blacklisted = max(0, _int_value(raw_items_blacklisted)) + effective_pending = sum(_int_value(item.get("items_pending")) for item in list(node_stats or [])) + effective_claimed = sum(_int_value(item.get("items_claimed")) for item in list(node_stats or [])) + effective_running = sum(_int_value(item.get("items_running")) for item in list(node_stats or [])) + effective_completed = sum(_int_value(item.get("items_completed")) for item in list(node_stats or [])) + effective_failed = sum(_int_value(item.get("items_failed")) for item in list(node_stats or [])) + effective_terminal = effective_completed + effective_blacklisted + effective_failed + return { + "items_total": effective_total, + "items_pending": max(effective_pending, 0), + "items_claimed": max(effective_claimed, 0), + "items_running": max(effective_running, 0), + "items_completed": max(effective_completed, 0), + "items_blacklisted": effective_blacklisted, + "items_failed": max(effective_failed, 0), + "items_terminal": max(effective_terminal, 0), + } + + def _extract_current_cycle_events(events: list[dict]) -> tuple[str, list[dict]]: if not events: return "", [] @@ -293,9 +383,24 @@ def _fetch_job_summary(cur, job_row, event_limit: int = 20) -> dict: local_node_stats=list(node_buckets.values()), runtime_node_rows=_load_runtime_display_rows(cur), ) - display_summary = _build_display_summary(distributed_node_stats) - total = sum(counts.values()) - terminal = int(counts.get("completed", 0)) + int(counts.get("blacklisted", 0)) + int(counts.get("failed", 0)) + raw_total = sum(counts.values()) + raw_pending = int(counts.get("pending", 0)) + raw_claimed = int(counts.get("claimed", 0)) + raw_running = int(counts.get("running", 0)) + raw_completed = int(counts.get("completed", 0)) + raw_blacklisted = int(counts.get("blacklisted", 0)) + raw_failed = int(counts.get("failed", 0)) + raw_terminal = raw_completed + raw_blacklisted + raw_failed + effective_node_stats = _build_effective_node_stats( + distributed_node_stats=distributed_node_stats, + raw_items_total=raw_total, + ) + effective_summary = _build_effective_summary( + node_stats=effective_node_stats, + raw_items_total=raw_total, + raw_items_blacklisted=raw_blacklisted, + ) + display_summary = _build_display_summary(effective_node_stats) return { "job_id": job_id, "job_code": job_row[1], @@ -305,17 +410,29 @@ def _fetch_job_summary(cur, job_row, event_limit: int = 20) -> dict: "created_at": _format_time(job_row[5]), "started_at": _format_time(job_row[6]), "finished_at": _format_time(job_row[7]), - "items_total": total, - "items_pending": int(counts.get("pending", 0)), - "items_claimed": int(counts.get("claimed", 0)), - "items_running": int(counts.get("running", 0)), - "items_completed": int(counts.get("completed", 0)), - "items_blacklisted": int(counts.get("blacklisted", 0)), - "items_failed": int(counts.get("failed", 0)), - "items_terminal": terminal, - "progress_percent": round((terminal / total) * 100, 2) if total else 0, - "node_stats": list(node_buckets.values()), - "distributed_node_stats": distributed_node_stats, + "items_total": int(effective_summary.get("items_total", 0) or 0), + "items_pending": int(effective_summary.get("items_pending", 0) or 0), + "items_claimed": int(effective_summary.get("items_claimed", 0) or 0), + "items_running": int(effective_summary.get("items_running", 0) or 0), + "items_completed": int(effective_summary.get("items_completed", 0) or 0), + "items_blacklisted": int(effective_summary.get("items_blacklisted", 0) or 0), + "items_failed": int(effective_summary.get("items_failed", 0) or 0), + "items_terminal": int(effective_summary.get("items_terminal", 0) or 0), + "progress_percent": round( + (int(effective_summary.get("items_terminal", 0) or 0) / int(effective_summary.get("items_total", 0) or 0)) * 100, + 2, + ) if int(effective_summary.get("items_total", 0) or 0) else 0, + "raw_items_total": raw_total, + "raw_items_pending": raw_pending, + "raw_items_claimed": raw_claimed, + "raw_items_running": raw_running, + "raw_items_completed": raw_completed, + "raw_items_blacklisted": raw_blacklisted, + "raw_items_failed": raw_failed, + "raw_items_terminal": raw_terminal, + "raw_node_stats": list(node_buckets.values()), + "node_stats": effective_node_stats, + "distributed_node_stats": effective_node_stats, "display_items_claimed": int(display_summary.get("items_claimed", 0) or 0), "display_items_running": int(display_summary.get("items_running", 0) or 0), "display_items_completed": int(display_summary.get("items_completed", 0) or 0), diff --git a/domain-api/tests/test_detect_job_service.py b/domain-api/tests/test_detect_job_service.py new file mode 100644 index 0000000..ab06d84 --- /dev/null +++ b/domain-api/tests/test_detect_job_service.py @@ -0,0 +1,122 @@ +import unittest + +from app.services.detect_job_service import _build_effective_node_stats, _build_effective_summary + + +class DetectJobServiceTests(unittest.TestCase): + def test_build_effective_node_stats_keeps_runtime_nodes_and_backfills_unassigned(self) -> None: + distributed_node_stats = [ + { + "node_code": "mainland-controller-01", + "items_total": 200, + "items_pending": 195, + "items_claimed": 0, + "items_running": 5, + "items_completed": 0, + "items_blacklisted": 0, + "items_failed": 0, + }, + { + "node_code": "mainland-worker-01", + "items_total": 70, + "items_pending": 56, + "items_claimed": 9, + "items_running": 5, + "items_completed": 0, + "items_blacklisted": 0, + "items_failed": 0, + }, + { + "node_code": "overseas-control-01", + "items_total": 50, + "items_pending": 0, + "items_claimed": 25, + "items_running": 4, + "items_completed": 21, + "items_blacklisted": 0, + "items_failed": 0, + }, + { + "node_code": "unassigned", + "items_total": 950, + "items_pending": 950, + "items_claimed": 0, + "items_running": 0, + "items_completed": 0, + "items_blacklisted": 0, + "items_failed": 0, + }, + ] + + node_stats = _build_effective_node_stats( + distributed_node_stats=distributed_node_stats, + raw_items_total=1000, + ) + + self.assertEqual(4, len(node_stats)) + unassigned = next(item for item in node_stats if item["node_code"] == "unassigned") + self.assertEqual(680, unassigned["items_total"]) + self.assertEqual(680, unassigned["items_pending"]) + self.assertEqual("central_queue", unassigned["metrics_source"]) + + def test_build_effective_summary_uses_effective_node_stats_and_raw_blacklisted(self) -> None: + node_stats = [ + { + "node_code": "mainland-controller-01", + "items_total": 200, + "items_pending": 195, + "items_claimed": 0, + "items_running": 5, + "items_completed": 0, + "items_blacklisted": 0, + "items_failed": 0, + }, + { + "node_code": "mainland-worker-01", + "items_total": 70, + "items_pending": 56, + "items_claimed": 9, + "items_running": 5, + "items_completed": 0, + "items_blacklisted": 0, + "items_failed": 0, + }, + { + "node_code": "overseas-control-01", + "items_total": 50, + "items_pending": 0, + "items_claimed": 25, + "items_running": 4, + "items_completed": 21, + "items_blacklisted": 0, + "items_failed": 0, + }, + { + "node_code": "unassigned", + "items_total": 680, + "items_pending": 680, + "items_claimed": 0, + "items_running": 0, + "items_completed": 0, + "items_blacklisted": 0, + "items_failed": 0, + }, + ] + + summary = _build_effective_summary( + node_stats=node_stats, + raw_items_total=1000, + raw_items_blacklisted=3, + ) + + self.assertEqual(1000, summary["items_total"]) + self.assertEqual(931, summary["items_pending"]) + self.assertEqual(34, summary["items_claimed"]) + self.assertEqual(14, summary["items_running"]) + self.assertEqual(21, summary["items_completed"]) + self.assertEqual(3, summary["items_blacklisted"]) + self.assertEqual(24, summary["items_terminal"]) + + +if __name__ == "__main__": + unittest.main()