agent-runner
a173fbaf76
- 新增 3.5 成本效益量化分析与表 5 结论-证据映射表（8 条结论可核验） - 结论 5.2 为五类受众补充执行周期、量化目标与证据关联 - 创建 notebooks/pilot_analysis.md（26 项数据一致性核验通过） - 创建 plots/ 三张图表 + CSV 数据 + 生成脚本 - 创建 processing/validate_data.py（27 项校验通过） - 更新 data/paper_loop/round_notes.md 与 sync_summary.md
182 lines
5.3 KiB
Python
182 lines
5.3 KiB
Python
#!/usr/bin/env python3
|
||
"""数据一致性校验脚本。
|
||
|
||
校验 notebooks/pilot_analysis.md 中的关键数据与 plots/*.csv 数据文件一致，
|
||
并验证论文 reports/paper_draft.md 中引用的核心数值与数据源吻合。
|
||
|
||
用法：
|
||
python3 processing/validate_data.py
|
||
"""
|
||
|
||
import csv
|
||
import os
|
||
import re
|
||
import sys
|
||
|
||
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
PLOTS = os.path.join(ROOT, "plots")
|
||
NOTEBOOK = os.path.join(ROOT, "notebooks", "pilot_analysis.md")
|
||
PAPER = os.path.join(ROOT, "reports", "paper_draft.md")
|
||
|
||
errors = []
|
||
checks = 0
|
||
|
||
|
||
def ok(msg):
|
||
global checks
|
||
checks += 1
|
||
print(f" PASS: {msg}")
|
||
|
||
|
||
def fail(msg):
|
||
global errors
|
||
errors.append(msg)
|
||
print(f" FAIL: {msg}")
|
||
|
||
|
||
def load_csv(name):
|
||
path = os.path.join(PLOTS, name)
|
||
with open(path, encoding="utf-8") as f:
|
||
return list(csv.DictReader(f))
|
||
|
||
|
||
def check_run_data():
|
||
print("[1] 校验运行数据 CSV")
|
||
rows = load_csv("pilot_run_data.csv")
|
||
cities = [r["city"] for r in rows]
|
||
if len(cities) == 8:
|
||
ok(f"地市数 = 8: {', '.join(cities)}")
|
||
else:
|
||
fail(f"地市数 = {len(cities)}, 期望 8")
|
||
|
||
producing = [r for r in rows if r["stage"] == "实际产出"]
|
||
if len(producing) == 2:
|
||
ok(f"实际产出地市 = 2: {', '.join(r['city'] for r in producing)}")
|
||
else:
|
||
fail(f"实际产出地市 = {len(producing)}, 期望 2")
|
||
|
||
huizhou = next(r for r in rows if r["city"] == "惠州")
|
||
if int(huizhou["alert_count"]) == 19800:
|
||
ok("惠州预警数据 = 19,800")
|
||
else:
|
||
fail(f"惠州预警数据 = {huizhou['alert_count']}, 期望 19800")
|
||
|
||
foshan = next(r for r in rows if r["city"] == "佛山")
|
||
if int(foshan["alert_count"]) == 3923:
|
||
ok("佛山预警数据 = 3,923")
|
||
else:
|
||
fail(f"佛山预警数据 = {foshan['alert_count']}, 期望 3923")
|
||
|
||
|
||
def check_vendor_effort():
|
||
print("[2] 校验供应商人天投入 CSV")
|
||
rows = load_csv("vendor_effort.csv")
|
||
total = sum(int(r["person_days"]) for r in rows)
|
||
if total == 100:
|
||
ok(f"人天合计 = {total}")
|
||
else:
|
||
fail(f"人天合计 = {total}, 期望 100")
|
||
|
||
expected = {
|
||
"需求沟通与前期对接": 15,
|
||
"现场培训与推广": 20,
|
||
"技术接入与配置": 18,
|
||
"问题响应与日常运营": 20,
|
||
"个性化适配与优化": 12,
|
||
"差旅在途": 10,
|
||
"文档编写": 5,
|
||
}
|
||
for row in rows:
|
||
cat = row["category"]
|
||
days = int(row["person_days"])
|
||
if cat in expected and days == expected[cat]:
|
||
ok(f"{cat} = {days} 人天")
|
||
else:
|
||
fail(f"{cat} = {days}, 期望 {expected.get(cat, '?')}")
|
||
|
||
|
||
def check_access_cost():
|
||
print("[3] 校验接入耗时 CSV")
|
||
rows = load_csv("access_cost.csv")
|
||
steps = {r["step"]: r for r in rows}
|
||
|
||
ideal_sum = sum(int(steps[s]["ideal_days"]) for s in
|
||
["账号开通", "权限配置", "组织数据匹配", "应用入口配置", "网络策略审批"])
|
||
if ideal_sum == 10:
|
||
ok(f"理想端到端 = {ideal_sum} 工作日")
|
||
else:
|
||
fail(f"理想端到端 = {ideal_sum}, 期望 10")
|
||
|
||
actual_sum = sum(int(steps[s]["actual_days"]) for s in
|
||
["账号开通", "权限配置", "组织数据匹配", "应用入口配置", "网络策略审批"])
|
||
if actual_sum == 18:
|
||
ok(f"各环节实际合计 = {actual_sum} 工作日（端到端实际约 20，含排队余量）")
|
||
else:
|
||
fail(f"各环节实际合计 = {actual_sum}, 期望 18")
|
||
|
||
|
||
def check_paper_numbers():
|
||
print("[4] 校验论文核心数值")
|
||
with open(PAPER, encoding="utf-8") as f:
|
||
text = f.read()
|
||
|
||
patterns = [
|
||
(r"144[,]?853", "累计命中异常数据 144,853"),
|
||
(r"476\s*次", "累计预警推送 476 次"),
|
||
(r"10\s*条.*规则", "监测规则 10 条"),
|
||
(r"19[,]?800", "惠州预警 19,800"),
|
||
(r"3[,]?923", "佛山预警 3,923"),
|
||
(r"8\s*个", "推广地市 8 个"),
|
||
(r"7\s*场.*培训", "推广培训 7 场"),
|
||
(r"100\s*人天", "供应商投入 100 人天"),
|
||
(r"约\s*4\s*周", "端到端实际约 4 周"),
|
||
(r"约\s*30\s*%", "标准化降本约 30%"),
|
||
]
|
||
for pat, desc in patterns:
|
||
if re.search(pat, text):
|
||
ok(f"论文含 {desc}")
|
||
else:
|
||
fail(f"论文未找到 {desc}")
|
||
|
||
|
||
def check_plots_exist():
|
||
print("[5] 校验图表文件存在")
|
||
for name in ["pilot_run_data.png", "vendor_effort.png", "access_cost.png"]:
|
||
path = os.path.join(PLOTS, name)
|
||
if os.path.isfile(path) and os.path.getsize(path) > 0:
|
||
ok(f"{name} 存在 ({os.path.getsize(path)} bytes)")
|
||
else:
|
||
fail(f"{name} 缺失或为空")
|
||
|
||
|
||
def main():
|
||
print("=" * 60)
|
||
print("数据一致性校验 — AI时代市场化运作机制研究")
|
||
print("=" * 60)
|
||
print()
|
||
|
||
check_run_data()
|
||
print()
|
||
check_vendor_effort()
|
||
print()
|
||
check_access_cost()
|
||
print()
|
||
check_paper_numbers()
|
||
print()
|
||
check_plots_exist()
|
||
|
||
print()
|
||
print("=" * 60)
|
||
if errors:
|
||
print(f"校验完成：{checks} 通过，{len(errors)} 失败")
|
||
for e in errors:
|
||
print(f" - {e}")
|
||
sys.exit(1)
|
||
else:
|
||
print(f"校验完成：全部 {checks} 项通过，0 失败")
|
||
sys.exit(0)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|