chenshoulin
58850314cf
用「WCF + 通用存储过程网关」接口替代 Selenium（D://Learning//Python//工单自动化）导入工单： 单 Sprint 12 个项目 175s 跑完（旧方案 25~35 分钟），失败可断点续跑且不重复建单。 - 分层：transport（HTTP/协议）/ api（存储过程语义）/ domain（纯规则）/ workflow（阶段编排）/ store（SQLite 台账）/ report / CLI - 幂等：台账驱动断点续跑；dry-run 与跳过行不污染进度；--rollback-on-fail 清理半成品 - 命令：plan / apply / report / status，真实写入需 --yes - tools：只读探针、存储过程源码导出（sp_helptext）、导入诊断 - docs：调研与设计文档（含联调踩到的 5 个坑）+ 6 个关键存储过程源码 实测：Sprint 2026 Sprint 10 的 12 个项目全部完成建单→导入→确认→推送 TFS， 工单 225134–225145，35 个功能模块，重跑 plan 全部 skip-pushed。69 个单测通过。
116 lines
4.0 KiB
Python
116 lines
4.0 KiB
Python
"""日志与报告。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import csv
|
||
import json
|
||
import os
|
||
import time
|
||
from typing import Any
|
||
|
||
from .store import Store
|
||
|
||
|
||
class RunLog:
|
||
"""控制台摘要 + JSONL 明细。"""
|
||
|
||
def __init__(self, run_id: str, log_dir: str = "logs", *, verbose: bool = True) -> None:
|
||
os.makedirs(log_dir, exist_ok=True)
|
||
self.path = os.path.join(log_dir, f"{run_id}.jsonl")
|
||
self.verbose = verbose
|
||
self.run_id = run_id
|
||
self._handle = open(self.path, "a", encoding="utf-8")
|
||
|
||
def close(self) -> None:
|
||
self._handle.close()
|
||
|
||
def _write(self, record: dict[str, Any]) -> None:
|
||
record.setdefault("ts", time.strftime("%Y-%m-%d %H:%M:%S"))
|
||
record.setdefault("run", self.run_id)
|
||
self._handle.write(json.dumps(record, ensure_ascii=False) + "\n")
|
||
self._handle.flush()
|
||
|
||
def info(self, message: str, **fields: Any) -> None:
|
||
self._write({"level": "info", "message": message, **fields})
|
||
if self.verbose:
|
||
print(message)
|
||
|
||
def event(self, stage: str, action: str, *, project: str = "", detail: str = "", elapsed_ms: int = 0) -> None:
|
||
self._write(
|
||
{
|
||
"level": "event",
|
||
"stage": stage,
|
||
"action": action,
|
||
"project": project,
|
||
"detail": detail,
|
||
"elapsed_ms": elapsed_ms,
|
||
}
|
||
)
|
||
|
||
def warn(self, message: str, **fields: Any) -> None:
|
||
self._write({"level": "warn", "message": message, **fields})
|
||
if self.verbose:
|
||
print(f" ! {message}")
|
||
|
||
|
||
def write_report(run_id: str, store: Store, out_dir: str = "reports") -> tuple[str, str]:
|
||
"""生成 summary.md 与 detail.csv。"""
|
||
target = os.path.join(out_dir, run_id)
|
||
os.makedirs(target, exist_ok=True)
|
||
|
||
jobs = store.jobs(run_id)
|
||
events = store.events(run_id)
|
||
|
||
md_path = os.path.join(target, "summary.md")
|
||
csv_path = os.path.join(target, "detail.csv")
|
||
|
||
counts: dict[str, int] = {}
|
||
for job in jobs:
|
||
counts[job["action"]] = counts.get(job["action"], 0) + 1
|
||
failures = [job for job in jobs if job["error"]]
|
||
|
||
with open(csv_path, "w", encoding="utf-8-sig", newline="") as handle:
|
||
writer = csv.writer(handle)
|
||
writer.writerow(
|
||
["项目", "Sprint", "文件", "动作", "阶段", "工单号", "版本", "耗时(ms)", "错误"]
|
||
)
|
||
for job in jobs:
|
||
elapsed = sum(
|
||
e["elapsed_ms"] or 0
|
||
for e in events
|
||
if e["project_id"] == job["project_id"] and e["elapsed_ms"]
|
||
)
|
||
writer.writerow(
|
||
[
|
||
job["project_name"],
|
||
job["system_name"],
|
||
os.path.basename(job["source_file"]),
|
||
job["action"],
|
||
job["stage"],
|
||
job["worksheet_id"] or "",
|
||
job["version_id"] or "",
|
||
elapsed,
|
||
(job["error"] or "").replace("\n", " ")[:300],
|
||
]
|
||
)
|
||
|
||
info = store.run_info(run_id) or {}
|
||
mode = "演练（未写入任何数据）" if info.get("dry_run") else "实际执行"
|
||
|
||
with open(md_path, "w", encoding="utf-8") as handle:
|
||
handle.write(f"# 导入报告 {run_id}\n\n")
|
||
handle.write(f"- 模式：{mode}\n")
|
||
handle.write(f"- Sprint：{info.get('sprint', '')}\n")
|
||
handle.write(f"- 项目数：{len(jobs)}\n")
|
||
handle.write(f"- 失败：{len(failures)}\n")
|
||
for action, count in sorted(counts.items()):
|
||
handle.write(f"- {action}：{count}\n")
|
||
handle.write("\n| 项目 | 动作 | 阶段 | 工单号 | 错误 |\n| --- | --- | --- | --- | --- |\n")
|
||
for job in jobs:
|
||
error = (job["error"] or "").replace("\n", " ")[:120].replace("|", "\\|")
|
||
handle.write(
|
||
f"| {job['project_name']} | {job['action']} | {job['stage']} | "
|
||
f"{job['worksheet_id'] or '-'} | {error or ''} |\n"
|
||
)
|
||
return md_path, csv_path
|