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 个单测通过。
208 lines
7.2 KiB
Python
208 lines
7.2 KiB
Python
"""SQLite 台账：断点、幂等判断与审计的唯一事实源。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sqlite3
|
|
import threading
|
|
import time
|
|
from typing import Any
|
|
|
|
SCHEMA = """
|
|
CREATE TABLE IF NOT EXISTS runs (
|
|
run_id TEXT PRIMARY KEY,
|
|
started_at TEXT NOT NULL,
|
|
finished_at TEXT,
|
|
sprint TEXT NOT NULL,
|
|
work_dir TEXT NOT NULL,
|
|
dry_run INTEGER NOT NULL DEFAULT 0,
|
|
status TEXT NOT NULL DEFAULT 'RUNNING'
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS jobs (
|
|
run_id TEXT NOT NULL,
|
|
project_id TEXT NOT NULL,
|
|
project_name TEXT NOT NULL,
|
|
system_name TEXT NOT NULL,
|
|
source_file TEXT NOT NULL,
|
|
action TEXT NOT NULL,
|
|
stage TEXT NOT NULL,
|
|
worksheet_id INTEGER,
|
|
version_id INTEGER,
|
|
reason TEXT,
|
|
error TEXT,
|
|
updated_at TEXT NOT NULL,
|
|
PRIMARY KEY (run_id, project_id)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS events (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
run_id TEXT NOT NULL,
|
|
project_id TEXT,
|
|
stage TEXT NOT NULL,
|
|
action TEXT NOT NULL,
|
|
detail TEXT,
|
|
elapsed_ms INTEGER,
|
|
created_at TEXT NOT NULL
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_jobs_project ON jobs (project_id, system_name, updated_at);
|
|
CREATE INDEX IF NOT EXISTS idx_events_run ON events (run_id, id);
|
|
"""
|
|
|
|
|
|
def now() -> str:
|
|
return time.strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
|
|
class Store:
|
|
def __init__(self, db_path: str) -> None:
|
|
directory = os.path.dirname(os.path.abspath(db_path))
|
|
os.makedirs(directory, exist_ok=True)
|
|
self._lock = threading.Lock()
|
|
self._conn = sqlite3.connect(db_path, check_same_thread=False)
|
|
self._conn.row_factory = sqlite3.Row
|
|
with self._lock:
|
|
self._conn.executescript(SCHEMA)
|
|
self._conn.commit()
|
|
|
|
def close(self) -> None:
|
|
with self._lock:
|
|
self._conn.close()
|
|
|
|
# ------------------------------------------------------------------ runs
|
|
|
|
def start_run(self, run_id: str, sprint: str, work_dir: str, dry_run: bool) -> None:
|
|
with self._lock:
|
|
self._conn.execute(
|
|
"INSERT OR REPLACE INTO runs (run_id, started_at, sprint, work_dir, dry_run, status) "
|
|
"VALUES (?, ?, ?, ?, ?, 'RUNNING')",
|
|
(run_id, now(), sprint, work_dir, int(dry_run)),
|
|
)
|
|
self._conn.commit()
|
|
|
|
def finish_run(self, run_id: str, status: str = "DONE") -> None:
|
|
with self._lock:
|
|
self._conn.execute(
|
|
"UPDATE runs SET finished_at = ?, status = ? WHERE run_id = ?", (now(), status, run_id)
|
|
)
|
|
self._conn.commit()
|
|
|
|
# ------------------------------------------------------------------ jobs
|
|
|
|
def upsert_job(
|
|
self,
|
|
run_id: str,
|
|
*,
|
|
project_id: str,
|
|
project_name: str,
|
|
system_name: str,
|
|
source_file: str,
|
|
action: str,
|
|
stage: str,
|
|
worksheet_id: int | None = None,
|
|
version_id: int | None = None,
|
|
reason: str = "",
|
|
error: str = "",
|
|
) -> None:
|
|
with self._lock:
|
|
self._conn.execute(
|
|
"""
|
|
INSERT INTO jobs (run_id, project_id, project_name, system_name, source_file, action, stage,
|
|
worksheet_id, version_id, reason, error, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(run_id, project_id) DO UPDATE SET
|
|
action = excluded.action,
|
|
stage = excluded.stage,
|
|
worksheet_id = COALESCE(excluded.worksheet_id, jobs.worksheet_id),
|
|
version_id = COALESCE(excluded.version_id, jobs.version_id),
|
|
reason = excluded.reason,
|
|
error = excluded.error,
|
|
updated_at = excluded.updated_at
|
|
""",
|
|
(
|
|
run_id,
|
|
project_id,
|
|
project_name,
|
|
system_name,
|
|
source_file,
|
|
str(action),
|
|
stage,
|
|
worksheet_id,
|
|
version_id,
|
|
reason,
|
|
error,
|
|
now(),
|
|
),
|
|
)
|
|
self._conn.commit()
|
|
|
|
def latest_stage(self, project_id: str, system_name: str) -> str | None:
|
|
"""跨 run 取该项目该 Sprint 最近一次落库的阶段，用于断点续跑。"""
|
|
job = self.latest_job(project_id, system_name)
|
|
return job["stage"] if job else None
|
|
|
|
def latest_job(self, project_id: str, system_name: str) -> dict[str, Any] | None:
|
|
"""最近一次**真实**执行记录。dry-run 不参与幂等判断，否则会把演练当成已完成。
|
|
|
|
`SKIPPED` 行也不参与：它只表示「这一轮不用做」，不带任何进度信息。若把它当成
|
|
最近记录，下次 plan 会把「已推送 TFS」看成「台账停在 SKIPPED」，误判 resume 并
|
|
再推一次 —— 而 TFS 推送接口没有服务端去重，重复推送就是重复工作项。
|
|
"""
|
|
with self._lock:
|
|
row = self._conn.execute(
|
|
"""
|
|
SELECT j.* FROM jobs j
|
|
JOIN runs r ON r.run_id = j.run_id
|
|
WHERE j.project_id = ? AND j.system_name = ? AND r.dry_run = 0
|
|
AND j.stage <> 'SKIPPED'
|
|
ORDER BY j.updated_at DESC LIMIT 1
|
|
""",
|
|
(project_id, system_name),
|
|
).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def jobs(self, run_id: str) -> list[dict[str, Any]]:
|
|
with self._lock:
|
|
rows = self._conn.execute(
|
|
"SELECT * FROM jobs WHERE run_id = ? ORDER BY project_name", (run_id,)
|
|
).fetchall()
|
|
return [dict(row) for row in rows]
|
|
|
|
# ------------------------------------------------------------------ events
|
|
|
|
def add_event(
|
|
self,
|
|
run_id: str,
|
|
*,
|
|
stage: str,
|
|
action: str,
|
|
project_id: str = "",
|
|
detail: str = "",
|
|
elapsed_ms: int = 0,
|
|
) -> None:
|
|
with self._lock:
|
|
self._conn.execute(
|
|
"INSERT INTO events (run_id, project_id, stage, action, detail, elapsed_ms, created_at) "
|
|
"VALUES (?, ?, ?, ?, ?, ?, ?)",
|
|
(run_id, project_id, stage, action, detail[:2000], elapsed_ms, now()),
|
|
)
|
|
self._conn.commit()
|
|
|
|
def events(self, run_id: str) -> list[dict[str, Any]]:
|
|
with self._lock:
|
|
rows = self._conn.execute(
|
|
"SELECT * FROM events WHERE run_id = ? ORDER BY id", (run_id,)
|
|
).fetchall()
|
|
return [dict(row) for row in rows]
|
|
|
|
def run_info(self, run_id: str) -> dict[str, Any] | None:
|
|
with self._lock:
|
|
row = self._conn.execute("SELECT * FROM runs WHERE run_id = ?", (run_id,)).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def last_run(self) -> str | None:
|
|
with self._lock:
|
|
row = self._conn.execute("SELECT run_id FROM runs ORDER BY started_at DESC LIMIT 1").fetchone()
|
|
return row["run_id"] if row else None
|