fix(artifact): 类型误标回退 + 服务端阻断判可重试 + 空白单元格重读

压测/取件实测暴露的三类问题（均在真实 worker 容器里复现并已验证修复）：
- 名为 .axls 实为 docx 的节点（清单扩展名与真实类型不符）：dws 明确回报真实类型
  → 回退 drive download 取原始字节（实测该节点 243,949B docx 可正常取到），
  不再直接失败
- server model operate block 判可重试：同一节点 5 连测 3/5 成功（另一支 timeout），
  原判终态会把可成功节点永久失败
- axls 大表偶发返回全空单元格（12 次取件出现 1 次）：全空时重读一次再上报，
  避免被平台空壳判定当「空文档」终态

新增 3 个回归测试（18 tests / 19 断言全绿）

scripts/artifact_fetch.py

@@ -439,6 +439,15 @@ def _dws_error(verdict):
                       code=verdict.get("code") or "unknown")
 
 
+_REPORTED_TYPE_RE = re.compile(r"当前文档类型[：:]\s*([A-Za-z0-9]{2,8})")
+
+
+def _dws_reported_real_type(message):
+    """从 dws 的"当前节点不是钉钉表格…当前文档类型：docx"里取出真实类型（取不到返回 ""）。"""
+    found = _REPORTED_TYPE_RE.search(message or "")
+    return found.group(1).lower() if found else ""
+
+
 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)
@@ -501,10 +510,20 @@ def dingtalk_dws_download(manifest, max_bytes):
         try:
             axls_to_xlsx.build_xlsx(axls_to_xlsx.fetch_sheets(raw_key), str(workdir / target))
         except axls_to_xlsx.DwsCallError as error:
