feat(artifact): 钉钉取件优先走 dws（DingTalk Workspace CLI），老路径保留为回退

- artifact_fetch.py: 新增 dingtalk dws 取件器（adoc→doc export、axls→sheet 转换、其余→drive 直下），
  ARTIFACT_DINGTALK_FETCHER=dws|auto|legacy 三级开关，auto 仅对可重试错误回落老路径；
  平台侧判定（pick_suffix/SLIDES/magic/大小上限/sha256/spool/空壳检查）全部保留；
  结果 JSON 字段不变（Go runner 用 DisallowUnknownFields 解析 stdout），取件器来源只写 stderr。
- axls_to_xlsx.py: 纯标准库 axls→xlsx（文本写入 sharedStrings 以通过空壳判定）。
- dws_error_map.py: dws 错误→平台 {ok,code,retryable,missing}；未分类错误默认不重试。
- Dockerfile: alpine 增加 gcompat/libc6-compat；dws v1.0.63 固定版本+sha256 校验；构建期 dws version 自检。
- compose.yaml: worker 增加 dws 环境与凭据挂载（DWS_HOME 方案，XDG_DATA_HOME 不被 dws 识别）。
- 测试: artifact_fetch_test 新增 11 项（路由/回落语义/契约字段锁定）；dws_error_map 17 断言。
- 实测: 真容器(--read-only --tmpfs --user 1000:1000) 端到端 adoc/axls/pdf 3/3 成功，docx sha 与平台基准一致。

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"]

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

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>:<gid>`（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 <worker> 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 |

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

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()

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'<c r="{ref}"><v>{v}</v></c>')
+                else:
+                    cells.append(f'<c r="{ref}" t="s"><v>{sidx(v)}</v></c>')
+            if cells:
+                chunks.append(f'<row r="{ri}">{"".join(cells)}</row>')
+        sheet_xml.append(
+            '<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
+            '<worksheet xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main">'
+            f'<sheetData>{"".join(chunks)}</sheetData></worksheet>')
+
+    n = len(names)
+    ct = ('<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
+          '<Types xmlns="http://schemas.openxmlformats.org/package/2006/content-types">'
+          '<Default Extension="rels" ContentType="application/vnd.openxmlformats-package.relationships+xml"/>'
+          '<Default Extension="xml" ContentType="application/xml"/>'
+          '<Override PartName="/xl/workbook.xml" ContentType="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet.main+xml"/>'
+          '<Override PartName="/xl/styles.xml" ContentType="application/vnd.openxmlformats-officedocument.spreadsheetml.styles+xml"/>'
+          '<Override PartName="/xl/sharedStrings.xml" ContentType="application/vnd.openxmlformats-officedocument.spreadsheetml.sharedStrings+xml"/>'
+          + "".join(f'<Override PartName="/xl/worksheets/sheet{i+1}.xml" ContentType="application/vnd.openxmlformats-officedocument.spreadsheetml.worksheet+xml"/>' for i in range(n))
+          + '</Types>')
+    rels = ('<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
+            '<Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships">'
+            '<Relationship Id="rId1" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/officeDocument" Target="xl/workbook.xml"/>'
+            '</Relationships>')
+    wb = ('<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
+          '<workbook xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main" '
+          'xmlns:r="http://schemas.openxmlformats.org/officeDocument/2006/relationships"><sheets>'
+          + "".join(f'<sheet name="{escape(names[i])}" sheetId="{i+1}" r:id="rId{i+1}"/>' for i in range(n))
+          + '</sheets></workbook>')
+    wb_rels = ('<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
+               '<Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships">'
+               + "".join(f'<Relationship Id="rId{i+1}" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/worksheet" Target="worksheets/sheet{i+1}.xml"/>' for i in range(n))
+               + f'<Relationship Id="rId{n+1}" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/styles" Target="styles.xml"/>'
+               + f'<Relationship Id="rId{n+2}" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/sharedStrings" Target="sharedStrings.xml"/>'
+               + '</Relationships>')
+    styles = ('<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
+              '<styleSheet xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main">'
+              '<fonts count="1"><font><sz val="11"/><name val="Calibri"/></font></fonts>'
+              '<fills count="1"><fill><patternFill patternType="none"/></fill></fills>'
+              '<borders count="1"><border/></borders>'
+              '<cellStyleXfs count="1"><xf/></cellStyleXfs>'
+              '<cellXfs count="1"><xf/></cellXfs></styleSheet>')
+    sst = ('<?xml version="1.0" encoding="UTF-8" standalone="yes"?>'
+           f'<sst xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main" count="{len(shared)}" uniqueCount="{len(shared)}">'
+           + "".join(f"<si><t xml:space=\"preserve\">{escape(t)}</t></si>" for t in shared)
+           + '</sst>')
+
+    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()

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))

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)