"""CLI：plan / apply / report / status。""" from __future__ import annotations import argparse import json import os import secrets import sys import time from . import __version__ from .config import DEFAULT_CONFIG, Config from .errors import QtError from .report import RunLog, write_report from .store import Store from .workflow import Importer, RunOptions def new_run_id() -> str: return time.strftime("%Y%m%d-%H%M%S") + "-" + secrets.token_hex(2) def build_importer(args) -> tuple[Importer, RunLog, Store, str]: cfg = Config.load(args.config) overrides = {"sprint": args.sprint, "work_dir": args.folder, "team_name": args.team} for key, value in overrides.items(): if value: setattr(cfg, key, value) if not cfg.sprint: raise SystemExit("配置缺少 sprint（或传 --sprint）") if not cfg.work_dir: raise SystemExit("配置缺少 work_dir（或传 --dir）") if getattr(args, "workers", None): cfg.workers = args.workers run_id = args.run_id or new_run_id() store = Store(cfg.db_path) log = RunLog(run_id, verbose=not args.quiet) store.start_run(run_id, cfg.sprint, cfg.work_dir, dry_run=args.dry_run) options = RunOptions( adopt=args.adopt, skip_tfs=args.skip_tfs, rollback_on_fail=args.rollback_on_fail, limit=args.limit, projects=args.project or [], force=args.force, dry_run=args.dry_run, ) importer = Importer(cfg, store, run_id, log, options) return importer, log, store, run_id def cmd_plan(args) -> int: importer, log, store, run_id = build_importer(args) try: rows = importer.plan() finally: pass width = max([len(row.project_name) for row in rows] + [8]) + 2 print(f"\nSprint：{importer.cfg.sprint} 目录：{importer.cfg.work_dir}") print(f"{'项目'.ljust(width)}{'动作':<12}说明") print("-" * (width + 60)) for row in rows: print(f"{row.project_name.ljust(width)}{str(row.action):<12}{row.reason}") print("-" * (width + 60)) counts: dict[str, int] = {} for row in rows: counts[str(row.action)] = counts.get(str(row.action), 0) + 1 print("合计 " + "，".join(f"{k} {v}" for k, v in sorted(counts.items()))) if args.json: payload = [ { "project": row.project_name, "file": row.file_path, "action": str(row.action), "reason": row.reason, "worksheet_id": row.worksheet.worksheet_id if row.worksheet else None, } for row in rows ] path = args.json if isinstance(args.json, str) else f"reports/{run_id}/plan.json" os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True) with open(path, "w", encoding="utf-8") as handle: json.dump(payload, handle, ensure_ascii=False, indent=2) print(f"计划已写入 {path}") log.close() store.finish_run(run_id, "PLANNED") store.close() return 0 def cmd_apply(args) -> int: if not args.dry_run and not args.yes: print("这会真实写入 QuickTeam（建单/导入/确认/推送）。确认后加 --yes 重跑；" "只想看会做什么请用 --dry-run。", file=sys.stderr) return 3 importer, log, store, run_id = build_importer(args) started = time.time() status = "DONE" try: results = importer.apply() failed = [r for r in results if r.stage.startswith("FAILED")] if failed: status = "PARTIAL" print(f"\n执行完成：{len(results)} 个项目，失败 {len(failed)}，" f"耗时 {time.time() - started:.1f}s") except QtError as error: print(f"[中止] {error}", file=sys.stderr) status = "ABORTED" return 2 finally: log.close() store.finish_run(run_id, status) md, csv = write_report(run_id, store) print(f"报告：{md} / {csv}") store.close() return 0 def cmd_report(args) -> int: cfg = Config.load(args.config, need_password=False) store = Store(cfg.db_path) run_id = args.run_id or store.last_run() if not run_id: print("没有可用的运行记录", file=sys.stderr) return 1 md, csv = write_report(run_id, store) print(f"已生成：{md}\n {csv}") store.close() return 0 def cmd_status(args) -> int: cfg = Config.load(args.config, need_password=False) store = Store(cfg.db_path) run_id = args.run_id or store.last_run() if not run_id: print("没有可用的运行记录", file=sys.stderr) return 1 jobs = store.jobs(run_id) print(f"run={run_id} 项目 {len(jobs)} 个") for job in jobs: flag = "!" if job["error"] else " " print(f"{flag} {job['project_name']:<28} {job['action']:<12} {job['stage']:<20} " f"ws={job['worksheet_id'] or '-'}") store.close() return 0 def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(prog="qtimport", description="QuickTeam 工单接口导入") parser.add_argument("--version", action="version", version=f"qtimport {__version__}") sub = parser.add_subparsers(dest="command", required=True) def common(p: argparse.ArgumentParser) -> None: p.add_argument("--config", default=DEFAULT_CONFIG, help="配置文件路径") p.add_argument("--sprint", help="覆盖配置中的 Sprint 名称") p.add_argument("--dir", dest="folder", help="覆盖配置中的 xlsx 目录") p.add_argument("--team", help="覆盖配置中的团队名称") p.add_argument("--run-id", help="复用已有 run_id（续跑）") p.add_argument("--adopt", action="store_true", help="接管无台账记录的既有工单") p.add_argument("--skip-tfs", action="store_true", help="不推送 TFS") p.add_argument("--limit", type=int, help="最多执行几个项目") p.add_argument("--project", action="append", help="只处理指定项目（可多次）") p.add_argument("--force", action="store_true", help="跳过 Excel 模板预检") p.add_argument("--dry-run", action="store_true", help="演练：不产生任何服务端写入") p.add_argument("--quiet", action="store_true", help="只输出结果") p_plan = sub.add_parser("plan", help="只读预演：列出每个项目会做什么") common(p_plan) p_plan.add_argument("--json", nargs="?", const=True, help="把计划写成 JSON") p_plan.set_defaults(func=cmd_plan, rollback_on_fail=False) p_apply = sub.add_parser("apply", help="执行导入") common(p_apply) p_apply.add_argument("--yes", action="store_true", help="确认真实写入") p_apply.add_argument("--rollback-on-fail", action="store_true", help="失败时删除本次新建的半成品工单") p_apply.add_argument("--workers", type=int, help="覆盖并发数（默认取配置）") p_apply.set_defaults(func=cmd_apply, json=False) p_report = sub.add_parser("report", help="生成报告") p_report.add_argument("--config", default=DEFAULT_CONFIG) p_report.add_argument("--run-id") p_report.set_defaults(func=cmd_report, quiet=False) p_status = sub.add_parser("status", help="查看最近一次运行") p_status.add_argument("--config", default=DEFAULT_CONFIG) p_status.add_argument("--run-id") p_status.set_defaults(func=cmd_status, quiet=False) args = parser.parse_args(argv) try: return args.func(args) except QtError as error: print(f"[ERROR] {error}", file=sys.stderr) return 2 if __name__ == "__main__": sys.exit(main())