姚章浩
873 lines
42 KiB
Python
873 lines
42 KiB
Python
#!/usr/bin/env python3
|
||
"""Fetch one policy-selected catalog artifact from its live source.
|
||
|
||
The Go worker passes a small JSON manifest (argv[1]) describing the intent.
|
||
This process never receives database credentials or secret values; it locates
|
||
credentials only via the pre-mounted /run/source-secrets paths, exactly like
|
||
live_catalog_refresh.py. Behavior per source type:
|
||
|
||
alist -> direct read-only API: raw_url then download.
|
||
pp -> manifest.json local_path first (already on disk), else public URL.
|
||
dingtalk -> prefer the local export tree (converted docx/xlsx/pdf); if it is
|
||
older than the source-side modification time, export fresh bytes
|
||
through the web internal API captured headers.
|
||
|
||
Before any network call the fetcher compares against local candidates by
|
||
size and, when a hash is cheap to obtain, by content: a match means "本地已有
|
||
- no pull". Verified bytes are moved into the ingestion spool and a single
|
||
JSON result line is printed on stdout. Credentials, tokens, signed URLs and
|
||
authorization material are never printed.
|
||
"""
|
||
import csv
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import pathlib
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
import urllib.error
|
||
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",
|
||
"png": "png", "jpg": "jpg", "jpeg": "jpeg", "gif": "gif", "bmp": "bmp", "webp": "webp",
|
||
}
|
||
# Office types we refuse to ingest directly: PPT/PPTX must arrive as a PDF twin
|
||
# (embedded images travel inside those files already).
|
||
SLIDES = {"ppt", "pptx"}
|
||
LEGACY = {"doc": "doc", "xls": "xls"}
|
||
MAX_BYTES_DEFAULT = 200 * 1024 * 1024
|
||
|
||
|
||
class FetchError(Exception):
|
||
def __init__(self, message, retryable=True, code="fetch_failed"):
|
||
super().__init__(message)
|
||
self.retryable = retryable
|
||
self.code = code
|
||
|
||
|
||
class MissingSource(Exception):
|
||
pass
|
||
|
||
|
||
def emit(result):
|
||
print(json.dumps(result, ensure_ascii=False))
|
||
|
||
|
||
def sha256_file(path):
|
||
digest = hashlib.sha256()
|
||
with open(path, "rb") as handle:
|
||
for block in iter(lambda: handle.read(1 << 20), b""):
|
||
digest.update(block)
|
||
return digest.hexdigest()
|
||
|
||
|
||
def digest_of(value):
|
||
return hashlib.sha256(value.encode()).hexdigest()
|
||
|
||
|
||
def validate_magic(path, suffix):
|
||
with open(path, "rb") as handle:
|
||
head = handle.read(12)
|
||
checks = {
|
||
"pdf": head.startswith(b"%PDF-"),
|
||
"docx": head.startswith(b"PK\x03\x04"),
|
||
"xlsx": head.startswith(b"PK\x03\x04"),
|
||
"doc": head.startswith(b"\xd0\xcf\x11\xe0"),
|
||
"xls": head.startswith(b"\xd0\xcf\x11\xe0"),
|
||
"png": head.startswith(b"\x89PNG\r\n\x1a\n"),
|
||
"jpg": head[:2] == b"\xff\xd8" and head[2:3] == b"\xff",
|
||
"jpeg": head[:2] == b"\xff\xd8" and head[2:3] == b"\xff",
|
||
"gif": head[:6] in (b"GIF87a", b"GIF89a"),
|
||
"bmp": head[:2] == b"BM",
|
||
"webp": head[:4] == b"RIFF" and head[8:12] == b"WEBP",
|
||
}
|
||
if suffix in checks and not checks[suffix]:
|
||
raise FetchError("downloaded bytes are not a valid %s file" % suffix,
|
||
retryable=False, code="invalid_format")
|
||
if suffix in ("md", "txt", "csv", "json"):
|
||
with open(path, "rb") as handle:
|
||
head = handle.read(4096)
|
||
continues = bool(handle.read(1))
|
||
try:
|
||
head.decode("utf-8", errors="strict")
|
||
except UnicodeDecodeError as error:
|
||
# The sniff boundary can cut a multibyte character in half; that
|
||
# is not corruption while the file has bytes beyond the sniff.
|
||
boundary_cut = (continues and error.reason == "unexpected end of data"
|
||
and error.start >= len(head) - 3)
|
||
if not boundary_cut:
|
||
raise FetchError("text artifact is not valid UTF-8", retryable=False, code="invalid_format")
|
||
head[:error.start].decode("utf-8", errors="strict")
|
||
|
||
|
||
def pick_suffix(name, declared_type):
|
||
suffix = pathlib.PurePosixPath(name).suffix.lower().lstrip(".")
|
||
if declared_type in ("adoc",) and suffix in ("", "adoc"):
|
||
return "docx"
|
||
if declared_type in ("able", "axls") and suffix in ("", "able", "axls"):
|
||
return "xlsx"
|
||
if suffix in SUPPORTED:
|
||
return SUPPORTED[suffix]
|
||
if suffix == "adoc":
|
||
return "docx"
|
||
if suffix in ("able", "axls"):
|
||
return "xlsx"
|
||
return None
|
||
|
||
|
||
def http_get(url, headers, timeout, max_bytes, cookie_opener=None):
|
||
request = urllib.request.Request(url, headers=headers or {})
|
||
|
||
class NoRedirect(urllib.request.HTTPRedirectHandler):
|
||
location = ""
|
||
|
||
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
||
NoRedirect.location = newurl
|
||
return None
|
||
|
||
handler = NoRedirect()
|
||
opener = cookie_opener or urllib.request.build_opener(handler)
|
||
try:
|
||
response = opener.open(request, timeout=timeout)
|
||
except urllib.error.HTTPError as error:
|
||
if error.code in (301, 302, 303, 307, 308):
|
||
target = handler.location or error.headers.get("Location") or ""
|
||
host = urllib.parse.urlparse(target).netloc.lower()
|
||
if any(marker in host for marker in ("login.", "sso", "oauth", "passport")):
|
||
raise FetchError("文件所在的来源存储需要网页登录授权（存储后端未配置服务凭据），请联系管理员", retryable=False, code="source_auth_required")
|
||
raise FetchError("redirect refused", retryable=True, code="redirect")
|
||
if error.code == 404:
|
||
raise MissingSource("source returned 404")
|
||
raise FetchError("source returned HTTP %d" % error.code, retryable=error.code >= 500, code="http_%d" % error.code)
|
||
with response:
|
||
declared = response.headers.get("Content-Length")
|
||
if declared and int(declared) > max_bytes:
|
||
raise FetchError("source file exceeds ingest size limit", retryable=False, code="over_limit")
|
||
destination = tempfile.NamedTemporaryFile(delete=False, suffix=".fetch")
|
||
total = 0
|
||
digest = hashlib.sha256()
|
||
try:
|
||
while True:
|
||
block = response.read(1 << 20)
|
||
if not block:
|
||
break
|
||
total += len(block)
|
||
if total > max_bytes:
|
||
raise FetchError("source stream exceeds ingest size limit", retryable=False, code="over_limit")
|
||
digest.update(block)
|
||
destination.write(block)
|
||
finally:
|
||
destination.close()
|
||
if total == 0:
|
||
os.unlink(destination.name)
|
||
raise FetchError("source returned empty body", retryable=True, code="empty")
|
||
return destination.name, total, digest.hexdigest()
|
||
|
||
|
||
# ---------------------------------------------------------------- AList ------
|
||
|
||
def alist_download(manifest, max_bytes):
|
||
sys.path.insert(0, "/sources/alist")
|
||
from alist_readonly_inventory import AListReadOnlyClient, read_connections # noqa
|
||
|
||
token = pathlib.Path("/run/source-secrets/alist_api_token").read_text(encoding="utf-8").strip()
|
||
if not token:
|
||
raise FetchError("AList token empty", retryable=False, code="credentials")
|
||
rows = [row for row in read_connections(pathlib.Path("/run/source-secrets/alist_connections.xlsx"), require_token=False)
|
||
if row["category"] == manifest["source_name"]]
|
||
if len(rows) != 1:
|
||
raise FetchError("AList connection configuration missing", retryable=False, code="credentials")
|
||
base_url = rows[0]["base_url"].rstrip("/")
|
||
client = AListReadOnlyClient(base_url, token)
|
||
path = "/" + manifest["relative_path"].lstrip("/")
|
||
payload = json.dumps({"path": path, "password": ""}).encode("utf-8")
|
||
request = urllib.request.Request(base_url + "/api/fs/get", data=payload, method="POST", headers={
|
||
"X-Gateway-Token": token, "Content-Type": "application/json", "Accept": "application/json",
|
||
"User-Agent": "governance-artifact-fetch/1.0",
|
||
})
|
||
try:
|
||
with client.opener.open(request, timeout=30) as response:
|
||
body = json.loads(response.read().decode("utf-8"))
|
||
except Exception as error:
|
||
raise FetchError("AList file lookup failed (%s)" % type(error).__name__, code="source_unavailable")
|
||
if int(body.get("code") or 0) == 404:
|
||
raise MissingSource("AList file not present at expected path")
|
||
if int(body.get("code") or 0) != 200:
|
||
raise FetchError("AList file lookup refused (code %s)" % body.get("code"), code="source_unavailable")
|
||
data = body.get("data") or {}
|
||
raw_url = str(data.get("raw_url") or "")
|
||
if not raw_url:
|
||
raise MissingSource("AList returned no download locator for this file")
|
||
if manifest.get("size_bytes") in (None, 0) and data.get("size"):
|
||
manifest["size_bytes"] = int(data["size"])
|
||
download_url = urllib.parse.urljoin(base_url + "/", raw_url.lstrip("/"))
|
||
# 2026-09-28 实测：该网关的下载直链 /p/<path>?sign=... 由签名鉴权，请求头带
|
||
# 什么都不影响（无头也是 206）；真正需要凭证的是 /api/fs/* 路由。这里仍带上
|
||
# 网关头，Authorization 兜底保留用于旧网关形态。
|
||
try:
|
||
return http_get(download_url, {"X-Gateway-Token": token}, 180, max_bytes)
|
||
except FetchError as error:
|
||
if error.code == "source_auth_required":
|
||
try:
|
||
return http_get(download_url, {"Authorization": token}, 180, max_bytes)
|
||
except FetchError:
|
||
raise error
|
||
if error.code == "redirect":
|
||
# Same-origin /d/ route 302s to a signed external URL: follow once
|
||
# WITHOUT the source token (never forward credentials cross-origin).
|
||
class LocationCapture(urllib.request.HTTPRedirectHandler):
|
||
captured = ""
|
||
|
||
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
||
LocationCapture.captured = newurl
|
||
return None
|
||
|
||
probe = urllib.request.build_opener(LocationCapture)
|
||
try:
|
||
probe.open(urllib.request.Request(download_url, headers={"Authorization": token}), timeout=30)
|
||
except urllib.error.HTTPError:
|
||
pass
|
||
except Exception as failure:
|
||
raise FetchError("AList redirect probe failed (%s)" % type(failure).__name__, code="source_unavailable")
|
||
if not LocationCapture.captured:
|
||
raise
|
||
return http_get(LocationCapture.captured, {}, 180, max_bytes)
|
||
raise
|
||
|
||
|
||
# ------------------------------------------------------------------ PP -------
|
||
|
||
def pp_find_entry(manifest):
|
||
locator = manifest.get("locator_hash", "")
|
||
if not locator:
|
||
return None
|
||
manifest_path = pathlib.Path("/sources/pp/data/public-documents/manifest.json")
|
||
ledger = json.loads(manifest_path.read_text(encoding="utf-8"))
|
||
for result in ledger.get("results", []):
|
||
if digest_of(str(result.get("url") or "")) == locator:
|
||
return result
|
||
return None
|
||
|
||
|
||
def pp_download(manifest, max_bytes):
|
||
entry = pp_find_entry(manifest)
|
||
if entry is None:
|
||
raise MissingSource("PP ledger entry for this locator no longer exists")
|
||
# Ledger titles may omit the extension; recover the true file name from the
|
||
# local path or source URL before type checks run.
|
||
manifest_name = manifest.setdefault("name", "")
|
||
if not pathlib.PurePosixPath(manifest_name).suffix:
|
||
from urllib.parse import unquote, urlparse
|
||
for probe in (entry.get("local_path", ""), unquote(urlparse(str(entry.get("url", ""))).path)):
|
||
suffix = pathlib.PurePosixPath(probe).suffix.lower().lstrip(".")
|
||
if suffix:
|
||
manifest["name"] = manifest_name + "." + suffix
|
||
break
|
||
local_path = entry.get("local_path", "")
|
||
if local_path:
|
||
candidate = pathlib.Path(local_path.replace("/home/zheromiku/dobot-portal-sync", "/sources/pp", 1))
|
||
if not candidate.is_file():
|
||
candidate = pathlib.Path(local_path)
|
||
if candidate.is_file():
|
||
size = candidate.stat().st_size
|
||
expected_hash = str(entry.get("sha256", "")).lower()
|
||
actual = sha256_file(candidate)
|
||
if expected_hash and len(expected_hash) == 64 and actual != expected_hash:
|
||
raise FetchError("PP local copy hash mismatch (ledger changed?)", retryable=True, code="local_mismatch")
|
||
return str(candidate), size, actual, True
|
||
url = entry.get("url", "")
|
||
if not url:
|
||
raise MissingSource("PP ledger entry has no local file and no download URL")
|
||
path, size, actual = http_get(url, {}, 180, max_bytes)
|
||
expected_hash = str(entry.get("sha256", "")).lower()
|
||
if expected_hash and len(expected_hash) == 64 and actual != expected_hash:
|
||
os.unlink(path)
|
||
raise FetchError("PP download hash mismatch", retryable=True, code="hash_mismatch")
|
||
return path, size, actual, False
|
||
|
||
|
||
# ------------------------------------------------------------ DingTalk -------
|
||
|
||
def load_capture_headers():
|
||
capture = json.loads(pathlib.Path("/run/source-secrets/dingtalk_capture.json").read_text(encoding="utf-8"))
|
||
headers = capture.get("headers") or {}
|
||
def header(name):
|
||
for key, value in headers.items():
|
||
if key.lower() == name:
|
||
return value
|
||
return ""
|
||
a_token = header("a-token")
|
||
xsrf = header("x-xsrf-token")
|
||
cookie = header("cookie")
|
||
uid = ""
|
||
match = re.search(r'"?uid"?[=:]"?(\d{4,})', cookie or "")
|
||
if match:
|
||
uid = match.group(1)
|
||
if not a_token:
|
||
raise FetchError("DingTalk capture has no a-token; re-capture in 来源与凭据", retryable=False, code="credentials")
|
||
return {"a-token": a_token, "x-xsrf-token": xsrf, "cookie": cookie, "uid": uid,
|
||
"referer": "https://alidocs.dingtalk.com/"}
|
||
|
||
|
||
def dingtalk_local_export(manifest):
|
||
"""Locate an already-exported copy; returns (path, fresher_than_source) or None."""
|
||
name = pathlib.PurePosixPath(manifest["relative_path"]).name
|
||
stem = pathlib.PurePosixPath(name).stem
|
||
raw_suffix = pathlib.PurePosixPath(name).suffix.lower().lstrip(".")
|
||
suffix = pick_suffix(name, manifest.get("declared_type", ""))
|
||
root = pathlib.Path("/sources/dingtalk/exported")
|
||
folder = str(pathlib.PurePosixPath(manifest["relative_path"]).parent).strip(".")
|
||
search_suffixes = [suffix] if raw_suffix not in SLIDES else ["pdf", "pptx", "ppt"]
|
||
candidates = []
|
||
for probe in search_suffixes:
|
||
if probe is None:
|
||
continue
|
||
exact = root / manifest["source_name"] / folder / (stem + "." + probe)
|
||
if exact.is_file():
|
||
candidates.append(exact)
|
||
else:
|
||
for base in (root / manifest["source_name"], pathlib.Path("/sources/dingtalk/converted_ppt_pdf"),
|
||
pathlib.Path("/sources/dingtalk/converted_ppt_pdf_full")):
|
||
if not base.is_dir():
|
||
continue
|
||
for hit in base.rglob(stem + "." + probe):
|
||
candidates.append(hit)
|
||
if candidates:
|
||
break
|
||
if not candidates:
|
||
if raw_suffix not in SLIDES and suffix is None:
|
||
raise MissingSource("dingtalk file type is not ingestable")
|
||
return None
|
||
modified = str(manifest.get("modified_at") or "")
|
||
candidate = candidates[0]
|
||
local_modified = time.strftime("%Y-%m-%dT%H:%M:%S+08:00", time.localtime(candidate.stat().st_mtime))
|
||
fresh = (not modified) or (local_modified >= modified[:19] + local_modified[19:])
|
||
return str(candidate), fresh
|
||
|
||
|
||
def dingtalk_live_raw_key(manifest):
|
||
"""Resolve in the configured source; names locate folders, never file identity."""
|
||
from live_catalog_refresh import dingtalk_root_key
|
||
from source_catalog_import import node_id as catalog_node_id, safe_relative_path
|
||
from dws_dingtalk_scan import normalise
|
||
|
||
match = re.fullmatch(r"spaces/([A-Za-z0-9_-]+)\?space=([A-Za-z0-9_-]+)", manifest['source_root'])
|
||
if not match:
|
||
raise FetchError('DingTalk source root is invalid', retryable=False, code='source_identity')
|
||
root_id, space_id = match.groups()
|
||
try:
|
||
source_key = 'dingtalk-' + dingtalk_root_key(manifest['connection_id'], space_id)
|
||
parts = safe_relative_path(manifest['relative_path']).split('/')
|
||
except (OSError, ValueError, RuntimeError):
|
||
raise FetchError('DingTalk source identity cannot be resolved', retryable=False, code='source_identity')
|
||
folder = ''
|
||
for index in range(len(parts)):
|
||
drive = space_id.isdigit()
|
||
args = ['drive', '+list', '--space-id', space_id] if drive else ['wiki', '+node-list', '--workspace', space_id]
|
||
args += ['--page-all', '--limit', '50', '--max-items', '10000', '--page-limit', '200']
|
||
if folder:
|
||
args += ['--folder', folder]
|
||
payload, verdict = dws_run(args, timeout=180)
|
||
if not verdict['ok']:
|
||
raise _dws_error(verdict)
|
||
data = payload.get('data', {}) if isinstance(payload, dict) else {}
|
||
entries = data.get('files' if drive else 'nodes')
|
||
if not isinstance(entries, list) or data.get('hasMore') is not False or data.get('autoPageComplete') is not True:
|
||
raise FetchError('DingTalk directory response is incomplete', retryable=True, code='source_lookup_incomplete')
|
||
matches = []
|
||
for entry in entries:
|
||
raw, node, is_folder = normalise(entry, folder or root_id)
|
||
if index == len(parts) - 1:
|
||
if not is_folder and catalog_node_id(source_key, raw) == manifest['node_id']:
|
||
matches.append(raw)
|
||
elif is_folder and node['name'] == parts[index]:
|
||
matches.append(raw)
|
||
if len(matches) != 1:
|
||
# A missing/ambiguous path is not evidence that the source file was deleted.
|
||
raise FetchError('DingTalk node identity unresolved; refresh 来源目录 before retrying',
|
||
retryable=False, code='source_node_unresolved')
|
||
folder = matches[0]
|
||
return folder, 'file'
|
||
|
||
|
||
def dingtalk_raw_keys(manifest):
|
||
"""Use current source coordinates; retain inventory support for old manifests."""
|
||
if manifest.get('source_root'):
|
||
return dingtalk_live_raw_key(manifest)
|
||
state_path = pathlib.Path("/sources/dingtalk/data/dingtalk-internal/inventory-state.json")
|
||
state = json.loads(state_path.read_text(encoding="utf-8"))
|
||
roots = state.get("roots") or {}
|
||
space_key = _space_key(roots, manifest["source_name"])
|
||
match = [value for key, value in sorted(roots.items()) if key == space_key]
|
||
if len(match) != 1:
|
||
raise FetchError("DingTalk space not present in the scanner inventory; refresh 来源目录 first",
|
||
retryable=False, code="space_missing")
|
||
sys.path.insert(0, "/app/scripts")
|
||
from source_catalog_import import node_id as catalog_node_id # noqa
|
||
source_id = "dingtalk-" + space_key
|
||
nodes = match[0].get("nodes") or {}
|
||
for raw, item in nodes.items():
|
||
if catalog_node_id(source_id, raw) == manifest["node_id"]:
|
||
return raw, str(item.get("dentry_type") or item.get("type") or "")
|
||
raise MissingSource("DingTalk node not found in current inventory (moved or deleted?)")
|
||
|
||
|
||
def _space_key(roots, name):
|
||
for key, value in roots.items():
|
||
if value.get("live_space_name") == name:
|
||
return key
|
||
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")
|
||
|
||
|
||
_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)
|
||
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)."""
|
||
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(".")
|
||
# Catalog size is only a true byte count for ORDINARY files. For online documents
|
||
# (adoc/axls/able/...) the catalog size is the container size, which can exceed the
|
||
# cap while the exported docx/xlsx stays small — pre-checking those would wrongly
|
||
# reject documents that the legacy path exported happily.
|
||
online_document = extension in ("adoc", "axls", "able", "appt", "amind", "adraw")
|
||
catalog_size = manifest.get("size_bytes")
|
||
if not online_document and isinstance(catalog_size, int) and catalog_size > max_bytes:
|
||
raise FetchError("artifact exceeds ingest size limit", retryable=False, code="over_limit")
|
||
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"
|
||
try:
|
||
axls_to_xlsx.build_xlsx(axls_to_xlsx.fetch_sheets(raw_key), str(workdir / target))
|
||
except axls_to_xlsx.DwsCallError as 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],
|
||
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"
|
||
|
||
def guarded(local_path):
|
||
if pathlib.PurePosixPath(str(local_path)).suffix.lower() in (".docx", ".xlsx") \
|
||
and exported_doc_is_blank(local_path, spreadsheet):
|
||
raise FetchError("钉钉文档正文为空，无需入库（如确有内容请在钉钉里补充后重新取件）",
|
||
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, "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, "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[0], live_path[1], live_path[2], live_path[3], "legacy"
|
||
|
||
|
||
def dingtalk_json_api(url, headers, payload=None, extra_headers=None, method=None):
|
||
request_headers = {
|
||
"a-token": headers["a-token"], "x-xsrf-token": headers.get("x-xsrf-token", ""),
|
||
"cookie": headers.get("cookie", ""), "content-type": "application/json",
|
||
"accept": "application/json", "accept-language": "zh-CN",
|
||
"origin": "https://alidocs.dingtalk.com", "referer": headers["referer"],
|
||
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
|
||
}
|
||
request_headers.update(extra_headers or {})
|
||
body = json.dumps(payload, ensure_ascii=False).encode("utf-8") if payload is not None else None
|
||
request = urllib.request.Request(url, headers=request_headers, data=body, method=method)
|
||
try:
|
||
with urllib.request.urlopen(request, timeout=60) as response:
|
||
return json.loads(response.read().decode("utf-8"))
|
||
except urllib.error.HTTPError as error:
|
||
raise FetchError("DingTalk web API returned HTTP %d" % error.code,
|
||
retryable=error.code == 429 or error.code >= 500, code="http_%d" % error.code)
|
||
except (urllib.error.URLError, TimeoutError) as error:
|
||
raise FetchError("DingTalk web API unavailable (%s)" % type(error).__name__, code="source_unavailable")
|
||
|
||
|
||
def dingtalk_put_bytes(url, payload):
|
||
request = urllib.request.Request(url, data=payload, method="PUT", headers={
|
||
"content-length": str(len(payload)), "content-type": "",
|
||
"referer": "https://alidocs.dingtalk.com/", "user-agent": "Mozilla/5.0",
|
||
})
|
||
try:
|
||
with urllib.request.urlopen(request, timeout=90) as response:
|
||
if response.status not in (200, 201, 204):
|
||
raise FetchError("DingTalk export upload returned HTTP %d" % response.status,
|
||
retryable=response.status >= 500, code="export_upload")
|
||
except urllib.error.HTTPError as error:
|
||
raise FetchError("DingTalk export upload returned HTTP %d" % error.code,
|
||
retryable=error.code == 429 or error.code >= 500, code="http_%d" % error.code)
|
||
|
||
|
||
def exported_doc_is_blank(path, spreadsheet):
|
||
"""DingTalk happily exports an empty online document to a valid but
|
||
contentless docx/xlsx. The parser rejects those with a generic
|
||
remote_parse_failed, so flag them here with an actionable message."""
|
||
try:
|
||
import zipfile
|
||
with zipfile.ZipFile(path) as archive:
|
||
names = archive.namelist()
|
||
if any(name.startswith("word/media/") or name.startswith("xl/media/") for name in names):
|
||
return False # image-only doc: the parser still has something to read
|
||
parts = ["xl/sharedStrings.xml"] if spreadsheet else ["word/document.xml"]
|
||
for part in parts:
|
||
try:
|
||
blob = archive.read(part)
|
||
except KeyError:
|
||
blob = b""
|
||
text = re.sub(rb"<[^>]+>", b"", blob).strip()
|
||
if text:
|
||
return False
|
||
return True
|
||
except Exception:
|
||
return False # unreadable zip: the magic/type checks handle that error
|
||
|
||
|
||
def dingtalk_export_adoc(info, headers, manifest, max_bytes, spreadsheet=False):
|
||
doc_key = str(info.get("docKey") or "")
|
||
dentry_key = str(info.get("dentryKey") or "")
|
||
if not doc_key:
|
||
raise MissingSource("DingTalk online document has no docKey")
|
||
source_app = "spreadsheet" if spreadsheet else "doc"
|
||
doc_type = "sheet" if spreadsheet else "doc"
|
||
referer = ("https://alidocs.dingtalk.com/note/edit?dt_editor_toolbar=true&biz_ver=10"
|
||
"&docType=%s&docId=%s&dontjump=true" % (doc_type, urllib.parse.quote(doc_key)))
|
||
extra = {"referer": referer, "source_doc_app": source_app,
|
||
"a-doc-key": doc_key, "a-dentry-key": dentry_key}
|
||
document = dingtalk_json_api("https://alidocs.dingtalk.com/api/document/data", headers,
|
||
{"docKey": doc_key}, extra_headers=extra, method="POST")
|
||
checkpoint = (((document.get("data") or {}).get("documentContent") or {}).get("checkpoint") or {})
|
||
content = checkpoint.get("content") or ""
|
||
if not isinstance(content, str) or not content:
|
||
raise FetchError("DingTalk online document content is empty", retryable=False, code="empty_document")
|
||
content_bytes = content.encode("utf-8")
|
||
if len(content_bytes) > max_bytes:
|
||
raise FetchError("source file exceeds ingest size limit", retryable=False, code="over_limit")
|
||
upload = dingtalk_json_api("https://alidocs.dingtalk.com/core/api/resources/9/upload_info", headers,
|
||
{"size": len(content_bytes), "resourceName": manifest["name"], "contentType": ""},
|
||
extra_headers={**extra, "a-host-doc-key": ""}, method="POST")
|
||
upload_info = upload.get("data") or {}
|
||
storage_path, upload_url = upload_info.get("storagePath"), upload_info.get("uploadUrl")
|
||
if not storage_path or not upload_url:
|
||
raise FetchError("DingTalk export upload slot is missing", retryable=True, code="export_slot")
|
||
dingtalk_put_bytes(upload_url, content_bytes)
|
||
submit_body = {"storagePath": storage_path,
|
||
"exportType": "dingTalksheetToxlsx" if spreadsheet else "dingTalkdocTodocx"}
|
||
if spreadsheet:
|
||
submit_body["extra"] = {"imgCnt": 0}
|
||
submitted = dingtalk_json_api("https://alidocs.dingtalk.com/core/api/document/submitExportJob", headers,
|
||
submit_body, extra_headers=extra, method="POST")
|
||
job_id = str((submitted.get("data") or {}).get("jobId") or "")
|
||
if not job_id:
|
||
raise FetchError("DingTalk export job was not created", retryable=True, code="export_submit")
|
||
deadline = time.time() + 120
|
||
while time.time() < deadline:
|
||
status = dingtalk_json_api("https://alidocs.dingtalk.com/core/api/document/queryExportJobInfo?jobId=" + urllib.parse.quote(job_id),
|
||
headers, extra_headers=extra)
|
||
detail = status.get("data") or {}
|
||
if detail.get("status") == "success" and detail.get("ossUrl"):
|
||
result = http_get(detail["ossUrl"], {}, 180, max_bytes)
|
||
if exported_doc_is_blank(result[0], spreadsheet):
|
||
raise FetchError("钉钉文档正文为空，无需入库（如确有内容请在钉钉里补充后重新取件）",
|
||
retryable=False, code="empty_document")
|
||
return result + (False,)
|
||
if detail.get("status") in ("fail", "failed", "error"):
|
||
raise FetchError("DingTalk export job failed", retryable=False, code="export_failed")
|
||
time.sleep(3)
|
||
raise FetchError("DingTalk export job timed out", retryable=True, code="export_timeout")
|
||
|
||
|
||
def live_export_dingtalk_document(raw_key, headers, manifest, max_bytes):
|
||
info = dingtalk_json_api("https://alidocs.dingtalk.com/box/api/v2/dentry/info?dentryUuid=" + urllib.parse.quote(raw_key), headers)
|
||
data = info.get("data") or {}
|
||
extension = str(data.get("extension") or pathlib.PurePosixPath(manifest["name"]).suffix.lstrip(".")).lower()
|
||
if extension in ("adoc", "axls"):
|
||
return dingtalk_export_adoc(data, headers, manifest, max_bytes, spreadsheet=extension == "axls")
|
||
space_id, dentry_id = str(data.get("spaceId") or ""), str(data.get("dentryId") or "")
|
||
if not space_id or not dentry_id:
|
||
raise MissingSource("DingTalk dentry not resolvable")
|
||
download = dingtalk_json_api("https://alidocs.dingtalk.com/box/api/v1/file/download?spaceId=%s&dentryId=%s" % (space_id, dentry_id), headers)
|
||
payload = download.get("data") or {}
|
||
if payload.get("downloadType") == "URL_PRE_SIGNATURE":
|
||
urls = (payload.get("ossUrlPreSignatureInfo") or {}).get("preSignUrls") or []
|
||
if urls:
|
||
return http_get(urls[0], {}, 180, max_bytes) + (False,)
|
||
if payload.get("downloadType") == "HEADER_SIGNATURE":
|
||
signature = payload.get("headerSignatureInfo") or {}
|
||
urls = signature.get("resourceUrl") or []
|
||
if urls:
|
||
return http_get(urls[0], signature.get("headers") or {}, 180, max_bytes) + (False,)
|
||
return None
|
||
|
||
|
||
# ---------------------------------------------------------------- main -------
|
||
|
||
def suffix_from_magic(path):
|
||
with open(path, "rb") as handle:
|
||
head = handle.read(12)
|
||
if head.startswith(b"%PDF-"):
|
||
return "pdf"
|
||
if head.startswith(b"PK\x03\x04"):
|
||
# zip container: peek one entry name to pick office family
|
||
try:
|
||
import zipfile
|
||
with zipfile.ZipFile(path) as archive:
|
||
names = set(archive.namelist()[:50])
|
||
if "word/document.xml" in names:
|
||
return "docx"
|
||
if "xl/workbook.xml" in names:
|
||
return "xlsx"
|
||
except Exception:
|
||
return None
|
||
return None
|
||
if head.startswith(b"\xd0\xcf\x11\xe0"):
|
||
return None
|
||
for suffix in ("png", "jpg", "gif", "bmp", "webp"):
|
||
checks = validate_magic_map(head)
|
||
if checks.get(suffix):
|
||
return suffix
|
||
try:
|
||
with open(path, "rb") as handle:
|
||
handle.read(4096).decode("utf-8", errors="strict")
|
||
return "md"
|
||
except (UnicodeDecodeError, OSError):
|
||
return None
|
||
|
||
|
||
def validate_magic_map(head):
|
||
return {
|
||
"png": head.startswith(b"\x89PNG\r\n\x1a\n"),
|
||
"jpg": head[:2] == b"\xff\xd8" and head[2:3] == b"\xff",
|
||
"gif": head[:6] in (b"GIF87a", b"GIF89a"),
|
||
"bmp": head[:2] == b"BM",
|
||
"webp": head[:4] == b"RIFF" and head[8:12] == b"WEBP",
|
||
}
|
||
|
||
|
||
def main():
|
||
manifest = json.loads(sys.argv[1])
|
||
max_bytes = int(os.environ.get("ARTIFACT_FETCH_MAX_BYTES") or MAX_BYTES_DEFAULT)
|
||
spool_root = pathlib.Path(os.environ.get("INGESTION_SPOOL_ROOT") or "/var/lib/governance/ingestion")
|
||
name = manifest["name"]
|
||
declared_type = manifest.get("declared_type", "")
|
||
matched_local = False
|
||
if manifest["source_type"] == "alist":
|
||
path, size, digest = alist_download(manifest, max_bytes)
|
||
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, 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
|
||
# titles may omit extensions and DingTalk exports always change them.
|
||
suffix = pick_suffix(pathlib.PurePosixPath(path).name, declared_type) or pick_suffix(name, declared_type)
|
||
if suffix is None:
|
||
suffix = suffix_from_magic(path)
|
||
if suffix is None:
|
||
raw_suffix = pathlib.PurePosixPath(name).suffix.lower().lstrip(".")
|
||
hint = " (convert PPT to PDF first)" if raw_suffix in SLIDES else ""
|
||
raise FetchError("file type .%s is not supported by the parser%s" % (raw_suffix or "?", hint),
|
||
retryable=False, code="unsupported_format")
|
||
try:
|
||
validate_magic(path, suffix)
|
||
if size > max_bytes:
|
||
raise FetchError("artifact exceeds ingest size limit", retryable=False, code="over_limit")
|
||
relative_ref = "artifact/%s/%s/%s.%s" % (digest[:2], digest[2:4], manifest["node_id"], suffix)
|
||
destination = spool_root / relative_ref
|
||
destination.parent.mkdir(parents=True, exist_ok=True)
|
||
staged = str(destination) + ".part"
|
||
shutil.copyfile(path, staged)
|
||
if sha256_file(staged) != digest:
|
||
raise FetchError("spooled bytes failed verification", code="spool_mismatch")
|
||
os.replace(staged, destination)
|
||
result = {"ok": True, "spec_id": manifest["spec_id"], "node_id": manifest["node_id"],
|
||
"connection_id": manifest["connection_id"], "sha256": digest, "size_bytes": size,
|
||
"spool_ref": os.environ.get("SPOOL_REF_PREFIX", "") + relative_ref, "validated_format": suffix, "matched_local": matched_local}
|
||
finally:
|
||
if path.startswith(tempfile.gettempdir()) and os.path.exists(path):
|
||
os.unlink(path)
|
||
emit(result)
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
sys.exit(main())
|
||
except MissingSource as error:
|
||
emit({"ok": False, "missing": True, "error": str(error)})
|
||
sys.exit(3)
|
||
except FetchError as error:
|
||
emit({"ok": False, "retryable": error.retryable, "code": error.code, "error": str(error)})
|
||
sys.exit(4 if error.retryable else 2)
|