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 个单测通过。
110 lines
4.7 KiB
Python
110 lines
4.7 KiB
Python
"""诊断工具：定位 BatchImportFeature 空引用异常。
|
||
|
||
只操作一张临时草稿工单，结束时删除，不影响业务数据。
|
||
|
||
用法：
|
||
python tools/diag_import.py # 建单→填单→读字段→试导入→删单
|
||
python tools/diag_import.py --keep # 保留草稿（便于在页面上比对）
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import sys
|
||
|
||
sys.path.insert(0, ".")
|
||
|
||
from qtimport.api import CONN_QUICKTEAM, SP_IMPORT_EXCEL, SP_WORKSHEET, QuickTeamApi # noqa: E402
|
||
from qtimport.config import Config # noqa: E402
|
||
from qtimport.transport import Transport # noqa: E402
|
||
|
||
|
||
def dump(title: str, rows) -> dict:
|
||
print(f"\n=== {title} ===")
|
||
if isinstance(rows, list) and rows:
|
||
print(json.dumps(rows[0], ensure_ascii=False, indent=2))
|
||
return rows[0]
|
||
print(json.dumps(rows, ensure_ascii=False, indent=2))
|
||
return {}
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser()
|
||
parser.add_argument("--config", default="config/config.json")
|
||
parser.add_argument("--project", default="资管 - 华泰保险资管")
|
||
parser.add_argument("--excel", default=r"D:\Work\QuickTeam\2026 Sprint 10\资管 - 华泰保险资管.xlsx")
|
||
parser.add_argument("--keep", action="store_true")
|
||
args = parser.parse_args()
|
||
|
||
cfg = Config.load(args.config)
|
||
t = Transport(cfg.base_url, timeout=cfg.timeout, retries=0)
|
||
api = QuickTeamApi(t)
|
||
user = api.login(cfg.username, cfg.password)
|
||
team = api.resolve_team(api.teams(user.user_id), cfg.team_name)
|
||
project = next(p for p in api.projects(team.team_id) if p.project_name == args.project)
|
||
print(f"登录：{user.real_name}；项目：{project.project_name}")
|
||
|
||
worksheet_id = api.create_worksheet()
|
||
print(f"草稿工单：{worksheet_id}")
|
||
try:
|
||
dump("建单后", t.exec_table(CONN_QUICKTEAM, SP_WORKSHEET,
|
||
[("WorkSheetId", worksheet_id, "int")], op="raw"))
|
||
|
||
evaluator, technical, customer = cfg.resolved_roles(user.real_name)
|
||
print(f"角色：评估={evaluator} 技术={technical} 客户={customer}")
|
||
result = api.fill_worksheet(
|
||
worksheet_id, project,
|
||
system_name=cfg.sprint, start_time=cfg.start_time, end_time=cfg.end_time,
|
||
evaluator=evaluator, technical_leader=technical, customer_leader=customer,
|
||
belong_team=cfg.belong_team, content=cfg.content_desc,
|
||
operator=user.real_name, service_type=cfg.service_type,
|
||
)
|
||
print(f"填单 Result={result}")
|
||
|
||
after = dump("填单后", t.exec_table(CONN_QUICKTEAM, SP_WORKSHEET,
|
||
[("WorkSheetId", worksheet_id, "int")], op="raw"))
|
||
print(f"CurrentVersionId={after.get('CurrentVersionId')!r} VersionCode={after.get('VersionCode')!r}")
|
||
|
||
remote = api.upload_excel(args.excel)
|
||
print(f"\n上传成功：{remote}")
|
||
|
||
# 逐个尝试参数组合，看哪种不报空引用
|
||
variants = [
|
||
("VersionId=0", [("WorkSheetId", worksheet_id, "int"), ("VersionId", 0, "int")]),
|
||
("VersionId=-1", [("WorkSheetId", worksheet_id, "int"), ("VersionId", -1, "int")]),
|
||
("VersionId 省略", [("WorkSheetId", worksheet_id, "int")]),
|
||
("VersionId=空串", [("WorkSheetId", worksheet_id, "int"), ("VersionId", "", "string")]),
|
||
]
|
||
for label, params in variants:
|
||
print(f"\n--- 尝试 {label} ---")
|
||
try:
|
||
out = t.batch_import(SP_IMPORT_EXCEL, params, remote)
|
||
print("返回：", json.dumps(out, ensure_ascii=False))
|
||
if isinstance(out, list) and out and isinstance(out[0], dict) and int(out[0].get("Column1", -1)) == 0:
|
||
print(f"✔ {label} 成功")
|
||
break
|
||
except Exception as error: # noqa: BLE001
|
||
print(f"✘ {type(error).__name__}: {error}")
|
||
|
||
dump("导入后", t.exec_table(CONN_QUICKTEAM, SP_WORKSHEET,
|
||
[("WorkSheetId", worksheet_id, "int")], op="raw"))
|
||
feats = api.features(worksheet_id)
|
||
print(f"功能模块数量：{len(feats)}")
|
||
for f in feats[:3]:
|
||
print(" -", f.feature_id, f.code, f.name)
|
||
finally:
|
||
if args.keep:
|
||
print(f"\n草稿保留：{worksheet_id}（请手工核对后删除）")
|
||
else:
|
||
try:
|
||
api.delete_worksheet(worksheet_id)
|
||
print(f"\n已删除草稿 {worksheet_id}")
|
||
except Exception as error: # noqa: BLE001
|
||
print(f"\n删除失败：{error}（请手工删除 {worksheet_id}）")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|