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 个单测通过。
59 lines
2.2 KiB
Python
59 lines
2.2 KiB
Python
"""异常与错误分类。
|
||
|
||
WCF 的错误是 HTML 页面，真正的信息藏在 "The exception message is '...'" 里。
|
||
分类的目的是决定「重试 / 跳过 / 整批中止」，而不是把所有失败都当网络问题重试。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from enum import StrEnum
|
||
|
||
|
||
class Category(StrEnum):
|
||
NET = "NET" # 网络层：可重试
|
||
SP_MISSING = "SP_MISSING" # 存储过程不存在：配置/版本问题，不重试
|
||
PARAM = "PARAM" # 参数错误/类型转换：数据问题，不重试
|
||
EXCEL = "EXCEL" # Excel 解析失败（Column1 != 0）
|
||
TFS = "TFS" # 推送 TFS 失败
|
||
AUTH = "AUTH" # 登录失败
|
||
UNKNOWN = "UNKNOWN"
|
||
|
||
|
||
class QtError(RuntimeError):
|
||
"""带分类的接口错误。"""
|
||
|
||
def __init__(self, message: str, category: Category = Category.UNKNOWN, *, retryable: bool = False):
|
||
super().__init__(message)
|
||
self.message = message
|
||
self.category = category
|
||
self.retryable = retryable
|
||
|
||
def __str__(self) -> str: # pragma: no cover - 直通
|
||
return f"[{self.category}] {self.message}"
|
||
|
||
@classmethod
|
||
def from_http(cls, status: int, body: str) -> "QtError":
|
||
"""把 WCF 的 HTML 错误页转成可读错误。"""
|
||
message = extract_exception_message(body) or f"HTTP {status}"
|
||
lowered = message.lower()
|
||
if "could not find stored procedure" in lowered:
|
||
return cls(message, Category.SP_MISSING)
|
||
if "missing parameter" in lowered or "conversion" in lowered or "procedure or function" in lowered:
|
||
return cls(message, Category.PARAM)
|
||
if status >= 500:
|
||
return cls(message, Category.NET, retryable=True)
|
||
return cls(message, Category.UNKNOWN)
|
||
|
||
|
||
def extract_exception_message(body: str) -> str:
|
||
"""从 WCF 错误页里抠出真正的异常文本。"""
|
||
import html
|
||
import re
|
||
|
||
match = re.search(r"exception message is '(.*?)'\s*\.?\s*See server logs", body, re.S)
|
||
if match:
|
||
return html.unescape(match.group(1)).strip()
|
||
plain = re.sub(r"<[^>]+>", " ", body)
|
||
plain = html.unescape(re.sub(r"\s+", " ", plain)).strip()
|
||
return plain[:400]
|