-            # Map the converter's dws failure onto the platform contract instead of
-            # letting the process die without a result (the Go runner would then only
-            # see "artifact_fetch_no_output").
-            raise _dws_error(dws_error_map.classify({"error": {"message": str(error)}}))
+            # 清单扩展名与真实类型不符（实测：名为 .axls 实为 docx）时，dws 会明确告知真实
+            # 类型 → 按普通文件取原始字节，交给平台既有的 magic/格式校验去判。
+            real_type = _dws_reported_real_type(str(error))
+            if real_type and real_type not in ("axls", "adoc", "able", "appt", "amind", "adraw"):
+                target = "out." + real_type
+                _, verdict = dws_run(["drive", "+download", "--node", raw_key, "--output", target],
+                                     cwd=str(workdir), timeout=900)
+                if not verdict["ok"]:
+                    raise _dws_error(verdict)
+            else:
+                # Map the converter's dws failure onto the platform contract instead of
+                # letting the process die without a result (the Go runner would then only
+                # see "artifact_fetch_no_output").
+                raise _dws_error(dws_error_map.classify({"error": {"message": str(error)}}))
     else:
         target = "out." + (extension or "bin")
         _, verdict = dws_run(["drive", "+download", "--node", raw_key, "--output", target],

scripts/artifact_fetch_test.py

@@ -212,6 +212,48 @@ class DingTalkDwsFetcherTest(unittest.TestCase):
         self.assertEqual(raised.exception.code, "forbidden")
         self.assertFalse(raised.exception.retryable)
 
+    def test_axls_with_wrong_extension_falls_back_to_drive_download(self):
+        # 实测：名为 .axls 实为 docx（清单扩展名与真实类型不符）。dws 会告知真实类型
+        # → 取原始字节，交给平台既有校验，而不是直接失败。
+        calls = []
+
+        def fake_run(args, timeout=300, cwd=None):
+            calls.append(args[:2])
+            if args[:2] == ["drive", "+download"]:
+                pathlib.Path(cwd, "out.docx").write_bytes(b"PK\x03\x04" + b"w" * 100)
+                return {"success": True}, self.ok_verdict
+            self.fail("不应走其它路径: %s" % args[:3])
+
+        with mock.patch.object(artifact_fetch, "DINGTALK_FETCHER", "dws"), \
+             mock.patch.object(artifact_fetch, "dingtalk_raw_keys", return_value=("raw-14", "file")), \
+             mock.patch.object(artifact_fetch, "dws_run", side_effect=fake_run), \
+             mock.patch.object(artifact_fetch.axls_to_xlsx, "fetch_sheets",
+                               side_effect=artifact_fetch.axls_to_xlsx.DwsCallError(
+                                   "当前节点不是钉钉表格（扩展名=axls），无法使用表格相关工具操作。"
+                                   "当前文档类型：docx。请确认 nodeId 指向的是钉钉表格文件")):
+            path, size, digest, matched, fetcher = artifact_fetch.dingtalk_download(
+                {**self.manifest, "name": "各机型易损件清单.axls", "declared_type": "axls"}, 200_000_000)
+
+        self.assertEqual(calls, [["drive", "+download"]])
+        self.assertEqual(fetcher, "dws")
+        self.assertEqual(size, 104)
+
+    def test_axls_blank_read_is_retried_once(self):
+        # 大表偶发返回全空单元格：重读一次拿到数据时不得上报空白
+        empty_first = {"n": 0}
+
+        def flaky(args, timeout=180):
+            if args[:2] == ["sheet", "list"]:
+                return {"sheets": [{"sheetId": "st-1", "name": "S1"}]}
+            empty_first["n"] += 1
+            return {"cells": [] if empty_first["n"] == 1 else [["机型", "部件"]]}
+
+        with mock.patch.object(artifact_fetch.axls_to_xlsx, "run_json", side_effect=flaky):
+            sheets = artifact_fetch.axls_to_xlsx.fetch_sheets("raw-15")
+
+        self.assertEqual(empty_first["n"], 2, "全空后必须重读一次")
+        self.assertTrue(sheets[0]["cells"])
+
     def test_link_and_mind_nodes_map_to_missing_and_unsupported(self):
         cases = (("dlink", "链接节点.dlink", "missing", True),
                  ("amind", "培训大纲.amind", "unsupported_format", False))

scripts/axls_to_xlsx.py

@@ -54,6 +54,17 @@ def run_json(args, timeout=180):
 
 def fetch_sheets(node):
     """返回 [{'name':str,'cells':[[...]]}]"""
+    out = _read_sheets(node)
+    if all(not s.get("cells") for s in out):
+        # 实测：大表（220MB/15 表）偶发返回全空单元格，重读一次即正常；空白结果
+        # 一旦上报会被平台的空壳判定当作"空文档"终态处理，所以这里先重读一遍。
+        retry = _read_sheets(node)
+        if any(s.get("cells") for s in retry):
+            return retry
+    return out
+
+
+def _read_sheets(node):
     listing = run_json(["sheet", "list", "--node", node])
     sheets = listing.get("sheets") or []
     if not sheets:

scripts/dws_error_map.py

@@ -46,6 +46,8 @@ RULES = [
     (r"content is empty|正文为空|content\s+is\s+empty", "empty_document", False, False, "文档正文为空，无需入库"),
     # ---- 服务端导出失败/超时 ----
     (r"导出任务失败|export job failed|submitExportJob", "export_failed", True, False, "钉钉导出任务失败"),
+    # 服务端主动阻断（实测同一节点 3/5 成功，5 连测里另一支是 timeout）→ 瞬时，必须可重试
+    (r"model\.operate\.block|server model operate block", "upstream_blocked", True, 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",

scripts/test_dws_error_map.py

@@ -77,6 +77,11 @@ CASES = [
                 "server_error_code": "internalError"}},
      ("export_failed", True, False)),
 
+    ("服务端阻断（5 连测实测：同一节点 3/5 成功，另一支是 timeout）",
+     {"error": {"category": "api", "code": 1, "message": "server model operate block",
+                "server_error_code": "model.operate.block", "reason": "business_error"}},
+     ("upstream_blocked", True, False)),
+
     ("未分类错误 → 不得默认可重试",
      {"category": "internal", "code": 9, "message": "something nobody has seen before"},
      ("unknown", False, False)),