diff --git a/Dockerfile b/Dockerfile index 58f5713..d7a656a 100644 --- a/Dockerfile +++ b/Dockerfile @@ -13,11 +13,25 @@ RUN CGO_ENABLED=0 GOOS=linux go build -trimpath -o /out/api ./cmd/api \ FROM alpine:3.22 ENV PYTHONDONTWRITEBYTECODE=1 -RUN apk add --no-cache ca-certificates python3 postgresql-client \ +RUN apk add --no-cache ca-certificates python3 postgresql-client curl gcompat libc6-compat \ && addgroup -g 65532 governance \ && adduser -D -H -u 65532 -G governance governance +# dws (DingTalk Workspace CLI) v1.0.63：版本+校验和固定（禁用 dws upgrade）。 +# 官方产物是动态链接 glibc 的二进制，alpine(musl) 靠 gcompat/libc6-compat 运行， +# 已验证端到端取件并与 glibc 原生环境字节一致。构建期 `dws version` 即自检。 +ARG DWS_VERSION=v1.0.63 +ARG DWS_SHA256=78cab668bf671c17128ac55347627fb28f80c15b9c30c5f86e32e2b5ee6e34bc +RUN set -eux; \ + curl -fsSL -o /tmp/dws.tgz \ + "https://github.com/DingTalk-Real-AI/dingtalk-workspace-cli/releases/download/${DWS_VERSION}/dws-linux-amd64.tar.gz"; \ + echo "${DWS_SHA256} /tmp/dws.tgz" | sha256sum -c -; \ + tar -xzf /tmp/dws.tgz -C /tmp ./dws; \ + install -m 0755 /tmp/dws /usr/local/bin/dws; \ + rm -rf /tmp/dws.tgz /tmp/dws; \ + dws version WORKDIR /app COPY --from=build /out/ /app/ -COPY scripts/source_catalog_import.py scripts/live_catalog_refresh.py scripts/artifact_fetch.py /app/scripts/ +COPY scripts/source_catalog_import.py scripts/live_catalog_refresh.py scripts/artifact_fetch.py \ + scripts/axls_to_xlsx.py scripts/dws_error_map.py /app/scripts/ USER 65532:65532 CMD ["/app/api"] diff --git a/compose.yaml b/compose.yaml index c71d6f9..8cd0f33 100644 --- a/compose.yaml +++ b/compose.yaml @@ -93,6 +93,15 @@ services: TEST_WEKNORA_STALLED_AFTER: "${TEST_WEKNORA_STALLED_AFTER:-30m}" TEST_WEKNORA_MAX_UPLOAD_BYTES: "${TEST_WEKNORA_MAX_UPLOAD_BYTES:-52428800}" TEST_ACCEPTANCE_POLICY_VERSION: "${TEST_ACCEPTANCE_POLICY_VERSION:-structural-v1}" + # DingTalk 取件走 dws（默认优先 dws，可重试失败回落老路径）：dws|auto|legacy + ARTIFACT_DINGTALK_FETCHER: "${ARTIFACT_DINGTALK_FETCHER:-auto}" + DWS_BIN: /usr/local/bin/dws + DWS_CONFIG_DIR: /data/dws + # dws 的文件型 keychain 在 $HOME/.local/share/dws-cli（XDG_DATA_HOME 不被识别）， + # 仅给 dws 子进程换 HOME，不动整个 worker 的 HOME。 + DWS_HOME: /data/home + DWS_DISABLE_KEYCHAIN: "1" + SSL_CERT_FILE: /etc/ssl/certs/ca-certificates.crt secrets: - pgpass - test_weknora_api_key @@ -101,6 +110,10 @@ services: - ./spool/ingestion:/var/lib/governance/ingestion - "${PP_SCAN_HOST_DATA_ROOT:-/home/zheromiku/dobot-portal-sync/data}:${PP_SCAN_CONTAINER_DATA_ROOT:-/home/zheromiku/dobot-portal-sync/data}:ro" - /home/zheromiku/dingtalk_kb_download:/sources/dingtalk:ro + # dws 凭据：与宿主共享同一份 state（refresh token 是轮换制，不能在两处各留一份） + - /home/zheromiku/.dws:/data/dws + # dws 的文件型 keychain：必须有 $HOME/.local/share/dws-cli，故挂到 DWS_HOME 下 + - /home/zheromiku/.local/share/dws-cli:/data/home/.local/share/dws-cli - /home/zheromiku/tmp/alist:/sources/alist:ro - /home/zheromiku/dobot-portal-sync:/sources/pp:ro - ./secrets/dingtalk_capture.json:/run/source-secrets/dingtalk_capture.json:ro diff --git a/docs/DINGTALK_DWS_FETCHER.md b/docs/DINGTALK_DWS_FETCHER.md new file mode 100644 index 0000000..69b1e3a --- /dev/null +++ b/docs/DINGTALK_DWS_FETCHER.md @@ -0,0 +1,78 @@ +# DingTalk 取件走 dws（DingTalk Workspace CLI） + +默认**优先使用 dws** 取钉钉知识库字节，老路径（本地镜像树 + 浏览器 a-token 实况导出）作为回退保留。 + +## 1. 开关 + +| 环境变量 | 取值 | 含义 | +|---|---|---| +| `ARTIFACT_DINGTALK_FETCHER` | `auto`（默认） | 先 dws；**仅当** dws 返回可重试错误（429/5xx/超时/连接层/dws 二进制缺失）时回落到老路径 | +| | `dws` | 只用 dws，失败即失败（灰度验证/压测用） | +| | `legacy` | 完全绕过 dws，走原逻辑（回退开关） | +| `DWS_BIN` | 默认 `/usr/local/bin/dws` | dws 可执行文件 | +| `DWS_CONFIG_DIR` | 容器内 `/data/dws` | dws 配置目录（profiles/identity） | +| `DWS_HOME` | 容器内 `/data/home` | **dws 子进程专用 HOME**：文件型 keychain 必须位于 `$HOME/.local/share/dws-cli`（`XDG_DATA_HOME` 不被识别），只给 dws 子进程换 HOME，不动 worker 全局 HOME | +| `DWS_DISABLE_KEYCHAIN` | `1` | 无系统 keychain 的环境用 file-DEK 模式 | + +## 2. 路由 + +按**已解析文件名后缀**路由（与老路径判定一致；manifest 无 `declared_type`，缺失时依次退回 `declared_type` → `dws drive +inspect`）： + +| 后缀 | 取件方式 | 产物 | +|---|---|---| +| `adoc` | `dws doc export --export-format docx` | docx | +| `axls` | `dws sheet list` + `sheet range read` → `scripts/axls_to_xlsx.py` 落盘 | xlsx | +| 其余可入库类型 | `dws drive +download` | 原后缀 | +| `dlink` / `hlink` | 直接 `missing`（链接节点 fileSize=0） | — | +| `amind` / `appt` | `unsupported_format`（CLI 无内容导出入口，与老路径等价） | — | + +**平台侧判定一律保留**：`pick_suffix`、`SLIDES`（PPT 必须 PDF 双胞胎）、magic 校验、大小上限、`sha256`、spool 原子写、`exported_doc_is_blank()` 空壳检查。dws 只负责“取字节”。 + +## 3. 错误映射 + +`scripts/dws_error_map.py` 把 dws 的错误体（严格版 `ok/error` 包裹、普通子命令 `error` 对象、本地错误、TLS/网络文本）统一映射为平台语义 `{ok, code, retryable, missing}`。要点： + +- 只有明确瞬时信号（429/5xx/超时/连接层失败/**dws 二进制缺失**）为 `retryable=true`； +- 未分类错误 `code=unknown` 且**不重试**——避免被无限重试掩盖真因； +- `workspace node not found` → `missing=true`（走 missing 分支而非 failed）； +- TLS 缺 CA（`x509: certificate signed by unknown authority`）→ `tls_ca`，终态（重试无意义，需补 CA）。 + +覆盖样本与断言见 `scripts/test_dws_error_map.py`（17 个断言，样本为实测捕获的真实错误体）。 + +## 4. 镜像与部署 + +- 镜像：`alpine:3.22` + `gcompat libc6-compat ca-certificates`（官方 dws 是动态链接 glibc 的二进制，靠 gcompat 在 musl 上运行；已验证端到端字节一致）；dws **版本与 sha256 固定在 Dockerfile 的 `DWS_VERSION`/`DWS_SHA256`**，**禁用 `dws upgrade`**。 +- 凭据：**与宿主共享同一份 state**（refresh token 是轮换制，两个副本会互相顶掉）： + - `~/.dws` → `/data/dws`（`DWS_CONFIG_DIR`） + - `~/.local/share/dws-cli` → `/data/home/.local/share/dws-cli`（`DWS_HOME`） +- 容器必须 `--user <宿主 uid>:`（worker 已是 `1000:1000`），否则 state 被写成 root 所有。 + +## 5. 续期 Runbook（refresh token 30 天） + +```bash +# 到期前 7 天（阈值告警）在目标机执行，本人完成浏览器授权 +dws auth login --device +# 验证必须真实调用，而不是只看 authenticated 字段 +dws wiki +space-list --limit 1 -f json # 期望 ok:true +docker exec dws auth status --readonly -f json +``` + +出现 `code=credentials` 时**立即停批**（不要重试风暴），重新授权后 `--retry-failed` 续跑。健康面板建议展示 `refresh_expires_at`：<7 天黄、<3 天红。 + +## 6. 回退演练（上线前必做） + +```bash +docker compose exec worker env ARTIFACT_DINGTALK_FETCHER=legacy ... # 或改 compose 环境变量后重建 worker +# 取同一批节点 → 断言 spool_ref/sha256 契约不变、无新增 code=* → 再切回 auto +``` + +保留 `dingtalk-login` sidecar 与 63GB 镜像树至少 2 周（镜像树可降级为只读热备）。 + +## 7. 验证证据（2026-10-09） + +| 项 | 结果 | +|---|---| +| 影子比对（真实 worker 容器 vs 宿主 dws） | adoc 24/24 等价（正文相似度 1.0、media 逐项相等、11/24 字节完全相同）；普通文件 5/5 `sha256` 一致；axls 归一化后值覆盖 100%/100% | +| 容器端到端（本改动 + 真容器约束 `--read-only --tmpfs /tmp --user 1000:1000`） | adoc/docx 305,078B（sha `77b393c3…`，与平台基准一致）、axls/xlsx 3,330B、pdf 27,140B → **3/3 成功**；结果 JSON 字段与改动前完全一致 | +| 单元测试 | `scripts/artifact_fetch_test.py` 14 项、`scripts/artifact_fetch_blank_doc_test.py` 5 项、`scripts/test_dws_error_map.py` 17 断言 —— 全绿 | +| 大表能力 | 平台老路径失败的 220MB/15 表/3,658 行 axls，dws + 转换器 18s 产出 128KB 有效 xlsx | \ No newline at end of file diff --git a/scripts/artifact_fetch.py b/scripts/artifact_fetch.py index aa6bdf7..c38cabd 100644 --- a/scripts/artifact_fetch.py +++ b/scripts/artifact_fetch.py @@ -34,6 +34,24 @@ import urllib.parse import urllib.request +# Sibling modules shipped into /app/scripts next to this file. +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import axls_to_xlsx # noqa: E402 +import dws_error_map # noqa: E402 + + +DWS_BIN = os.environ.get("DWS_BIN") or "/usr/local/bin/dws" +# dws 在文件型 keychain 模式下把凭据放在 $HOME/.local/share/dws-cli（XDG_DATA_HOME 不被识别）， +# 所以给 dws 子进程单独指定一个 HOME，避免动整个 worker 的 HOME。 +DWS_HOME = os.environ.get("DWS_HOME") or "" +# dws -> only the dws CLI; auto -> dws first, fall back to the legacy path on +# retryable failures; legacy -> only the captured-header web path. +DINGTALK_FETCHER = (os.environ.get("ARTIFACT_DINGTALK_FETCHER") or "auto").strip().lower() +DINGTALK_FETCHER_MODES = ("dws", "auto", "legacy") +if DINGTALK_FETCHER not in DINGTALK_FETCHER_MODES: + DINGTALK_FETCHER = "auto" + + SUPPORTED = { "pdf": "pdf", "docx": "docx", "xlsx": "xlsx", "md": "md", "txt": "txt", "csv": "csv", "json": "json", @@ -380,6 +398,118 @@ def _space_key(roots, name): return "" +def _dws_binary(): + if os.path.exists(DWS_BIN) and os.access(DWS_BIN, os.X_OK): + return DWS_BIN + return shutil.which("dws") or "" + + +def dws_run(args, timeout=300, cwd=None): + """Run the dws CLI and translate its result into the platform's semantics.""" + binary = _dws_binary() + if not binary: + # Infrastructure missing: auto mode must be able to fall back, so this is + # marked retryable rather than terminal. + return None, {"ok": False, "code": "dws_missing", "retryable": True, "missing": False, + "error": "dws binary not found (DWS_BIN=%s)" % DWS_BIN} + try: + env = dict(os.environ) + if DWS_HOME: + env["HOME"] = DWS_HOME + completed = subprocess.run([binary, *args], capture_output=True, text=True, + timeout=timeout, cwd=cwd, env=env) + except subprocess.TimeoutExpired: + return None, {"ok": False, "code": "export_timeout", "retryable": True, "missing": False, + "error": "dws call timed out: %s" % " ".join(args[:3])} + payload = None + if completed.stdout.strip(): + try: + payload = json.loads(completed.stdout) + except ValueError: + payload = None + verdict = dws_error_map.classify(payload, stderr=completed.stderr, exit_code=completed.returncode) + return payload, verdict + + +def _dws_error(verdict): + if verdict.get("missing"): + return MissingSource(verdict.get("error") or "DingTalk node missing on the source side") + return FetchError(verdict.get("error") or "dws fetch failed", + retryable=bool(verdict.get("retryable")), + code=verdict.get("code") or "unknown") + + +def _dws_inspect_extension(raw_key): + """Last-resort extension lookup (only when the manifest carries no usable suffix).""" + payload, verdict = dws_run(["drive", "+inspect", "--node", raw_key], timeout=60) + if not verdict["ok"] or not isinstance(payload, dict): + return "" + found = [] + + def walk(node, depth=0): + if depth > 5 or found: + return + if isinstance(node, dict): + for key, value in node.items(): + if key == "extension" and isinstance(value, str) and value: + found.append(value.lower()) + return + walk(value, depth + 1) + elif isinstance(node, list): + for value in node[:10]: + walk(value, depth + 1) + + walk(payload.get("data") if "data" in payload else payload) + return found[0] if found else "" + + +def dingtalk_dws_download(manifest, max_bytes): + """Fetch one DingTalk node through dws: adoc -> docx export, axls -> converted + xlsx, everything else -> raw drive download. Returns (path, size, digest, matched_local).""" + catalog_size = manifest.get("size_bytes") + if isinstance(catalog_size, int) and catalog_size > max_bytes: + # The catalog already knows the size: refuse before burning a download. + raise FetchError("artifact exceeds ingest size limit", retryable=False, code="over_limit") + raw_key, _dentry_type = dingtalk_raw_keys(manifest) + # The node type comes from the RESOLVED name suffix first (that is what the legacy + # path keys on); declared_type is an optional hint and dingtalk_raw_keys returns a + # dentry type ("file"), never a file extension. + raw_suffix = pathlib.PurePosixPath(manifest["name"]).suffix.lower().lstrip(".") + extension = raw_suffix + if not extension or extension in ("file", "folder"): + declared = (manifest.get("declared_type") or "").lower() + extension = declared if declared not in ("", "file", "folder") else _dws_inspect_extension(raw_key) + extension = (extension or "").lower().lstrip(".") + if extension in ("dlink", "hlink"): + raise MissingSource("DingTalk link node has no bytes (fileSize=0)") + if extension in ("amind", "appt"): + raise FetchError("dws has no content-export entry for .%s nodes" % extension, + retryable=False, code="unsupported_format") + workdir = pathlib.Path(tempfile.mkdtemp(prefix="dws-fetch-")) + if extension == "adoc": + target = "out.docx" + _, verdict = dws_run(["doc", "export", "--node", raw_key, "--export-format", "docx", + "--output", target], cwd=str(workdir)) + if not verdict["ok"]: + raise _dws_error(verdict) + elif extension == "axls": + target = "out.xlsx" + axls_to_xlsx.build_xlsx(axls_to_xlsx.fetch_sheets(raw_key), str(workdir / target)) + else: + target = "out." + (extension or "bin") + _, verdict = dws_run(["drive", "+download", "--node", raw_key, "--output", target], + cwd=str(workdir), timeout=900) + if not verdict["ok"]: + raise _dws_error(verdict) + exported = workdir / target + if not exported.is_file() or exported.stat().st_size == 0: + raise FetchError("dws produced no usable bytes", retryable=True, code="dws_empty_output") + size = exported.stat().st_size + if size > max_bytes: + raise FetchError("artifact exceeds ingest size limit", retryable=False, code="over_limit") + return str(exported), size, sha256_file(str(exported)), False + + def dingtalk_download(manifest, max_bytes): spreadsheet = pathlib.PurePosixPath(manifest["name"]).suffix.lower() == ".axls" @@ -390,23 +520,38 @@ def dingtalk_download(manifest, max_bytes): retryable=False, code="empty_document") return local_path + # dws first (this is the configured preference); the captured-header path stays + # as the fallback for auto mode and as the only path in legacy mode. + if DINGTALK_FETCHER in ("dws", "auto"): + try: + path, size, digest, matched = dingtalk_dws_download(manifest, max_bytes) + return guarded(path), size, digest, matched, "dws" + except MissingSource: + raise + except FetchError as error: + if DINGTALK_FETCHER == "dws" or not error.retryable: + raise + # auto: only retryable failures fall back, terminal ones must surface as-is + if DINGTALK_FETCHER == "dws": + raise FetchError("dws fetcher requested but unavailable", retryable=False, code="dws_missing") + export = dingtalk_local_export(manifest) modified = str(manifest.get("modified_at") or "") if export: path, fresh = export size = os.path.getsize(path) if size > 0 and (fresh or not modified): - return guarded(path), size, sha256_file(path), True + return guarded(path), size, sha256_file(path), True, "legacy" # stale local export -> fall through to a live export when available raw_key, declared_type = dingtalk_raw_keys(manifest) headers = load_capture_headers() live_path = live_export_dingtalk_document(raw_key, headers, manifest, max_bytes) if export and not live_path: path, fresh = export - return guarded(path), os.path.getsize(path), sha256_file(path), True # keep old bytes; staleness shown in UI + return guarded(path), os.path.getsize(path), sha256_file(path), True, "legacy" # keep old bytes; staleness shown in UI if not live_path: raise MissingSource("DingTalk file is not exported and cannot be exported automatically") - return live_path + return live_path[0], live_path[1], live_path[2], live_path[3], "legacy" def dingtalk_json_api(url, headers, payload=None, extra_headers=None, method=None): @@ -602,7 +747,10 @@ def main(): elif manifest["source_type"] == "pp": path, size, digest, matched_local = pp_download(manifest, max_bytes) elif manifest["source_type"] == "dingtalk": - path, size, digest, matched_local = dingtalk_download(manifest, max_bytes) + path, size, digest, matched_local, fetcher = dingtalk_download(manifest, max_bytes) + # Only stderr: the Go runner decodes stdout with DisallowUnknownFields, so the + # result JSON shape must stay exactly as before. + print("[artifact_fetch] dingtalk fetcher=%s node=%s" % (fetcher, manifest.get("node_id")), file=sys.stderr) else: raise FetchError("unknown source type", retryable=False, code="bad_manifest") # Type is decided on the RESOLVED bytes, not the catalog title: PP ledger diff --git a/scripts/artifact_fetch_test.py b/scripts/artifact_fetch_test.py index f5b2d9a..e5f80d0 100644 --- a/scripts/artifact_fetch_test.py +++ b/scripts/artifact_fetch_test.py @@ -1,5 +1,8 @@ import importlib.util +import json +import os import pathlib +import sys import unittest from unittest import mock @@ -66,5 +69,194 @@ class DingTalkOnlineDocumentExportTest(unittest.TestCase): download.assert_called_once_with("https://oss.example/result.docx", {}, 180, 1_000_000) +class DingTalkDwsFetcherTest(unittest.TestCase): + """dws 取件器：路由、回落语义、以及"结果 JSON 形状不得变化"的契约锁定。""" + + def setUp(self): + self.manifest = { + "spec_id": "spec-1", "connection_id": "conn-1", "node_id": "sn-test", + "source_type": "dingtalk", "source_name": "公司级技术支持知识库", + "name": "说明.adoc", "relative_path": "公司级技术支持知识库/说明.adoc", + "declared_type": "adoc", "size_bytes": 1000, "modified_at": None, "locator_hash": "loc", + } + self.ok_verdict = {"ok": True, "code": "", "retryable": False, "missing": False, "error": ""} + + def _fail(self, code, retryable, missing=False): + return {"ok": False, "code": code, "retryable": retryable, "missing": missing, "error": code} + + def test_dws_adoc_export_is_preferred_and_reported(self): + def fake_run(args, timeout=300, cwd=None): + self.assertEqual(args[:2], ["doc", "export"]) + self.assertIn("--export-format", args) + pathlib.Path(cwd, "out.docx").write_bytes(b"PK\x03\x04" + b"x" * 100) + return {"success": True}, self.ok_verdict + + with mock.patch.object(artifact_fetch, "DINGTALK_FETCHER", "dws"), \ + mock.patch.object(artifact_fetch, "dingtalk_raw_keys", return_value=("raw-1", "adoc")), \ + mock.patch.object(artifact_fetch, "dws_run", side_effect=fake_run), \ + mock.patch.object(artifact_fetch, "exported_doc_is_blank", return_value=False): + path, size, digest, matched, fetcher = artifact_fetch.dingtalk_download(self.manifest, 1_000_000) + + self.assertEqual(fetcher, "dws") + self.assertFalse(matched) + self.assertEqual(size, 104) + self.assertEqual(digest, artifact_fetch.sha256_file(path)) + + def test_axls_is_routed_through_the_converter(self): + built = {} + + def build(sheets, out): + built["sheets"] = sheets + pathlib.Path(out).write_bytes(b"PK\x03\x04" + b"y" * 60) + return {"sheets": ["S1"]} + + with mock.patch.object(artifact_fetch, "DINGTALK_FETCHER", "dws"), \ + mock.patch.object(artifact_fetch, "dingtalk_raw_keys", return_value=("raw-2", "axls")), \ + mock.patch.object(artifact_fetch.axls_to_xlsx, "fetch_sheets", return_value=[{"name": "S1", "cells": []}]), \ + mock.patch.object(artifact_fetch.axls_to_xlsx, "build_xlsx", side_effect=build), \ + mock.patch.object(artifact_fetch, "dws_run") as dws_run, \ + mock.patch.object(artifact_fetch, "exported_doc_is_blank", return_value=False): + path, size, digest, matched, fetcher = artifact_fetch.dingtalk_download( + {**self.manifest, "name": "汇总.axls", "declared_type": "axls"}, 1_000_000) + + self.assertEqual(fetcher, "dws") + self.assertTrue(path.endswith("out.xlsx")) + self.assertFalse(dws_run.called) # 转换器内部自己调 dws，取件器不再直接调 + self.assertEqual(built["sheets"], [{"name": "S1", "cells": []}]) + + def test_auto_falls_back_to_legacy_on_retryable_failure(self): + with mock.patch.object(artifact_fetch, "DINGTALK_FETCHER", "auto"), \ + mock.patch.object(artifact_fetch, "dingtalk_dws_download", + side_effect=artifact_fetch.FetchError("dws missing", retryable=True, code="dws_missing")), \ + mock.patch.object(artifact_fetch, "dingtalk_local_export", return_value=("/tmp/legacy.docx", True)), \ + mock.patch.object(artifact_fetch, "exported_doc_is_blank", return_value=False), \ + mock.patch.object(artifact_fetch, "sha256_file", return_value="b" * 64), \ + mock.patch.object(artifact_fetch.os.path, "getsize", return_value=42): + path, size, digest, matched, fetcher = artifact_fetch.dingtalk_download(self.manifest, 1_000_000) + + self.assertEqual((path, size, matched, fetcher), ("/tmp/legacy.docx", 42, True, "legacy")) + + def test_auto_does_not_fall_back_on_terminal_failure(self): + with mock.patch.object(artifact_fetch, "DINGTALK_FETCHER", "auto"), \ + mock.patch.object(artifact_fetch, "dingtalk_dws_download", + side_effect=artifact_fetch.FetchError("nope", retryable=False, code="forbidden")), \ + mock.patch.object(artifact_fetch, "dingtalk_local_export") as legacy: + with self.assertRaises(artifact_fetch.FetchError) as raised: + artifact_fetch.dingtalk_download(self.manifest, 1_000_000) + + self.assertEqual(raised.exception.code, "forbidden") + self.assertFalse(legacy.called) # 终态错误不得回落（否则掩盖真因） + + def test_dws_mode_never_falls_back(self): + with mock.patch.object(artifact_fetch, "DINGTALK_FETCHER", "dws"), \ + mock.patch.object(artifact_fetch, "dingtalk_dws_download", + side_effect=artifact_fetch.FetchError("dws missing", retryable=True, code="dws_missing")), \ + mock.patch.object(artifact_fetch, "dingtalk_local_export") as legacy: + with self.assertRaises(artifact_fetch.FetchError) as raised: + artifact_fetch.dingtalk_download(self.manifest, 1_000_000) + + self.assertEqual(raised.exception.code, "dws_missing") + self.assertFalse(legacy.called) + + def test_legacy_mode_skips_dws_entirely(self): + with mock.patch.object(artifact_fetch, "DINGTALK_FETCHER", "legacy"), \ + mock.patch.object(artifact_fetch, "dingtalk_dws_download") as dws_download, \ + mock.patch.object(artifact_fetch, "dingtalk_local_export", return_value=("/tmp/legacy.docx", True)), \ + mock.patch.object(artifact_fetch, "exported_doc_is_blank", return_value=False), \ + mock.patch.object(artifact_fetch, "sha256_file", return_value="c" * 64), \ + mock.patch.object(artifact_fetch.os.path, "getsize", return_value=7): + path, size, digest, matched, fetcher = artifact_fetch.dingtalk_download(self.manifest, 1_000_000) + + self.assertEqual(fetcher, "legacy") + self.assertFalse(dws_download.called) + + def test_catalog_size_over_limit_never_starts_a_download(self): + with mock.patch.object(artifact_fetch, "DINGTALK_FETCHER", "dws"), \ + mock.patch.object(artifact_fetch, "dingtalk_raw_keys") as raw_keys, \ + mock.patch.object(artifact_fetch, "dws_run") as dws_run: + with self.assertRaises(artifact_fetch.FetchError) as raised: + artifact_fetch.dingtalk_download({**self.manifest, "size_bytes": 500_000_000}, 1_000_000) + + self.assertEqual(raised.exception.code, "over_limit") + self.assertFalse(raw_keys.called) + self.assertFalse(dws_run.called) + + def test_link_and_mind_nodes_map_to_missing_and_unsupported(self): + cases = (("dlink", "链接节点.dlink", "missing", True), + ("amind", "培训大纲.amind", "unsupported_format", False)) + for declared, name, expected, expected_missing in cases: + with mock.patch.object(artifact_fetch, "DINGTALK_FETCHER", "dws"), \ + mock.patch.object(artifact_fetch, "dingtalk_raw_keys", return_value=("raw-3", "file")), \ + mock.patch.object(artifact_fetch, "dws_run") as dws_run: + with self.assertRaises((artifact_fetch.FetchError, artifact_fetch.MissingSource)) as raised: + artifact_fetch.dingtalk_download( + {**self.manifest, "name": name, "declared_type": declared}, 1_000_000) + if expected_missing: + self.assertIsInstance(raised.exception, artifact_fetch.MissingSource) + else: + self.assertEqual(raised.exception.code, expected) + self.assertFalse(dws_run.called) + + def test_routing_uses_name_suffix_without_declared_type(self): + # 生产 manifest 里没有 declared_type 字段；必须靠文件名后缀路由（与老路径一致） + def fake_run(args, timeout=300, cwd=None): + self.assertEqual(args[:2], ["doc", "export"]) + pathlib.Path(cwd, "out.docx").write_bytes(b"PK\x03\x04" + b"q" * 20) + return {"success": True}, self.ok_verdict + + manifest = {k: v for k, v in self.manifest.items() if k != "declared_type"} + with mock.patch.object(artifact_fetch, "DINGTALK_FETCHER", "dws"), \ + mock.patch.object(artifact_fetch, "dingtalk_raw_keys", return_value=("raw-9", "file")), \ + mock.patch.object(artifact_fetch, "dws_run", side_effect=fake_run), \ + mock.patch.object(artifact_fetch, "exported_doc_is_blank", return_value=False): + path, size, digest, matched, fetcher = artifact_fetch.dingtalk_download(manifest, 1_000_000) + + self.assertEqual(fetcher, "dws") + self.assertTrue(path.endswith("out.docx")) + + def test_routing_falls_back_to_dws_inspect_when_name_has_no_suffix(self): + calls = [] + + def fake_run(args, timeout=300, cwd=None): + calls.append(args) + if args[:2] == ["drive", "+inspect"]: + return {"ok": True, "data": {"data": {"file": {"extension": "adoc", "name": "无后缀节点"}}}}, self.ok_verdict + pathlib.Path(cwd, "out.docx").write_bytes(b"PK\x03\x04" + b"w" * 30) + return {"success": True}, self.ok_verdict + + with mock.patch.object(artifact_fetch, "DINGTALK_FETCHER", "dws"), \ + mock.patch.object(artifact_fetch, "dingtalk_raw_keys", return_value=("raw-10", "file")), \ + mock.patch.object(artifact_fetch, "dws_run", side_effect=fake_run), \ + mock.patch.object(artifact_fetch, "exported_doc_is_blank", return_value=False): + path, size, digest, matched, fetcher = artifact_fetch.dingtalk_download( + {k: v for k, v in self.manifest.items() if k != "declared_type"} | {"name": "无后缀节点"}, 1_000_000) + + self.assertEqual(calls[0][:2], ["drive", "+inspect"]) + self.assertEqual(calls[1][:2], ["doc", "export"]) + self.assertEqual(fetcher, "dws") + + def test_result_json_shape_is_unchanged(self): + # 平台 Go runner 用 DisallowUnknownFields 解析 stdout：多一个字段就会让所有取件失败。 + captured = [] + temp = pathlib.Path("/tmp/kg-artifact-fetch-shape.docx") + temp.write_bytes(b"PK\x03\x04" + b"z" * 200) + try: + with mock.patch.object(artifact_fetch, "emit", side_effect=captured.append), \ + mock.patch.object(artifact_fetch, "dingtalk_download", + return_value=(str(temp), temp.stat().st_size, + artifact_fetch.sha256_file(str(temp)), False, "dws")), \ + mock.patch.object(sys, "argv", ["artifact_fetch.py", json.dumps(self.manifest)]), \ + mock.patch.dict(os.environ, {"INGESTION_SPOOL_ROOT": "/tmp/kg-shape-spool", + "ARTIFACT_FETCH_MAX_BYTES": "100000"}): + artifact_fetch.main() + finally: + temp.unlink(missing_ok=True) + + self.assertEqual(len(captured), 1) + self.assertEqual(set(captured[0].keys()), { + "ok", "spec_id", "node_id", "connection_id", "sha256", "size_bytes", + "spool_ref", "validated_format", "matched_local"}) + + if __name__ == "__main__": unittest.main() diff --git a/scripts/axls_to_xlsx.py b/scripts/axls_to_xlsx.py new file mode 100644 index 0000000..920becb --- /dev/null +++ b/scripts/axls_to_xlsx.py @@ -0,0 +1,189 @@ +#!/usr/bin/env python3 +"""dws axls → xlsx 转换器（纯标准库，无 openpyxl）。 + +用途：dws 没有 axls 直出接口，只能 sheet range read 拿单元格；本脚本把单元格写成 +合法的 .xlsx，供平台 artifact_fetch 的 dingtalk dws 取件器落盘入库。 + +兼容平台既有检查： + - `exported_doc_is_blank()` 对 spreadsheet 只读 `xl/sharedStrings.xml` → 本脚本**把文本写进 sharedStrings**， + 否则会被判定为"空壳文档"而拒收。 + - 必须 PK magic、能被子解析器识别为 xlsx。 + +归一化约定（与平台既有 xlsx 产物对齐）： + - 单元格换行 CRLF/CR → LF + - 日期保持 dws 返回的显示文本（平台旧产物是 Excel 序列号，转换后语义一致） + +dws 可执行文件解析顺序：$DWS_BIN → PATH 里的 dws。 +""" +import argparse, json, os, re, shutil, subprocess, sys, zipfile +from xml.sax.saxutils import escape + +NUM_RE = re.compile(r"^-?\d+(\.\d+)?$") + + +def dws_binary(): + return os.environ.get("DWS_BIN") or shutil.which("dws") or "/usr/local/bin/dws" + + +def run_json(args, timeout=180): + # dws 的文件型 keychain 在 $HOME/.local/share/dws-cli，容器里用 DWS_HOME 单独指定 + env = dict(os.environ) + dws_home = os.environ.get("DWS_HOME") + if dws_home: + env["HOME"] = dws_home + p = subprocess.run([dws_binary(), *args, "-f", "json"], capture_output=True, text=True, + timeout=timeout, env=env) + try: + return json.loads(p.stdout) + except Exception: + raise SystemExit(f"dws 调用失败: {' '.join(args)} :: {(p.stdout or p.stderr)[:300]}") + + +def fetch_sheets(node): + """返回 [{'name':str,'cells':[[...]]}]""" + listing = run_json(["sheet", "list", "--node", node]) + sheets = listing.get("sheets") or [] + if not sheets: + raise SystemExit(f"sheet list 无工作表: {json.dumps(listing, ensure_ascii=False)[:200]}") + out = [] + for s in sheets: + sid = s.get("sheetId") or s.get("name") + r = run_json(["sheet", "range", "read", "--node", node, "--sheet-id", sid]) + out.append({"name": s.get("name") or sid, "cells": r.get("cells") or []}) + return out + + +def norm(v): + if v is None: + return None + s = str(v) + if "\r" in s: + s = s.replace("\r\n", "\n").replace("\r", "\n") + return s + + +def col_letter(idx): + """0-based 列号 → A/B/.../AA""" + s = "" + idx += 1 + while idx: + idx, rem = divmod(idx - 1, 26) + s = chr(65 + rem) + s + return s + + +def sheet_name(name, used): + n = re.sub(r"[\[\]:*?/\\]", "_", str(name or "Sheet")).strip()[:31] or "Sheet" + base, i = n, 1 + while n.lower() in used: + suffix = f"_{i}" + n = base[: 31 - len(suffix)] + suffix + i += 1 + used.add(n.lower()) + return n + + +def build_xlsx(sheets, path): + # 1) 收集共享字符串 + 每表 XML + shared, shared_idx = [], {} + def sidx(text): + if text not in shared_idx: + shared_idx[text] = len(shared) + shared.append(text) + return shared_idx[text] + + used, sheet_xml, names = set(), [], [] + for s in sheets: + sname = sheet_name(s.get("name"), used) + names.append(sname) + rows = s.get("cells") or [] + chunks = [] + for ri, row in enumerate(rows, start=1): + cells = [] + for ci, cell in enumerate(row): + v = cell.get("value") if isinstance(cell, dict) else cell + v = norm(v) + if v is None or v == "": + continue + ref = f"{col_letter(ci)}{ri}" + if NUM_RE.match(v): + cells.append(f'{v}') + else: + cells.append(f'{sidx(v)}') + if cells: + chunks.append(f'{"".join(cells)}') + sheet_xml.append( + '' + '' + f'{"".join(chunks)}') + + n = len(names) + ct = ('' + '' + '' + '' + '' + '' + '' + + "".join(f'' for i in range(n)) + + '') + rels = ('' + '' + '' + '') + wb = ('' + '' + + "".join(f'' for i in range(n)) + + '') + wb_rels = ('' + '' + + "".join(f'' for i in range(n)) + + f'' + + f'' + + '') + styles = ('' + '' + '' + '' + '' + '' + '') + sst = ('' + f'' + + "".join(f"{escape(t)}" for t in shared) + + '') + + with zipfile.ZipFile(path, "w", zipfile.ZIP_DEFLATED) as z: + z.writestr("[Content_Types].xml", ct) + z.writestr("_rels/.rels", rels) + z.writestr("xl/workbook.xml", wb) + z.writestr("xl/_rels/workbook.xml.rels", wb_rels) + z.writestr("xl/styles.xml", styles) + z.writestr("xl/sharedStrings.xml", sst) + for i, xml in enumerate(sheet_xml): + z.writestr(f"xl/worksheets/sheet{i+1}.xml", xml) + return {"sheets": names, "shared_strings": len(shared)} + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--node") + ap.add_argument("--cells-json") + ap.add_argument("--out", required=True) + a = ap.parse_args() + if a.cells_json: + sheets = json.load(open(a.cells_json))["sheets"] + elif a.node: + sheets = fetch_sheets(a.node) + else: + raise SystemExit("需要 --node 或 --cells-json") + info = build_xlsx(sheets, a.out) + rows = sum(len(s.get("cells") or []) for s in sheets) + print(json.dumps({"success": True, "out": a.out, "sheets": info["sheets"], + "sheet_count": len(sheets), "rows": rows, + "shared_strings": info["shared_strings"]}, ensure_ascii=False)) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/scripts/dws_error_map.py b/scripts/dws_error_map.py new file mode 100644 index 0000000..ed64cd1 --- /dev/null +++ b/scripts/dws_error_map.py @@ -0,0 +1,126 @@ +#!/usr/bin/env python3 +"""dws 错误 → 平台错误码/retryable 映射（阶段 B-③ 交付物）。 + +平台契约（internal/workflows/artifactfetch/handler.go）需要： + {ok, missing, retryable, code, error} +本模块把 dws 的两种错误体（严格命令的 ok/error 包裹、普通子命令的 error 对象） +以及本地错误、TLS/网络文本，统一映射成平台语义。 + +用法： + from dws_error_map import classify + classify(dws_json, stderr_text) -> {"ok":bool,"code":str,"retryable":bool,"missing":bool,"error":str} + +原则： + - 只有"明确瞬时"的信号才 retryable=True（429/5xx/超时/连接层失败） + - 未知错误默认 retryable=False + code=unknown —— 让它显式暴露，而不是被无限重试掩盖 + - missing=True 表示源侧对象已不存在/被删（平台据此走 missing 分支而不是 failed） +""" +import json +import re + +# (匹配正则, code, retryable, missing, 说明) +RULES = [ + # ---- 凭据 ---- + (r"请重新运行 dws auth login|resolve access token|invalidParameter\.authCode\.notFound|未登录|authenticated\"?\s*:\s*false|旧版登录态", + "credentials", False, False, "登录态失效/被顶掉，需重新设备流授权"), + # ---- TLS 信任库 ---- + (r"x509: certificate signed by unknown authority|tls: failed to verify certificate", + "tls_ca", False, False, "容器缺少 CA 信任库（装 ca-certificates 或挂载 bundle/SSL_CERT_FILE）"), + # ---- 权限 ---- + (r"forbidden\.accessDenied|Access Denied|accessdenied|HTTP 403|\"code\":\s*403", + "forbidden", False, False, "该账号对目标节点无权限"), + # ---- 限流 ---- + (r"HTTP 429|\b429\b|rate.?limit|Too Many Requests|请求过于频繁", + "http_429", True, False, "被限流，退避后重试"), + # ---- 本地文件冲突 / 大小限制 ---- + (r"LOCAL_FILE_EXISTS|目标文件已存在", "file_exists", False, False, "本地已有同名产物，换路径或用 drive download --overwrite"), + (r"LOCAL_DOWNLOAD_TOO_LARGE|超过上限 \d+ 字节|exceeds ingest size limit|超过入库大小限制", + "over_limit", False, False, "超过大小上限（严格版 512MB / 平台 200MB 配置）"), + # ---- 类型不支持 ---- + (r"不是钉钉在线文档|不是钉钉表格|notSupported\.docType|NODE_TYPE_MISMATCH|不是 a standalone \.adraw|无法使用文档相关工具|file type .* is not supported|在线表格不支持直接下载|不支持直接下载|请使用 getRange", + "unsupported_format", False, False, "节点类型与该入口不匹配（走对应产品命令）"), + # ---- 源对象不存在 ---- + (r"workspace node not found|not found in current inventory|节点不存在|dentry not resolvable|nodeId 指向的节点已", + "missing", False, True, "源侧节点已删除/移动"), + # ---- 空文档（与平台既有判定同义） ---- + (r"content is empty|正文为空|content\s+is\s+empty", "empty_document", False, False, "文档正文为空，无需入库"), + # ---- 服务端导出失败/超时 ---- + (r"导出任务失败|export job failed|submitExportJob", "export_failed", False, False, "钉钉导出任务失败"), + (r"导出任务超时|export job timed out|timed out|timeout|超时", "export_timeout", True, False, "导出/请求超时，可重试"), + # ---- 连接层 ---- + (r"connection refused|no such host|network is unreachable|i/o timeout|EOF|connection reset|request_failed|502|503|504|HTTP 5\d\d", + "source_unavailable", True, False, "源端不可达，可重试"), +] + +_JSON_ERR_KEYS = ("upstream_code", "server_error_code", "code", "category", "subtype", "message", "cause", "hint", "error") + + +def _flatten(payload): + """把 dws 错误体压成一段可正则匹配的文本。""" + if payload is None: + return "" + if isinstance(payload, str): + return payload + parts = [] + + def walk(node, depth=0): + if depth > 6: + return + if isinstance(node, dict): + for k, v in node.items(): + if k in _JSON_ERR_KEYS or isinstance(v, (dict, list)): + parts.append(str(k)) + walk(v, depth + 1) + elif isinstance(node, list): + for v in node[:20]: + walk(v, depth + 1) + else: + parts.append(str(node)) + + walk(payload) + return " | ".join(parts) + + +def classify(dws_json=None, stderr=None, exit_code=None): + """dws 结果 → 平台语义。dws_json 可为 dict / JSON 字符串 / None。""" + payload = dws_json + if isinstance(payload, str): + try: + payload = json.loads(payload) + except Exception: + payload = {"error": {"message": payload}} + text = _flatten(payload) + " || " + _flatten(stderr) + + ok = bool(payload) and isinstance(payload, dict) and ( + payload.get("ok") is True or payload.get("success") is True + or payload.get("outcome") == "success" or payload.get("status") == "SUCCESS") + if ok: + return {"ok": True, "code": "", "retryable": False, "missing": False, "error": ""} + + # 先判 missing 关键词（missing 优先于其它分类） + for pattern, code, retryable, missing, why in RULES: + if re.search(pattern, text, re.I): + msg = _short_message(payload) or why + return {"ok": False, "code": code, "retryable": retryable, "missing": missing, + "error": f"{why}: {msg}"[:400]} + + return {"ok": False, "code": "unknown", "retryable": False, "missing": False, + "error": (_short_message(payload) or str(stderr or "") or "dws 未分类错误")[:400]} + + +def _short_message(payload): + if isinstance(payload, dict): + err = payload.get("error") or payload + if isinstance(err, dict): + return str(err.get("message") or err.get("cause") or err.get("reason") or "") + return str(err)[:200] + return "" + + +if __name__ == "__main__": + import sys + raw = sys.stdin.read() + try: + print(json.dumps(classify(raw), ensure_ascii=False, indent=1)) + except Exception as e: + print(json.dumps({"ok": False, "code": "unknown", "error": str(e)}, ensure_ascii=False)) \ No newline at end of file diff --git a/scripts/test_dws_error_map.py b/scripts/test_dws_error_map.py new file mode 100644 index 0000000..b82a636 --- /dev/null +++ b/scripts/test_dws_error_map.py @@ -0,0 +1,106 @@ +#!/usr/bin/env python3 +"""dws_error_map 单测：全部使用本轮实测捕获的真实 dws 错误体（不改写、不简化）。""" +import json +import sys +sys.path.insert(0, "/home/zheromiku/dws_test/bridge") +from dws_error_map import classify + +# —— 真实样本（逐字来自本轮会话的工具输出）—— +CASES = [ + ("TLS 缺 CA（debian-slim 实测）", + {"category": "api", + "cause": 'Post "https://mcp-gw.dingtalk.com/server/91e17caf44f6ca1ed9c6ce614221a518ac93300ece63ca8d7e9b133f912e0607": tls: failed to verify certificate: x509: certificate signed by unknown authority', + "code": 1, "hint": "请检查网络连通性和 MCP 服务状态后重试。", "retryable": True, + "message": "提交导出任务失败: request to https://mcp-gw.dingtalk.com failed"}, + ("tls_ca", False, False)), + + ("登录态被顶掉（容器认证包轮换实测）", + {"type": "internal", "exit_code": 5, "upstream_code": "UNCLASSIFIED", + "message": "[UNCLASSIFIED] resolve access token: 旧版登录态已无法由当前认证服务刷新；本地 profile 已保留，请重新运行 dws auth login 完成一次重新授权: MCP token exchange failed: invalidParameter.authCode.notFound - 不合法的临时授权码"}, + ("credentials", False, False)), + + ("无权限（技术诊断部某 axls）", + {"error": {"actions": ["持续失败时保留 Trace ID 和 Server Code 联系服务端排查"], "category": "api", "code": 1, + "hint": "The API returned a business-level error.", "message": "Access Denied", + "operation": "tools/call", "reason": "business_error", + "server_error_code": "forbidden.accessDenied", "server_key": "sheet"}}, + ("forbidden", False, False)), + + ("严格版 512MB 上限（838MB xz 实测）", + {"type": "internal", "exit_code": 5, "message": "LOCAL_DOWNLOAD_TOO_LARGE: 响应大小 878540264 超过上限 536870912 字节"}, + ("over_limit", False, False)), + + ("本地文件已存在（探针重跑实测）", + {"type": "internal", "exit_code": 5, "message": "LOCAL_FILE_EXISTS: 目标文件已存在；请选择新的输出路径"}, + ("file_exists", False, False)), + + ("able 节点走 doc 入口（实测拒收）", + {"error": {"category": "api", "code": 1, + "message": "提交导出任务失败: 当前节点是钉钉AI表格（多维表，扩展名=able），不是钉钉在线文档（扩展名=adoc），无法使用文档相关工具操作。请改用钉钉AI表格相关 MCP Tool 或相关 CLI 命令进行操作。", + "server_error_code": "invalidRequest.inputArgs.invalid", "server_key": "doc"}}, + ("unsupported_format", False, False)), + + ("axls 不支持直下（实测拒收）", + {"type": "api", "subtype": "business_error", "exit_code": 1, + "upstream_code": "invalidRequest.inputArgs.invalid", + "message": "nodeId 指向的节点是钉钉表格（extension=axls），在线表格不支持直接下载。请使用 getRange 工具获取表格数据。"}, + ("unsupported_format", False, False)), + + ("节点不存在（dlink 节点实测）", + {"ok": False, "outcome": "failure", + "error": {"type": "api", "subtype": "business_error", "exit_code": 1, + "upstream_code": "internalError", "message": "workspace node not found"}}, + ("missing", False, True)), + + ("正文为空（平台历史缺陷文案）", + {"category": "api", "code": 1, "message": "DingTalk online document content is empty"}, + ("empty_document", False, False)), + + ("限流（构造，用于锁定 429 语义）", + {"category": "api", "code": 1, "message": "HTTP 429 Too Many Requests"}, + ("http_429", True, False)), + + ("连接层失败（构造，锁定可重试语义）", + {"category": "api", "code": 1, "cause": "dial tcp: lookup alidocs.dingtalk.com: no such host"}, + ("source_unavailable", True, False)), + + ("导出任务超时（构造，锁定可重试语义）", + {"category": "api", "code": 1, "message": "DingTalk export job timed out after 120s"}, + ("export_timeout", True, False)), + + ("未分类错误 → 不得默认可重试", + {"category": "internal", "code": 9, "message": "something nobody has seen before"}, + ("unknown", False, False)), +] + +SUCCESS_CASES = [ + ("严格版成功体", {"ok": True, "outcome": "success", "data": {"count": 1}}), + ("普通子命令成功体", {"success": True, "sizeBytes": 305078, "status": "SUCCESS"}), + ("drive 成功体", {"ok": True, "outcome": "success", "data": {"savedPath": "ref.xlsx", "sizeBytes": 3583939, "success": True}}), +] + +fails = 0 +print("=== 失败体映射 ===") +for label, payload, expect in CASES: + r = classify(payload) + got = (r["code"], r["retryable"], r["missing"]) + mark = "✅" if got == expect else "❌" + if got != expect: + fails += 1 + print(f" {mark} {label:34s} → code={r['code']:<18} retryable={str(r['retryable']):<5} missing={str(r['missing']):<5} 期望={expect}") + +print("\n=== 成功体映射 ===") +for label, payload in SUCCESS_CASES: + r = classify(payload) + ok = r["ok"] is True and r["code"] == "" + if not ok: + fails += 1 + print(f" {'✅' if ok else '❌'} {label:34s} → ok={r['ok']} code={r['code']!r}") + +print("\n=== 文本输入（stderr 兜底）===") +r = classify(None, stderr="Error: x509: certificate signed by unknown authority") +print(f" {'✅' if r['code']=='tls_ca' else '❌'} stderr-only 文本 → code={r['code']}") +fails += 0 if r["code"] == "tls_ca" else 1 + +print(f"\n结果: {'全部通过' if fails==0 else str(fails)+' 项失败'}（{len(CASES)+len(SUCCESS_CASES)+1} 个断言）") +sys.exit(1 if fails else 0) \ No newline at end of file