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 个单测通过。
366 lines
14 KiB
Python
366 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""QuickTeam 接口探针（只读）。
|
||
|
||
用途：在不动任何业务数据的前提下，验证「通过接口驱动 QuickTeam」的可行性，
|
||
并输出一份「本 Sprint 各项目工单现状」清单（plan 阶段的雏形）。
|
||
|
||
只调用以下只读接口：
|
||
usp_Authentication / usp_GetTeamsByUserId / usp_GetProjectsByTeamId
|
||
usp_GetWeeklyWorkSheetsByProjectId / usp_GetWorkSheetById / usp_GetProjectUsersByProjectId
|
||
可选（--upload-test）额外验证文件上传通道，会在服务器 WorkSheetImport 目录留一个临时文件，
|
||
不触发任何解析或写库动作。
|
||
|
||
用法：
|
||
python tools/api_probe.py --config config/config.json
|
||
python tools/api_probe.py --config config/config.json --sprint "2026 Sprint 10" --dir "D:/Work/QuickTeam/2026 Sprint 10"
|
||
python tools/api_probe.py --config config/config.json --upload-test
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import time
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
from dataclasses import dataclass, field
|
||
from typing import Any
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 传输层：WCF + 通用存储过程网关
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class QtError(RuntimeError):
|
||
"""接口/业务错误。"""
|
||
|
||
|
||
@dataclass
|
||
class QtClient:
|
||
base_url: str
|
||
timeout: int = 60
|
||
retries: int = 2
|
||
|
||
# WCF 服务路径（来自前端 app.js 的 a078 配置模块）
|
||
common_url: str = "/Services/CommonService.svc/"
|
||
file_url: str = "/Services/FileOperationService.svc/"
|
||
|
||
# -- 底层 ---------------------------------------------------------------
|
||
|
||
def _post(self, path: str, payload: Any = None) -> str:
|
||
url = self.base_url.rstrip("/") + path
|
||
data = b"" if payload is None else json.dumps(payload).encode("utf-8")
|
||
last: Exception | None = None
|
||
for attempt in range(self.retries + 1):
|
||
req = urllib.request.Request(
|
||
url,
|
||
data=data,
|
||
headers={"Content-Type": "application/json", "Accept": "application/json"},
|
||
method="POST",
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=self.timeout) as resp:
|
||
return resp.read().decode("utf-8-sig", "replace")
|
||
except urllib.error.HTTPError as exc:
|
||
raise QtError(self._extract_error(exc)) from None
|
||
except (urllib.error.URLError, TimeoutError) as exc: # 网络层可重试
|
||
last = exc
|
||
if attempt < self.retries:
|
||
time.sleep(1.5 * (attempt + 1))
|
||
continue
|
||
raise QtError(f"网络不可达：{exc}") from None
|
||
raise QtError(str(last))
|
||
|
||
@staticmethod
|
||
def _extract_error(exc: urllib.error.HTTPError) -> str:
|
||
body = exc.read().decode("utf-8-sig", "replace")
|
||
# WCF 返回 HTML 错误页，真正的错误在 "The exception message is '...'"
|
||
match = re.search(r"exception message is '(.*?)'.?\s*See server logs", body, re.S)
|
||
if match:
|
||
return f"HTTP {exc.code}: {match.group(1).strip()}"
|
||
plain = re.sub(r"<[^>]+>", " ", body)
|
||
plain = re.sub(r"\s+", " ", plain).strip()
|
||
return f"HTTP {exc.code}: {plain[-300:]}"
|
||
|
||
@staticmethod
|
||
def _decode(raw: str) -> Any:
|
||
"""WCF 返回值常是「JSON 字符串里再套一层 JSON」，最多解三层。"""
|
||
value: Any = raw.strip()
|
||
for _ in range(3):
|
||
if not isinstance(value, str):
|
||
break
|
||
try:
|
||
value = json.loads(value)
|
||
except json.JSONDecodeError:
|
||
break
|
||
return value
|
||
|
||
def _exec(self, conn: str, sp: str, params: list[tuple[str, Any, str]]) -> Any:
|
||
"""调用存储过程。params: [(Name, Value, DBType), ...]"""
|
||
inner = {
|
||
"SPName": sp,
|
||
"SQLParams": [
|
||
{"Name": name, "Value": "" if value is None else str(value), "DBType": dbtype}
|
||
for name, value, dbtype in params
|
||
],
|
||
}
|
||
payload = {
|
||
"connectionName": conn,
|
||
"param": urllib.parse.quote(json.dumps(inner, separators=(",", ":"))),
|
||
}
|
||
return self._decode(self._post(self.common_url + "ExecTable", payload))
|
||
|
||
def _exec_result(self, conn: str, sp: str, params: list[tuple[str, Any, str]]) -> Any:
|
||
"""同 _exec，但走 ExecResult（标量/返回值语义，如 Result、Column1）。"""
|
||
inner = {
|
||
"SPName": sp,
|
||
"SQLParams": [
|
||
{"Name": name, "Value": "" if value is None else str(value), "DBType": dbtype}
|
||
for name, value, dbtype in params
|
||
],
|
||
}
|
||
payload = {
|
||
"connectionName": conn,
|
||
"param": urllib.parse.quote(json.dumps(inner, separators=(",", ":"))),
|
||
}
|
||
return self._decode(self._post(self.common_url + "ExecResult", payload))
|
||
|
||
# -- 业务语义 -----------------------------------------------------------
|
||
|
||
def login(self, username: str, password: str) -> dict:
|
||
digest = hashlib.md5(password.encode("utf-8")).hexdigest()
|
||
rows = self._exec(
|
||
"UserCenter",
|
||
"dbo.usp_Authentication",
|
||
[("UserName", username, "string"), ("Password", digest, "string")],
|
||
)
|
||
if not rows:
|
||
raise QtError("登录失败：账号或密码不匹配")
|
||
return rows[0]
|
||
|
||
def teams(self, user_id: str) -> list[dict]:
|
||
return self._exec("QuickTeam", "QuickTeam.usp_GetTeamsByUserId", [("UserId", user_id, "string")]) or []
|
||
|
||
def projects(self, team_id: str) -> list[dict]:
|
||
return self._exec(
|
||
"QuickTeam",
|
||
"QuickTeam.usp_GetProjectsByTeamId",
|
||
[("TeamId", team_id, "string"), ("IsTemplate", "1", "string")],
|
||
) or []
|
||
|
||
def worksheets(self, project_id: str, start: str, end: str) -> list[dict]:
|
||
return self._exec(
|
||
"QuickTeam",
|
||
"QuickTeam.usp_GetWeeklyWorkSheetsByProjectId",
|
||
[("ProjectId", project_id, "string"), ("Monday", start, "string"), ("Sunday", end, "string")],
|
||
) or []
|
||
|
||
def worksheet(self, worksheet_id: int) -> dict | None:
|
||
rows = self._exec("QuickTeam", "QuickTeam.usp_GetWorkSheetById", [("WorkSheetId", worksheet_id, "int")])
|
||
return rows[0] if rows else None
|
||
|
||
def project_users(self, project_id: str) -> list[dict]:
|
||
return self._exec(
|
||
"QuickTeam", "QuickTeam.usp_GetProjectUsersByProjectId", [("ProjectId", project_id, "string")]
|
||
) or []
|
||
|
||
# -- 文件通道 -----------------------------------------------------------
|
||
|
||
def upload(self, local_path: str, folder: str = "WorkSheetImport") -> str:
|
||
"""上传文件，返回服务器端绝对路径（BatchImportFeature 的 filePath）。"""
|
||
name = f"{int(time.time() * 1000)}{os.path.splitext(local_path)[1]}"
|
||
url = (
|
||
f"{self.base_url.rstrip('/')}{self.file_url}FileUpload"
|
||
f"?fileName={urllib.parse.quote(name)}&fileFolder={urllib.parse.quote(folder)}"
|
||
)
|
||
boundary = "----qtprobe" + hashlib.md5(name.encode()).hexdigest()
|
||
body = (
|
||
f"--{boundary}\r\n"
|
||
f'Content-Disposition: form-data; name="file"; filename="{name}"\r\n'
|
||
f"Content-Type: application/octet-stream\r\n\r\n"
|
||
).encode("utf-8")
|
||
body += open(local_path, "rb").read()
|
||
body += f"\r\n--{boundary}--\r\n".encode("utf-8")
|
||
req = urllib.request.Request(
|
||
url, data=body, headers={"Content-Type": f"multipart/form-data; boundary={boundary}"}, method="POST"
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=self.timeout) as resp:
|
||
data = self._decode(resp.read().decode("utf-8-sig", "replace"))
|
||
except urllib.error.HTTPError as exc:
|
||
raise QtError(self._extract_error(exc)) from None
|
||
path = (data or {}).get("FileUploadResult")
|
||
if not path:
|
||
raise QtError(f"上传未返回 FileUploadResult：{data!r}")
|
||
return path
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 报表
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def ms_to_date(value: Any) -> str:
|
||
"""QuickTeam 日期为 /Date(1785081600000)/ 形式。"""
|
||
if not value:
|
||
return ""
|
||
match = re.search(r"\((\d+)", str(value))
|
||
if not match:
|
||
return str(value)
|
||
return time.strftime("%Y-%m-%d", time.localtime(int(match.group(1)) / 1000))
|
||
|
||
|
||
def list_xlsx(folder: str) -> list[str]:
|
||
if not os.path.isdir(folder):
|
||
raise SystemExit(f"目录不存在：{folder}")
|
||
names = []
|
||
for entry in sorted(os.listdir(folder)):
|
||
path = os.path.join(folder, entry)
|
||
if os.path.isfile(path) and entry.lower().endswith(".xlsx") and not entry.startswith("~$"):
|
||
names.append(path)
|
||
return names
|
||
|
||
|
||
def normalize(name: str) -> str:
|
||
return re.sub(r"\s+", "", name).replace("－", "-").replace("—", "-")
|
||
|
||
|
||
@dataclass
|
||
class Row:
|
||
project: str
|
||
file: str
|
||
project_id: str = ""
|
||
existing: int = 0
|
||
has_sprint: bool = False
|
||
worksheet_ids: list[int] = field(default_factory=list)
|
||
note: str = ""
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="QuickTeam 接口探针（只读）")
|
||
parser.add_argument("--config", default="config/config.json", help="配置文件路径")
|
||
parser.add_argument("--sprint", help="覆盖配置中的 sprint 名称（SystemName）")
|
||
parser.add_argument("--dir", dest="folder", help="覆盖配置中的 xlsx 目录")
|
||
parser.add_argument("--team", help="覆盖配置中的团队名称")
|
||
parser.add_argument("--upload-test", action="store_true", help="额外验证上传通道（会留一个临时文件）")
|
||
parser.add_argument("--json-out", help="把结果另存为 JSON")
|
||
args = parser.parse_args()
|
||
|
||
if not os.path.exists(args.config):
|
||
raise SystemExit(f"缺少配置文件：{args.config}（可从 config/config.example.json 复制）")
|
||
with open(args.config, encoding="utf-8-sig") as handle:
|
||
cfg = json.load(handle)
|
||
|
||
sprint = args.sprint or cfg["sprint"]
|
||
folder = args.folder or cfg["work_dir"]
|
||
team_name = args.team or cfg.get("team_name", "当前项目")
|
||
start, end = cfg.get("start_time", ""), cfg.get("end_time", "")
|
||
if not start or not end:
|
||
today = time.strftime("%Y-%m-%d")
|
||
start, end = "2020-01-01", today
|
||
|
||
client = QtClient(cfg["base_url"], timeout=int(cfg.get("timeout", 60)))
|
||
t0 = time.time()
|
||
user = client.login(cfg["username"], cfg["password"])
|
||
t_login = time.time() - t0
|
||
print(f"[登录] {user['RealName']} <{user['UserName']}> UserId={user['GuidId']} {t_login:.2f}s")
|
||
|
||
t1 = time.time()
|
||
teams = client.teams(user["GuidId"])
|
||
team = next((x for x in teams if x.get("TeamName") == team_name), None)
|
||
if not team:
|
||
raise SystemExit(f"未找到团队「{team_name}」，可选：{[x.get('TeamName') for x in teams]}")
|
||
t_teams = time.time() - t1
|
||
|
||
t2 = time.time()
|
||
projects = client.projects(team["TeamId"])
|
||
t_projects = time.time() - t2
|
||
print(f"[团队] {team['TeamName']} ({team['TeamId']}) {t_teams:.2f}s")
|
||
print(f"[项目] 共 {len(projects)} 个 {t_projects:.2f}s")
|
||
|
||
by_name = {normalize(p["ProjectName"]): p for p in projects}
|
||
rows: list[Row] = []
|
||
t3 = time.time()
|
||
for path in list_xlsx(folder):
|
||
stem = os.path.splitext(os.path.basename(path))[0]
|
||
row = Row(project=stem, file=os.path.basename(path))
|
||
hit = by_name.get(normalize(stem))
|
||
if not hit:
|
||
row.note = "项目名在 QuickTeam 中不存在"
|
||
rows.append(row)
|
||
continue
|
||
row.project_id = hit["ProjectId"]
|
||
sheets = client.worksheets(row.project_id, start, end)
|
||
row.existing = len(sheets)
|
||
for sheet in sheets:
|
||
detail = client.worksheet(sheet["WorkSheetId"]) or {}
|
||
row.worksheet_ids.append(sheet["WorkSheetId"])
|
||
if detail.get("SystemName") == sprint:
|
||
row.has_sprint = True
|
||
rows.append(row)
|
||
t_scan = time.time() - t3
|
||
|
||
width = max([len(r.project) for r in rows] + [8]) + 2
|
||
print()
|
||
print(f"{'项目'.ljust(width)} {'工单数':>5} {'本Sprint':>8} 状态")
|
||
print("-" * (width + 34))
|
||
for row in rows:
|
||
if row.note:
|
||
state = f"!! {row.note}"
|
||
elif row.has_sprint:
|
||
state = "已存在（应跳过或续跑）"
|
||
else:
|
||
state = "待创建"
|
||
print(f"{row.project.ljust(width)} {row.existing:>5} {'是' if row.has_sprint else '否':>8} {state}")
|
||
|
||
todo = sum(1 for r in rows if not r.note and not r.has_sprint)
|
||
missing = sum(1 for r in rows if r.note)
|
||
print("-" * (width + 34))
|
||
print(f"合计 {len(rows)} 个文件：待创建 {todo}，已存在 {len(rows) - todo - missing}，异常 {missing}")
|
||
print(f"[耗时] 扫描 {t_scan:.2f}s 平均 {t_scan / max(len(rows), 1):.2f}s/项目")
|
||
|
||
if args.upload_test:
|
||
sample = next((p for p in list_xlsx(folder)), None)
|
||
if sample:
|
||
t4 = time.time()
|
||
remote = client.upload(sample)
|
||
print(f"\n[上传] {os.path.basename(sample)} -> {remote} {time.time() - t4:.2f}s（已上传但不解析）")
|
||
|
||
if args.json_out:
|
||
os.makedirs(os.path.dirname(os.path.abspath(args.json_out)), exist_ok=True)
|
||
with open(args.json_out, "w", encoding="utf-8") as handle:
|
||
json.dump(
|
||
{
|
||
"sprint": sprint,
|
||
"team": team["TeamName"],
|
||
"team_id": team["TeamId"],
|
||
"user": {"RealName": user["RealName"], "GuidId": user["GuidId"]},
|
||
"rows": [row.__dict__ for row in rows],
|
||
"timing": {
|
||
"login": round(t_login, 3),
|
||
"teams": round(t_teams, 3),
|
||
"projects": round(t_projects, 3),
|
||
"scan": round(t_scan, 3),
|
||
},
|
||
},
|
||
handle,
|
||
ensure_ascii=False,
|
||
indent=2,
|
||
)
|
||
print(f"\n结果已写入 {args.json_out}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
sys.exit(main())
|
||
except QtError as error:
|
||
print(f"[ERROR] {error}", file=sys.stderr)
|
||
sys.exit(2)
|