65 lines
2.6 KiB
Python
65 lines
2.6 KiB
Python
|
# ============================================================
|
|||
|
|
# 生成计划员视口 E2E 的离线桩数据：现场 world.json → 一组接口原始响应 JSON
|
|||
|
|
# 用法：python scripts/gen_planner_e2e_fixtures.py <world.json> <输出目录>
|
|||
|
|
# 例：python scripts/gen_planner_e2e_fixtures.py `
|
|||
|
|
# server/data/tenants/<tenant>/projects/<project>/world.json $env:TEMP\aps-planner-fix
|
|||
|
|
#
|
|||
|
|
# 只读：仅读取 world.json，并调用网关端点用的同一批域函数生成响应体。
|
|||
|
|
# 不手写任何业务数字，也不写回项目数据。
|
|||
|
|
# 产物目录交给 apps/web/e2e/planner-real-data-views.spec.ts 的 E2E_PLANNER_FIXTURES 使用。
|
|||
|
|
# ============================================================
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import json
|
|||
|
|
import os
|
|||
|
|
import sys
|
|||
|
|
|
|||
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|||
|
|
|
|||
|
|
from server.aps_domain.analytics import (
|
|||
|
|
build_compare_table,
|
|||
|
|
build_kpi_dashboard,
|
|||
|
|
build_utilization_report,
|
|||
|
|
)
|
|||
|
|
from server.aps_domain.flex import capacity_analysis, flex_gantt_view, flex_overview
|
|||
|
|
from server.aps_domain.planning import build_plan_buckets
|
|||
|
|
from server.aps_domain.views import due_view, gantt_view, load_view, world_summary
|
|||
|
|
|
|||
|
|
|
|||
|
|
def build_payloads(world: dict) -> dict:
|
|||
|
|
return {
|
|||
|
|
"world-summary.json": world_summary(world),
|
|||
|
|
"world-gantt.json": gantt_view(world),
|
|||
|
|
"world-load.json": load_view(world, 14),
|
|||
|
|
"world-due.json": {"rows": due_view(world)},
|
|||
|
|
"flex-gantt.json": flex_gantt_view(world),
|
|||
|
|
"flex-capacity.json": capacity_analysis(world),
|
|||
|
|
"flex-world.json": flex_overview(world),
|
|||
|
|
"analytics-kpi.json": build_kpi_dashboard(world),
|
|||
|
|
"analytics-util.json": build_utilization_report(world, "flex", 7),
|
|||
|
|
"analytics-compare.json": build_compare_table(world, track="both"),
|
|||
|
|
"plan-buckets.json": build_plan_buckets(
|
|||
|
|
world, mode="HYBRID", horizon_days=90, include_forecast=True, capacity_mode="FINITE"
|
|||
|
|
),
|
|||
|
|
"timeline.json": {"versions": [], "checkpoints": []},
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def main(argv: list[str]) -> int:
|
|||
|
|
if len(argv) != 3:
|
|||
|
|
print(__doc__)
|
|||
|
|
return 2
|
|||
|
|
world_path, out_dir = argv[1], argv[2]
|
|||
|
|
with open(world_path, encoding="utf-8") as fh:
|
|||
|
|
world = json.load(fh)
|
|||
|
|
os.makedirs(out_dir, exist_ok=True)
|
|||
|
|
for name, body in build_payloads(world).items():
|
|||
|
|
with open(os.path.join(out_dir, name), "w", encoding="utf-8") as fh:
|
|||
|
|
json.dump(body, fh, ensure_ascii=False)
|
|||
|
|
print("wrote", name, len(json.dumps(body, ensure_ascii=False)), "bytes")
|
|||
|
|
return 0
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
raise SystemExit(main(sys.argv))
|