#!/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


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_raw_keys(manifest):
    """Recover the dentry uuid behind a node id via the scanner inventory."""
    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 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

    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
        # 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
    if not live_path:
        raise MissingSource("DingTalk file is not exported and cannot be exported automatically")
    return live_path


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 = dingtalk_download(manifest, max_bytes)
    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)