#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ mm_cli.py — multimodal-skill CLI（纯 Python 标准库，零三方依赖） 为纯文本 LLM（如 DeepSeek）提供多模态能力： * doc parse — 文档/PDF/图片 → Markdown（OCR、表格、公式、版面） * image ask — 图片理解与问答（描述、图表、截图 QA） * doctor — 各 provider 健康检查（连通性 / key / 模型） * cache — 结果缓存管理（stats / clear） 设计依据 research.md §9： - 能力分裂：文档走专用解析（PaddleOCR 官方 API / MinerU / DeepSeek-OCR）， 图片走通用 VLM（智谱 GLM-4V-Flash 等，OpenAI 兼容）。 - 错误自适应：模型名过期→候选列表轮换；限流/队列满→退避+切 provider； 鉴权失败→明确指引（exit 3）；探测结果缓存 6h。 - 缓存：sha256(文件字节)|provider|model|sha256(prompt)|参数签名；TTL 与 LRU。 环境变量（官方命名）： ZHIPUAI_API_KEY / PADDLEOCR_ACCESS_TOKEN / MINERU_API_TOKEN / SILICONFLOW_API_KEY / DASHSCOPE_API_KEY 优先级：CLI 参数 > 系统环境变量 > skill 目录 .env 文件。 输出纪律：结果→stdout；进度/诊断→stderr；退出码：0 成功 / 1 运行时 / 2 用法 / 3 鉴权 / 4 限流配额 / 5 模型不可用 / 6 网络。 """ import argparse import base64 import csv import hashlib import html.parser import json import os import re import sys import time import urllib.error import urllib.parse import urllib.request import uuid import zipfile # ---- Windows 下强制 UTF-8 stdio（避免编码异常） ---- for stream in (sys.stdout, sys.stderr): try: stream.reconfigure(encoding="utf-8", errors="replace") except Exception: pass __version__ = "0.2.0" # -------------------------------------------------------------------------- # 常量与配置 # -------------------------------------------------------------------------- EXIT_OK = 0 EXIT_ERR = 1 # 运行时/服务端 5xx/未知 EXIT_USAGE = 2 EXIT_AUTH = 3 EXIT_RATE = 4 # 限流 / 配额 / 队列满 EXIT_MODEL = 5 # 模型不存在 / 已下线 EXIT_NET = 6 CACHE_ROOT = os.environ.get( "MM_SKILL_CACHE_DIR", os.path.join(os.path.expanduser("~"), ".cache", "multimodal-skill"), ) RESULTS_DIR = os.path.join(CACHE_ROOT, "results") # provider 结果缓存（json） EXPORT_DIR = os.path.join(CACHE_ROOT, "export") # 超限策略：落盘给 LLM 的解析结果（.md） PROBE_FILE = os.path.join(CACHE_ROOT, "probe.json") CACHE_MAX_ENTRIES = 2000 CACHE_MAX_BYTES = 2 * 1024 * 1024 * 1024 # 2GB CACHE_EVICT_RATIO = 0.8 PROBE_TTL = 6 * 3600 # provider 健康探测缓存 BUSY_TTL = 5 * 60 # 限流后的短时跳过 DOC_TTL_DEFAULT = 30 * 86400 # 文档解析结果默认 30 天 IMAGE_TTL_DEFAULT = 24 * 3600 # 图片问答结果默认 24h # provider 配置：端点、key 环境变量、模型候选（按优先级）、类型 PROVIDERS = { "zhipu": { "kind": "openai_chat", "base": "https://open.bigmodel.cn/api/paas/v4", "env": "ZHIPUAI_API_KEY", "image_models": ["glm-4v-flash", "glm-4.6v-flash", "glm-4.1v-thinking-flash"], "doc_models": [], # 智谱对话接口不做文档解析主力 "note": "免费档（glm-4v-flash 完全免费；4.6v-flash 免费但高峰 429）", }, "siliconflow": { "kind": "openai_chat", "base": "https://api.siliconflow.cn/v1", "env": "SILICONFLOW_API_KEY", "image_models": ["Qwen/Qwen3-VL-8B-Instruct", "Qwen/Qwen3-VL-32B-Instruct"], "doc_models": ["deepseek-ai/DeepSeek-OCR", "PaddlePaddle/PaddleOCR-VL-1.5"], "models_endpoint": "/models", "doc_formats": ["pdf", "image"], "note": "按量计费；DeepSeek-OCR 实测零扣费；PaddleOCR-VL-1.5 亦可作文档 fallback", }, "dashscope": { "kind": "openai_chat", "base": "https://dashscope.aliyuncs.com/compatible-mode/v1", "env": "DASHSCOPE_API_KEY", "image_models": ["qwen-vl-max", "qwen-vl-plus"], "doc_models": ["qwen-vl-ocr"], "doc_formats": ["image"], # qwen-vl-ocr 只吃页图（PDF 需先转图，CLI 不做） "note": "新用户每模型约 100 万 token 免费（90 天，北京地域）", }, "paddleocr": { "kind": "paddleocr_job", "base": "https://paddleocr.aistudio-app.com", "env": "PADDLEOCR_ACCESS_TOKEN", "doc_models": ["PaddleOCR-VL-1.6", "PaddleOCR-VL-1.5", "PaddleOCR-VL", "PP-StructureV3"], "ocr_models": ["PP-OCRv6", "PP-OCRv5"], "max_pages": 100, "doc_formats": ["pdf", "image"], "note": "官方托管 API：每模型 3000 页/日免费；提交→轮询异步任务", }, "mineru": { "kind": "mineru", "base": "https://mineru.net", "env": "MINERU_API_TOKEN", # 仅精准解析用；flash 免 key "flash_max_mb": 10, "flash_max_pages": 20, "doc_formats": ["pdf", "image", "office", "html"], "note": "flash 免 key（IP 限频，≤20 页/10MB）；精准 v4 需 URL（1000 页/日高优）", }, } # 各命令默认 provider 链（可用 config.json 覆盖，见 _load_config） CHAIN_DEFAULTS = { "doc": ["paddleocr", "mineru", "siliconflow"], "image": ["zhipu", "siliconflow", "dashscope"], } # 输出校验限制（可配置：config.json 的 "limits" 段覆盖） LIMITS = { "max_bytes": 5 * 1024 * 1024, # 输出文本 UTF-8 字节数上限 "max_lines": 20000, # 输出行数上限 "max_line_bytes": 4096, # 单行字节上限（防 minified/base64 巨块） "max_tokens": 65536, # 估算 token 上限（默认 64K，CJK 加权估算） "hard_max_bytes": 20 * 1024 * 1024, # 输入文件读取前硬上限（stat 预检，防内存爆炸） } SKILL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) CONFIG_PATHS = [ os.environ.get("MM_SKILL_CONFIG"), os.path.join(os.path.expanduser("~"), ".config", "multimodal-skill", "config.json"), os.path.join(SKILL_ROOT, "config.json"), ] CONFIG_DEFAULT_PATH = CONFIG_PATHS[1] # 默认新建配置的位置 def _strip_jsonc(text): """Remove // and /* */ comments outside string literals (JSONC).""" out, i, n = [], 0, len(text) in_str = False while i < n: c = text[i] if in_str: out.append(c) if c == "\\" and i + 1 < n: out.append(text[i + 1]) i += 2 continue if c == '"': in_str = False i += 1 continue if c == '"': in_str = True out.append(c) i += 1 continue if c == "/" and i + 1 < n and text[i + 1] == "/": while i < n and text[i] != "\n": i += 1 continue if c == "/" and i + 1 < n and text[i + 1] == "*": i += 2 while i + 1 < n and not (text[i] == "*" and text[i + 1] == "/"): i += 1 i += 2 continue out.append(c) i += 1 return "".join(out) # -------------------------------------------------------------------------- # 用户配置（覆盖默认模型候选与 provider 链，支持付费模型） # 说明：宿主（Pi / DeepSeek Harness）无 skill 级配置机制；这里遵循 XDG 惯例： # ~/.config/multimodal-skill/config.json（git 更新 skill 不会覆盖它）。 # 新增 provider 需要代码支持（各 provider 协议不同），配置只能覆盖模型与链，不能增删 provider。 # -------------------------------------------------------------------------- def _load_config(): """Merge existing config files (JSONC allowed): $MM_SKILL_CONFIG > ~/.config/multimodal-skill/config.json > /config.json Schema (see config.example.json): { "doc_chain": [...], // providers serving `doc parse` "image_chain": [...], // providers serving `image ask` "providers": { "": { // only known providers; overrides merge into PROVIDERS "image_models": [...], // tried in order for `image ask` "doc_models": [...], // tried in order for `doc parse` "ocr_models": [...], // tried in order for `doc parse --is-ocr` } } } """ for p in CONFIG_PATHS: if not p or not os.path.isfile(p): continue try: with open(p, "r", encoding="utf-8-sig") as f: data = json.loads(_strip_jsonc(f.read())) if not isinstance(data, dict): continue for k in ("doc_chain", "image_chain"): if isinstance(data.get(k), list) and data[k]: CHAIN_DEFAULTS[k.replace("_chain", "")] = [ x for x in data[k] if x in PROVIDERS] dropped = [x for x in data[k] if x not in PROVIDERS] if dropped: _err(f"warn: config 中的未知 provider 已忽略：{', '.join(dropped)}（provider 列表不可扩展）") for pname, overrides in (data.get("providers") or {}).items(): if pname in PROVIDERS and isinstance(overrides, dict): PROVIDERS[pname] = {**PROVIDERS[pname], **overrides} elif pname not in PROVIDERS: _err(f"warn: 忽略未知 provider 配置：{pname}（provider 列表不可扩展）") for k, v in (data.get("limits") or {}).items(): if k in LIMITS and isinstance(v, (int, float)) and v > 0: LIMITS[k] = int(v) elif k not in LIMITS: _err(f"warn: 忽略未知 limits 配置项：{k}") _log(f"config: loaded {p}") except (OSError, ValueError) as e: _err(f"warn: 配置文件读取失败（{p}）：{e}") def effective_config(): """返回当前生效的完整配置（默认值 + 覆盖），供 config show 使用。""" return { "doc_chain": list(CHAIN_DEFAULTS["doc"]), "image_chain": list(CHAIN_DEFAULTS["image"]), "limits": dict(LIMITS), "providers": { name: {k: list(v) for k, v in cfg.items() if k in ("image_models", "doc_models", "ocr_models") and v} for name, cfg in PROVIDERS.items() }, } # -------------------------------------------------------------------------- # 环境加载（.env：脚本目录 + 父目录，utf-8-sig，注释/引号容错） # -------------------------------------------------------------------------- def _load_env(): for d in (os.path.dirname(os.path.abspath(__file__)), os.path.dirname(os.path.dirname(os.path.abspath(__file__)))): env_path = os.path.join(d, ".env") if not os.path.isfile(env_path): continue try: with open(env_path, "r", encoding="utf-8-sig") as f: for line in f: line = line.strip() if not line or line.startswith("#") or "=" not in line: continue k, v = line.split("=", 1) k, v = k.strip(), v.strip().strip("\"'") if k and k not in os.environ: os.environ[k] = v except OSError as e: _err(f"warn: 读取 .env 失败（{env_path}）：{e}") def _get_key(provider): env = PROVIDERS[provider].get("env") if not env: return None return os.environ.get(env, "").strip() # -------------------------------------------------------------------------- # 工具函数 # -------------------------------------------------------------------------- def _err(msg): print(msg, file=sys.stderr) def _log(msg): print(msg, file=sys.stderr) def http_json(url, method="GET", data=None, headers=None, timeout=60, raw=False): """发起 HTTP 请求。返回 (status, body)。 raw=True 时 body 为原始 bytes；否则尝试解析 JSON，失败保留文本。""" h = dict(headers or {}) req = urllib.request.Request(url, data=data, method=method, headers=h) try: with urllib.request.urlopen(req, timeout=timeout) as resp: body = resp.read() return resp.status, (body if raw else _parse_body(body)) except urllib.error.HTTPError as e: body = e.read() return e.code, (body if raw else _parse_body(body)) except urllib.error.URLError as e: raise MMError("network", f"网络错误：{e.reason}（{url}）") except OSError as e: raise MMError("network", f"网络错误：{e}（{url}）") def _parse_body(body): try: return json.loads(body.decode("utf-8", errors="replace")) except Exception: return body.decode("utf-8", errors="replace") def download(url, timeout=120): """下载远程文件到内存 bytes（带 hard_max_bytes 上限，防内存爆炸；非 200 报错）。""" req = urllib.request.Request(url, method="GET") try: with urllib.request.urlopen(req, timeout=timeout) as resp: cl = resp.headers.get("Content-Length") if cl and int(cl) > LIMITS["hard_max_bytes"]: raise MMError("usage", f"URL 文件过大（{int(cl) / 1048576:.1f}MB > 硬上限 " f"{LIMITS['hard_max_bytes'] // 1048576}MB）：请先下载拆分或换小文件") chunks, total = [], 0 while True: chunk = resp.read(65536) if not chunk: break total += len(chunk) if total > LIMITS["hard_max_bytes"]: raise MMError("usage", f"URL 文件下载超过硬上限 {LIMITS['hard_max_bytes'] // 1048576}MB：已中止") chunks.append(chunk) if resp.status != 200: raise MMError("usage", f"URL 下载失败：HTTP {resp.status}（{url}）") return b"".join(chunks) except urllib.error.HTTPError as e: raise MMError("usage", f"URL 下载失败：HTTP {e.code}（{url}）") except urllib.error.URLError as e: raise MMError("network", f"网络错误：{e.reason}（{url}）") except OSError as e: raise MMError("network", f"网络错误：{e}（{url}）") TIFF_MAGIC = (b"II*\x00", b"MM\x00*") OLE2_MAGIC = b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1" # ---- 文档类型分组（格式感知路由用） ---- DOC_KIND_PDF = "pdf" DOC_KIND_IMAGE = "image" DOC_KIND_OFFICE = "office" # docx/xlsx/pptx（含未知 zip 兜底） DOC_KIND_LEGACY_OFFICE = "legacy_office" # doc/xls/ppt（OLE2 复合文档） DOC_KIND_TEXT = "text" # txt/md/csv/tsv/log/json/yaml/ini… DOC_KIND_HTML = "html" DOC_KIND_UNKNOWN = "unknown" # 扩展名 → 分组（魔数失败时的兜底，兼 URL 场景） EXT_KIND = { ".pdf": DOC_KIND_PDF, ".png": DOC_KIND_IMAGE, ".jpg": DOC_KIND_IMAGE, ".jpeg": DOC_KIND_IMAGE, ".webp": DOC_KIND_IMAGE, ".gif": DOC_KIND_IMAGE, ".bmp": DOC_KIND_IMAGE, ".tif": DOC_KIND_IMAGE, ".tiff": DOC_KIND_IMAGE, ".docx": DOC_KIND_OFFICE, ".xlsx": DOC_KIND_OFFICE, ".pptx": DOC_KIND_OFFICE, ".doc": DOC_KIND_LEGACY_OFFICE, ".xls": DOC_KIND_LEGACY_OFFICE, ".ppt": DOC_KIND_LEGACY_OFFICE, ".txt": DOC_KIND_TEXT, ".md": DOC_KIND_TEXT, ".markdown": DOC_KIND_TEXT, ".csv": DOC_KIND_TEXT, ".tsv": DOC_KIND_TEXT, ".log": DOC_KIND_TEXT, ".json": DOC_KIND_TEXT, ".yaml": DOC_KIND_TEXT, ".yml": DOC_KIND_TEXT, ".ini": DOC_KIND_TEXT, ".conf": DOC_KIND_TEXT, ".text": DOC_KIND_TEXT, ".html": DOC_KIND_HTML, ".htm": DOC_KIND_HTML, } # Office 魔数/Content-Type 标记 → 具体类型 _OFFICE_CT = ( ("wordprocessingml", "application/vnd.openxmlformats-officedocument.wordprocessingml.document"), ("spreadsheetml", "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"), ("presentationml", "application/vnd.openxmlformats-officedocument.presentationml.presentation"), ) def _zip_office_mime(data): """PK zip 内 [Content_Types].xml 细分 docx/xlsx/pptx；读失败回退 application/zip。""" try: with zipfile.ZipFile(io_bytes(data)) as zf: ct = zf.read("[Content_Types].xml")[:8192].decode("utf-8", "replace") for marker, mime in _OFFICE_CT: if marker in ct: return mime except (KeyError, zipfile.BadZipFile, OSError, ValueError): pass return "application/zip" def _looks_text(data): """启发式：前 512B 无 NUL 且控制字符比例低 → 文本。""" sample = data[:512] if not sample or b"\x00" in sample: return False ctrl = sum(1 for b in sample if b < 0x20 and b not in (0x09, 0x0A, 0x0D)) return ctrl / len(sample) < 0.05 def _probe_text(data): """unknown 兜底探测：整文件（前 64KB）解码后可打印比例 ≥95% → 按文本处理。 覆盖：带二进制头部/尾部噪声、UTF-16 无 BOM、编码怪异但主体可读的文件。 代价：大文件只解码前 64KB 判定，实际解析仍走 decode_text 全量。 """ sample = data[:65536] if not sample: return False try: text = decode_text(sample) except Exception: return False if not text.strip(): return False printable = sum(1 for c in text if c.isprintable() or c in "\n\r\t") return printable / max(len(text), 1) >= 0.95 def sniff_mime(data): """魔数嗅探 MIME（不依赖 PIL/文件后缀）。""" if data[:4] == b"\x89PNG": return "image/png" if data[:2] == b"\xff\xd8": return "image/jpeg" if data[:4] == b"RIFF" and data[8:12] == b"WEBP": return "image/webp" if data[:6] in (b"GIF87a", b"GIF89a"): return "image/gif" if data[:2] == b"BM": return "image/bmp" if data[:4] == b"%PDF": return "application/pdf" if data[:4] in TIFF_MAGIC: return "image/tiff" if data[:8] == OLE2_MAGIC: return "application/x-ole-storage" if data[:4] in (b"PK\x03\x04", b"PK\x05\x06", b"PK\x07\x08"): return _zip_office_mime(data) if data[:2] in (b"\xff\xfe", b"\xfe\xff"): return "text/plain" # UTF-16 BOM if _looks_text(data): return "text/plain" return "application/octet-stream" def detect_doc_kind(data, name=""): """魔数优先、扩展名兜底 → 文档类型分组。name 可为本地路径或 URL。""" mime = sniff_mime(data) if mime == "application/pdf": return DOC_KIND_PDF if mime.startswith("image/"): return DOC_KIND_IMAGE if mime in ("application/zip",) + tuple(m for _, m in _OFFICE_CT): return DOC_KIND_OFFICE if mime == "application/x-ole-storage": return DOC_KIND_LEGACY_OFFICE if mime == "text/plain": ext = os.path.splitext((name or "").split("?")[0])[1].lower() if ext in (".html", ".htm"): return DOC_KIND_HTML return DOC_KIND_TEXT ext = os.path.splitext((name or "").split("?")[0])[1].lower() kind = EXT_KIND.get(ext) if kind: return kind # 未知扩展名/魔数 → 先按文本探测（点 3：不直接报错） if _probe_text(data): return DOC_KIND_TEXT return DOC_KIND_UNKNOWN def data_uri(data, mime): return f"data:{mime};base64," + base64.b64encode(data).decode() def is_url(s): return s.startswith(("http://", "https://")) def sha256(b): return hashlib.sha256(b).hexdigest() # -------------------------------------------------------------------------- # 错误分类与自适应 # -------------------------------------------------------------------------- class MMError(Exception): """kind ∈ auth/rate/model/network/retry/usage/unknown""" def __init__(self, kind, message, exit_code=None, detail=None): super().__init__(message) self.kind = kind self.exit_code = exit_code or { "auth": EXIT_AUTH, "rate": EXIT_RATE, "model": EXIT_MODEL, "network": EXIT_NET, "usage": EXIT_USAGE, "retry": EXIT_ERR, "unknown": EXIT_ERR, }[kind] self.detail = detail MODEL_HINTS = re.compile( r"model|模型|not found|不存在|未找到|unknown model|invalid model|no such", re.I, ) RATE_HINTS = re.compile( r"429|1305|10010|访问量过大|队列已满|quota|rate.?limit|限流|频率|繁忙|too many|try again later|稍后再试", re.I, ) AUTH_HINTS = re.compile(r"401|403|unauthorized|invalid.*key|api[ _-]?key|鉴权|无效|token 错误|token 过期", re.I) def classify(status, body_text): """把 HTTP 状态码 + 响应体文本映射为错误类别。""" if status in (401, 403): if RATE_HINTS.search(body_text) and not AUTH_HINTS.search(body_text): return "rate" return "auth" if status == 429: return "rate" if status in (404, 400): if MODEL_HINTS.search(body_text): return "model" if RATE_HINTS.search(body_text): return "rate" if AUTH_HINTS.search(body_text): return "auth" return "usage" if 500 <= status < 600: return "retry" return "unknown" def describe_error(status, body): """返回 (kind, 友好信息)。""" if isinstance(body, bytes): text = body.decode("utf-8", errors="replace") elif isinstance(body, dict): text = json.dumps(body, ensure_ascii=False) else: text = str(body) kind = classify(status, text) snippet = text[:400] return kind, f"HTTP {status}: {snippet}" # -------------------------------------------------------------------------- # 缓存（§9.5：sha256 key、TTL、LRU、探测缓存） # -------------------------------------------------------------------------- def _cache_dir(provider): d = os.path.join(RESULTS_DIR, provider) os.makedirs(d, exist_ok=True) return d def cache_key(file_bytes, provider, model, prompt, params): sig = json.dumps(params, sort_keys=True, ensure_ascii=False) raw = "|".join([sha256(file_bytes), provider, model, sha256(prompt.encode()), sig]) return sha256(raw.encode()) def cache_get(cache_key_, provider, now=None): now = now if now is not None else time.time() path = os.path.join(_cache_dir(provider), cache_key_ + ".json") if not os.path.isfile(path): return None try: with open(path, "r", encoding="utf-8") as f: ent = json.load(f) except (OSError, ValueError): return None if ent.get("expires_at", 0) < now: return None ent["last_access"] = now try: with open(path, "w", encoding="utf-8") as f: json.dump(ent, f, ensure_ascii=False) except OSError: pass return ent def cache_put(cache_key_, provider, result, ttl, usage=None, model=None, meta=None): now = time.time() ent = { "created_at": now, "last_access": now, "expires_at": now + ttl, "result": result, "usage": usage or {}, "model": model or "", "meta": meta or {}, } try: with open(os.path.join(_cache_dir(provider), cache_key_ + ".json"), "w", encoding="utf-8") as f: json.dump(ent, f, ensure_ascii=False) _evict_if_needed() except OSError as e: _err(f"warn: 缓存写入失败：{e}") def _evict_if_needed(): """LRU：条目数/体积超限时按 last_access 淘汰到 80%。""" try: entries = [] total = 0 for prov in os.listdir(RESULTS_DIR): pd = os.path.join(RESULTS_DIR, prov) if not os.path.isdir(pd): continue for fn in os.listdir(pd): if not fn.endswith(".json"): continue p = os.path.join(pd, fn) try: st = os.stat(p) except OSError: continue entries.append((p, st.st_mtime, st.st_size)) total += st.st_size if len(entries) <= CACHE_MAX_ENTRIES and total <= CACHE_MAX_BYTES: return entries.sort(key=lambda x: x[1]) # 最旧在前 target_count = int(CACHE_MAX_ENTRIES * CACHE_EVICT_RATIO) target_size = int(CACHE_MAX_BYTES * CACHE_EVICT_RATIO) count = len(entries) for p, _, size in entries: if count <= target_count and total <= target_size: break try: os.remove(p) count -= 1 total -= size except OSError: pass except OSError: pass def cache_stats(): total, count = 0, 0 oldest = newest = None for prov in os.listdir(RESULTS_DIR): pd = os.path.join(RESULTS_DIR, prov) if not os.path.isdir(pd): continue for fn in os.listdir(pd): if not fn.endswith(".json"): continue p = os.path.join(pd, fn) try: st = os.stat(p) except OSError: continue total += st.st_size count += 1 oldest = min(oldest, st.st_mtime) if oldest else st.st_mtime newest = max(newest, st.st_mtime) if newest else st.st_mtime return {"entries": count, "bytes": total, "oldest": oldest, "newest": newest} def cache_clear(): removed = 0 for prov in os.listdir(RESULTS_DIR): pd = os.path.join(RESULTS_DIR, prov) if not os.path.isdir(pd): continue for fn in os.listdir(pd): try: os.remove(os.path.join(pd, fn)) removed += 1 except OSError: pass # 超限落盘结果一并清除 if os.path.isdir(EXPORT_DIR): for fn in os.listdir(EXPORT_DIR): try: os.remove(os.path.join(EXPORT_DIR, fn)) removed += 1 except OSError: pass return removed # ---- 探测缓存：provider 健康状态 ---- def probe_get(provider): try: with open(PROBE_FILE, "r", encoding="utf-8") as f: probes = json.load(f) except (OSError, ValueError): return None p = probes.get(provider) if not p: return None if p.get("expires_at", 0) < time.time(): return None return p def probe_put(provider, ok, note, ttl=PROBE_TTL): try: probes = {} if os.path.isfile(PROBE_FILE): with open(PROBE_FILE, "r", encoding="utf-8") as f: probes = json.load(f) probes[provider] = {"ok": ok, "note": note, "at": time.time(), "expires_at": time.time() + ttl} os.makedirs(os.path.dirname(PROBE_FILE), exist_ok=True) with open(PROBE_FILE, "w", encoding="utf-8") as f: json.dump(probes, f, ensure_ascii=False) except OSError as e: _err(f"warn: 探测缓存写入失败：{e}") def probe_skip(provider): """返回 True 表示该 provider 在探测缓存中被标记为不可用（未过期）。""" p = probe_get(provider) return bool(p and not p.get("ok")) # -------------------------------------------------------------------------- # OpenAI 兼容 chat（智谱 / 硅基流动 / 百炼） # -------------------------------------------------------------------------- def openai_chat(provider, model, prompt, image_url, mime=None, detail=None, timeout=120, max_tokens=None, doc_prompt=None): base = PROVIDERS[provider]["base"] key = _get_key(provider) if not key: raise MMError("auth", f"[{provider}] 缺少 API key：请设置环境变量 {PROVIDERS[provider]['env']}（或 skill 目录 .env）") content = [] if mime and image_url.startswith("data:"): content.append({"type": "image_url", "image_url": {"url": image_url, **({"detail": detail} if detail else {})}}) else: content.append({"type": "image_url", "image_url": {"url": image_url}}) content.append({"type": "text", "text": prompt}) payload = { "model": model, "messages": [{"role": "user", "content": content}], } if max_tokens: payload["max_tokens"] = max_tokens st, body = http_json(f"{base}/chat/completions", "POST", data=json.dumps(payload).encode(), headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"}, timeout=timeout) if st != 200: kind, msg = describe_error(st, body) raise MMError(kind, f"[{provider}/{model}] {msg}") if isinstance(body, dict): try: text = body["choices"][0]["message"]["content"] usage = body.get("usage", {}) return text, usage except (KeyError, IndexError, TypeError): raise MMError("unknown", f"[{provider}/{model}] 响应结构异常：{json.dumps(body, ensure_ascii=False)[:300]}") raise MMError("unknown", f"[{provider}/{model}] 响应非 JSON：{str(body)[:300]}") # -------------------------------------------------------------------------- # PaddleOCR 官方 API（异步任务：提交 → 轮询 → JSONL 结果） # -------------------------------------------------------------------------- def paddleocr_parse(source, model="PaddleOCR-VL-1.6", pages=None, timeout=600, payload_opts=None): key = _get_key("paddleocr") if not key: raise MMError("auth", "[paddleocr] 缺少 API key：请设置 PADDLEOCR_ACCESS_TOKEN") base = PROVIDERS["paddleocr"]["base"] jobs_url = f"{base}/api/v2/ocr/jobs" auth = {"Authorization": f"Bearer {key}"} opts = {"prettifyMarkdown": True, **(payload_opts or {})} # 1) 提交 job_id = None for attempt in range(3): if is_url(source): payload = {"fileUrl": source, "model": model, "optionalPayload": json.dumps(opts)} if pages: payload["pageRanges"] = pages st, body = http_json(jobs_url, "POST", data=json.dumps(payload).encode(), headers={**auth, "Content-Type": "application/json"}, timeout=180) else: if not os.path.isfile(source): raise MMError("usage", f"文件不存在：{source}") boundary = "----mm" + uuid.uuid4().hex buf = bytearray() def field(name, value): buf.extend(b"--" + boundary.encode() + b"\r\n") buf.extend(f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode()) buf.extend(value.encode() + b"\r\n") field("model", model) field("optionalPayload", json.dumps(opts)) if pages: field("pageRanges", pages) with open(source, "rb") as f: filedata = f.read() buf.extend(b"--" + boundary.encode() + b"\r\n") buf.extend(f'Content-Disposition: form-data; name="file"; filename="{os.path.basename(source)}"\r\n'.encode()) buf.extend(f"Content-Type: {sniff_mime(filedata)}\r\n\r\n".encode()) buf.extend(filedata + b"\r\n") buf.extend(b"--" + boundary.encode() + b"--\r\n") st, body = http_json(jobs_url, "POST", data=bytes(buf), headers={**auth, "Content-Type": f"multipart/form-data; boundary={boundary}"}, timeout=180) if st == 200: try: job_id = body["data"]["jobId"] break except (KeyError, TypeError): pass kind, msg = describe_error(st, body) if kind == "rate" and attempt < 2: _log(f"[paddleocr] 提交队列繁忙（{msg[:120]}），{10 + attempt * 10}s 后重试…") time.sleep(10 + attempt * 10) continue raise MMError(kind, f"[paddleocr] 提交失败：{msg}") if not job_id: raise MMError("unknown", "[paddleocr] 提交失败：未获得 jobId") # 2) 轮询 deadline = time.time() + timeout last_log = 0.0 while time.time() < deadline: st, body = http_json(f"{jobs_url}/{job_id}", "GET", headers=auth, timeout=60) if st != 200: kind, msg = describe_error(st, body) raise MMError(kind, f"[paddleocr] 状态查询失败：{msg}") state = body.get("data", {}).get("state", "").lower() now = time.time() if now - last_log > 30: prog = body.get("data", {}).get("extractProgress", {}) _log(f"[paddleocr] job={job_id} state={state} " f"progress={prog.get('extractedPages')}/{prog.get('totalPages')} elapsed={int(now - (deadline - timeout))}s") last_log = now if state == "done": break if state in ("failed", "error"): raise MMError("unknown", f"[paddleocr] 任务失败：{json.dumps(body.get('data', {}), ensure_ascii=False)[:300]}") time.sleep(5) else: raise MMError("retry", "[paddleocr] 轮询超时（任务仍在队列）") # 3) 取 JSONL 结果 try: json_url = body["data"]["resultUrl"]["jsonUrl"] except (KeyError, TypeError): raise MMError("unknown", "[paddleocr] 结果 URL 缺失") st, raw = http_json(json_url, timeout=60, raw=True) if st != 200: raise MMError("unknown", f"[paddleocr] 结果下载失败 HTTP {st}") return parse_paddleocr_result(raw, model) def parse_paddleocr_result(raw, model): """JSONL → (markdown文本, usage信息)。兼容 doc_parsing 与 OCR 两种结果形状。""" pages_md, usage = [], {"pages": 0} for line in raw.decode("utf-8", errors="replace").splitlines(): line = line.strip() if not line: continue try: d = json.loads(line) except ValueError: continue usage["pages"] += 1 if d.get("errorCode") not in (0, None): raise MMError("unknown", f"[paddleocr] 结果错误：{d.get('errorMsg')}") result = d.get("result") or {} lpr = result.get("layoutParsingResults") or [] for page in lpr: md = page.get("markdown") or {} text = md.get("text") or "" if not text: pr = page.get("prunedResult") or {} text = pr.get("markdownText") or "" if not text and isinstance(pr.get("parsing_res_list"), list): text = "\n".join( str(e.get("markdownText", "")) for e in pr["parsing_res_list"] if isinstance(e, dict) and e.get("markdownText")) if text: pages_md.append(text) if not pages_md: raise MMError("unknown", "[paddleocr] 未解析出任何文本（页面为空或模型不适用）") return "\n\n".join(pages_md), usage # -------------------------------------------------------------------------- # MinerU（flash 免 key 上传；v4 精准解析走 URL） # -------------------------------------------------------------------------- def mineru_flash(source, language="ch", enable_table=True, enable_formula=True, is_ocr=False, page_range=None, timeout=600): base = PROVIDERS["mineru"]["base"] if is_url(source): # URL 走 parse/url 接口 payload = {"url": source, "language": language, "enable_table": enable_table, "enable_formula": enable_formula, "is_ocr": is_ocr} st, body = http_json(f"{base}/api/v1/agent/parse/url", "POST", data=json.dumps(payload).encode(), headers={"Content-Type": "application/json"}, timeout=120) if st != 200: kind, msg = describe_error(st, body) raise MMError(kind, f"[mineru-flash] 提交失败：{msg}") task_id = body["data"]["task_id"] return _mineru_poll(f"{base}/api/v1/agent/parse/{task_id}", timeout, token=None) # 本地文件：两步流程（建任务 → PUT OSS） if not os.path.isfile(source): raise MMError("usage", f"文件不存在：{source}") size = os.path.getsize(source) if size > PROVIDERS["mineru"]["flash_max_mb"] * 1024 * 1024: raise MMError("usage", f"[mineru-flash] 文件 {size / 1048576:.1f}MB 超过 flash 上限 {PROVIDERS['mineru']['flash_max_mb']}MB（请换 PaddleOCR 或精准 API）") payload = {"file_name": os.path.basename(source), "language": language, "enable_table": enable_table, "enable_formula": enable_formula, "is_ocr": is_ocr} if page_range: payload["page_range"] = page_range st, body = http_json(f"{base}/api/v1/agent/parse/file", "POST", data=json.dumps(payload).encode(), headers={"Content-Type": "application/json"}, timeout=120) if st != 200: kind, msg = describe_error(st, body) raise MMError(kind, f"[mineru-flash] 建任务失败：{msg}") data = body["data"] task_id, upload_url = data["task_id"], data["file_url"] with open(source, "rb") as f: filedata = f.read() # 关键坑：PUT 必须带空 Content-Type（任何真实值都会 OSS 签名 403） st, rbody = http_json(upload_url, "PUT", data=filedata, headers={"Content-Type": ""}, timeout=300) if st != 200: kind, msg = describe_error(st, rbody) raise MMError(kind, f"[mineru-flash] 文件上传失败：{msg}") return _mineru_poll(f"{base}/api/v1/agent/parse/{task_id}", timeout, token=None) def mineru_precision(url, model_version="vlm", language="ch", enable_table=True, enable_formula=True, is_ocr=False, timeout=900): key = _get_key("mineru") if not key: raise MMError("auth", "[mineru] 缺少 API key：请设置 MINERU_API_TOKEN") if not is_url(url): raise MMError("usage", "[mineru-precision] v4 接口只接受 http(s) URL（不支持本地文件直传）；本地文件请用 flash 或 PaddleOCR") base = PROVIDERS["mineru"]["base"] auth = {"Authorization": f"Bearer {key}", "Content-Type": "application/json"} payload = {"url": url, "model_version": model_version, "language": language, "enable_table": enable_table, "enable_formula": enable_formula, "is_ocr": is_ocr} st, body = http_json(f"{base}/api/v4/extract/task", "POST", data=json.dumps(payload).encode(), headers=auth, timeout=180) if st != 200: kind, msg = describe_error(st, body) raise MMError(kind, f"[mineru-precision] 提交失败：{msg}") task_id = body["data"]["task_id"] return _mineru_poll(f"{base}/api/v4/extract/task/{task_id}", timeout, token=key, precision=True) def _mineru_poll(status_url, timeout, token, precision=False): deadline = time.time() + timeout headers = {"Authorization": f"Bearer {token}"} if token else {} while time.time() < deadline: st, body = http_json(status_url, "GET", headers=headers, timeout=60) if st != 200: kind, msg = describe_error(st, body) raise MMError(kind, f"[mineru] 轮询失败：{msg}") data = body.get("data", {}) state = str(data.get("state", "")).lower() if state in ("done", "success", "finished", "completed"): if precision: return _mineru_precision_result(data) md_url = data.get("markdown_url") if not md_url: raise MMError("unknown", f"[mineru] 结果缺少 markdown_url：{json.dumps(data, ensure_ascii=False)[:300]}") st2, raw = http_json(md_url, timeout=120, raw=True) if st2 != 200: raise MMError("unknown", f"[mineru] Markdown 下载失败 HTTP {st2}") text = raw.decode("utf-8", errors="replace") if not text.strip(): raise MMError("unknown", "[mineru] Markdown 为空（文件可能无法解析）") return text, {"pages": None} if state in ("failed", "error"): raise MMError("unknown", f"[mineru] 任务失败：{json.dumps(data, ensure_ascii=False)[:300]}") time.sleep(4) raise MMError("retry", "[mineru] 轮询超时") def _mineru_precision_result(data): """v4 结果：zip 包内含 Markdown。返回 (文本, usage)。""" zip_url = None for k in ("extract_result", "zip_url", "result", "result_url"): v = data.get(k) if isinstance(v, str) and v.startswith("http"): zip_url = v break if isinstance(v, dict): for vk in ("zip_url", "url", "result_url"): if isinstance(v.get(vk), str) and v[vk].startswith("http"): zip_url = v[vk] break if not zip_url: raise MMError("unknown", f"[mineru-precision] 结果字段未知：{json.dumps(data, ensure_ascii=False)[:300]}") st, raw = http_json(zip_url, timeout=180, raw=True) if st != 200: raise MMError("unknown", f"[mineru-precision] 结果下载失败 HTTP {st}") try: zf = zipfile.ZipFile(io_bytes(raw)) except zipfile.BadZipFile: raise MMError("unknown", "[mineru-precision] 结果不是有效 zip 包") md_files = [n for n in zf.namelist() if n.endswith((".md", ".markdown"))] if not md_files: raise MMError("unknown", f"[mineru-precision] zip 内无 Markdown：{zf.namelist()[:10]}") parts = [zf.read(n).decode("utf-8", errors="replace") for n in sorted(md_files)] return "\n\n".join(parts), {"pages": None} def io_bytes(b): import io return io.BytesIO(b) # -------------------------------------------------------------------------- # 链式执行（§9.2 自适应：模型候选轮换 + provider 切换） # -------------------------------------------------------------------------- def build_doc_chain(kind, default_chain): """按文档类型分组构建 provider 链（格式感知路由）。 - office（docx/xlsx/pptx）→ 只保留声明支持 office 的 provider，链首即 mineru； - legacy_office（doc/xls/ppt）→ 无 provider 支持，usage 报错提示转换； - unknown → usage 报错并列出支持格式； - 其余（pdf/image/html）→ 按 provider 能力过滤默认链。 手动 --provider 不走此函数（调试用，直连指定 provider）。 """ if kind == DOC_KIND_LEGACY_OFFICE: raise MMError("usage", "旧版 Office 格式（.doc/.xls/.ppt）不受支持：请先用 WPS/Office 另存为 " ".docx/.xlsx/.pptx，或转 PDF 后再解析") if kind == DOC_KIND_UNKNOWN: raise MMError("usage", "无法识别的文件格式。支持：PDF、常见图片（PNG/JPEG/WebP/GIF/BMP/TIFF）、" "Office（.docx/.xlsx/.pptx）、纯文本（.txt/.md/.csv/.tsv/.log/.json/.yaml）与 .html") chain = [p for p in default_chain if kind in (PROVIDERS[p].get("doc_formats") or ())] if not chain: raise MMError("usage", f"格式 {kind} 没有可用的 provider（默认链：{' → '.join(default_chain)}）") return chain def run_chain(kind, file_ref, prompt, chain, model_hint=None, params=None): """按 provider 链逐个尝试。kind ∈ doc|image。 返回 (provider, model, text, usage)。失败时抛出聚合错误。""" params = params or {} failures = [] kinds = [] for provider in chain: cfg = PROVIDERS[provider] if probe_skip(provider): failures.append(f"{provider}: 探测缓存标记不可用") continue models = _candidate_models(cfg, kind, model_hint) if not models: models = [None] # 无模型概念的 provider（如 mineru）也执行一次 for model in models: net_retried = False retries = 0 while True: try: if kind == "doc": text, usage = _doc_call(provider, cfg, model, file_ref, params) else: text, usage = _image_call(provider, cfg, model, file_ref, prompt, params) probe_put(provider, True, f"{kind}/{model} ok") return provider, model, text, usage except MMError as e: failures.append(f"{provider}/{model}: {e}") kinds.append(e.kind) if e.kind == "model": break # 下一个候选模型 if e.kind == "rate": # 限流可能是模型级（如 4.6v 过载）也可能是 provider 级： # 先试下一个候选模型；全部候选耗尽后由末尾统一标记 busy 并切 provider break if e.kind == "network": if not net_retried: net_retried = True # 瞬时抖动重试一次 time.sleep(2) continue break if e.kind == "auth": probe_put(provider, False, f"auth({e})", ttl=BUSY_TTL) break if e.kind == "retry": if retries < 2: retries += 1 # 5xx/轮询超时最多重试 2 次 time.sleep(3) continue break if e.kind in ("usage", "unknown"): raise # 主观错误不自动换 break probe_put(provider, False, "all models failed", ttl=BUSY_TTL) # 聚合退出码：按主要错误类型 exit_code = EXIT_ERR for k in ("auth", "rate", "model", "network"): if k in kinds: exit_code = MMError(k, "").exit_code break raise MMError("unknown", "所有 provider 均失败：\n " + "\n ".join(failures[:12]), exit_code=exit_code) def _candidate_models(cfg, kind, model_hint): if kind == "doc": pool = (cfg.get("doc_models") or []) + (cfg.get("ocr_models") or []) else: pool = cfg.get("image_models") or [] if model_hint and model_hint in pool: return [model_hint] + [m for m in pool if m != model_hint] if model_hint: return [model_hint] + pool return pool def _doc_call(provider, cfg, model, file_ref, params): """文档解析调用。返回 (text, usage)。""" if provider == "paddleocr": return paddleocr_parse(file_ref, model=model, pages=params.get("pages"), timeout=params.get("timeout", 600), payload_opts=params.get("paddleocr_opts")) if provider == "mineru": if params.get("precision"): return mineru_precision(file_ref, language=params.get("language", "ch")) return mineru_flash(file_ref, language=params.get("language", "ch"), page_range=params.get("pages"), is_ocr=params.get("is_ocr", False)) if provider == "siliconflow": # DeepSeek-OCR：PDF/图片以 data URI 直传，提示词固定为转 Markdown if is_url(file_ref): image_url = file_ref else: if not os.path.isfile(file_ref): raise MMError("usage", f"文件不存在：{file_ref}") with open(file_ref, "rb") as f: data = f.read() mime = sniff_mime(data) image_url = data_uri(data, mime) prompt = params.get("doc_prompt") or ( "Convert the document to markdown. Keep tables and formulas as Markdown/LaTeX." if "pdf" in (sniff_mime(data) if not is_url(file_ref) else "application/pdf") else "Convert the document to markdown.") text, usage = openai_chat(provider, model, prompt, image_url, timeout=params.get("timeout", 180), max_tokens=params.get("max_tokens")) return text, usage if provider == "dashscope": if is_url(file_ref): image_url = file_ref else: with open(file_ref, "rb") as f: data = f.read() image_url = data_uri(data, sniff_mime(data)) text, usage = openai_chat(provider, model, "Convert the document to markdown. Keep tables and formulas.", image_url, timeout=params.get("timeout", 180), max_tokens=params.get("max_tokens")) return text, usage raise MMError("unknown", f"[{provider}] 不支持文档解析") def _image_call(provider, cfg, model, file_ref, prompt, params): if is_url(file_ref): image_url = file_ref else: if not os.path.isfile(file_ref): raise MMError("usage", f"文件不存在：{file_ref}") with open(file_ref, "rb") as f: data = f.read() mime = sniff_mime(data) if mime not in ("image/png", "image/jpeg", "image/webp", "image/gif", "image/bmp"): raise MMError("usage", f"不支持的图片格式：{mime}（image ask 只接受图片；PDF/文档请用 doc parse）") image_url = data_uri(data, mime) text, usage = openai_chat(provider, model, prompt, image_url, detail=params.get("detail"), timeout=params.get("timeout", 120), max_tokens=params.get("max_tokens")) return text, usage # -------------------------------------------------------------------------- # doctor # -------------------------------------------------------------------------- def cmd_doctor(provider_filter=None): # 先重置探测缓存：doctor 即重新体检 try: if os.path.isfile(PROBE_FILE): os.remove(PROBE_FILE) except OSError: pass rows = [] for provider in sorted(PROVIDERS): if provider_filter and provider not in provider_filter: continue cfg = PROVIDERS[provider] env = cfg.get("env") key = _get_key(provider) row = {"provider": provider, "kind": cfg["kind"], "key": bool(key), "note": cfg.get("note", "")} if env and not key: row["status"] = "no-key" rows.append(row) continue try: if provider == "zhipu": # 真实鉴权探测：免费文本模型 ping（1 token） payload = {"model": "glm-4-flash", "messages": [{"role": "user", "content": "ping"}], "max_tokens": 1} st, body = http_json(cfg["base"] + "/chat/completions", "POST", data=json.dumps(payload).encode(), headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"}, timeout=30) if st == 200: row["status"] = "ok" else: kind, msg = describe_error(st, body) row["status"] = f"{kind}: {msg[:80]}" elif provider == "paddleocr": # 路由探测：GET jobs → 405 表示服务与 key 头已受理（405 为正常） st, _ = http_json(cfg["base"] + "/api/v2/ocr/jobs", headers={"Authorization": f"Bearer {key}"}, timeout=20) row["http"] = st row["status"] = "ok" if st in (200, 405, 401) else f"unexpected http {st}" else: st, _ = http_json(cfg["base"], timeout=15) row["reachable"] = True row["http"] = st if cfg.get("models_endpoint"): st2, body = http_json(cfg["base"] + cfg["models_endpoint"], headers={"Authorization": f"Bearer {key}"}, timeout=20) if st2 == 200 and isinstance(body, dict): models = [m.get("id") for m in body.get("data", [])] row["model_count"] = len(models) row["models"] = models[:8] else: row["models_err"] = st2 row["status"] = "ok" except MMError as e: row["reachable"] = False row["status"] = f"unreachable: {e}" rows.append(row) for r in rows: status = r.pop("status") print(f"[{r['provider']}] {status}" + (f" key={'✓' if r.get('key') else '✗'}" if "key" in r else "") + (f" http={r.get('http')}" if r.get("http") else "") + (f" models={r.get('model_count')}" if r.get("model_count") else "") + (f" {r['note']}" if r.get("note") else "")) if r.get("models"): print(f" -> {', '.join(r['models'])}") # -------------------------------------------------------------------------- # 主入口 # -------------------------------------------------------------------------- def _read_or_download(file_ref, allow_pdf=True): """本地文件→bytes；URL→下载。""" if is_url(file_ref): return download(file_ref) if not os.path.isfile(file_ref): raise MMError("usage", f"文件不存在：{file_ref}") with open(file_ref, "rb") as f: return f.read() # -------------------------------------------------------------------------- # 本地全文解析（txt/md/csv/html —— 零模型、零配额；调研见 references/formats.md） # -------------------------------------------------------------------------- LOCAL_ENCODINGS = ("utf-8-sig", "utf-8", "gb18030", "latin-1") LOCAL_MAX_ROWS = 5000 # csv/html 表格行数上限 LOCAL_MAX_COLS = 100 # 列数上限 LOCAL_MIN_TEXT = 20 # 提取结果低于此长度视为"需模型"（JS 渲染页等） def decode_text(data): """编码探测：UTF-16 BOM → UTF-16 无 BOM 启发式 → UTF-8(BOM) → GB18030(GBK 超集) → Latin-1 兜底。""" if data[:2] == b"\xff\xfe": try: return data.decode("utf-16-le") except UnicodeDecodeError: pass if data[:2] == b"\xfe\xff": try: return data.decode("utf-16-be") except UnicodeDecodeError: pass # UTF-16 无 BOM 启发式：ASCII 字符高位字节为 0 → 奇/偶字节位 0x00 比例失衡。 # （必须抢在 gb18030 之前：gb18030 全字节映射、永不失败，会把 UTF-16 解成乱码） n = len(data) if n >= 8: evens, odds = data[0::2], data[1::2] if odds.count(0) / len(odds) > 0.3 and evens.count(0) / len(evens) < 0.1: try: return data.decode("utf-16-le") except UnicodeDecodeError: pass if evens.count(0) / len(evens) > 0.3 and odds.count(0) / len(odds) < 0.1: try: return data.decode("utf-16-be") except UnicodeDecodeError: pass for enc in LOCAL_ENCODINGS: try: return data.decode(enc) except (UnicodeDecodeError, LookupError): continue return data.decode("latin-1", errors="replace") def _md_cell(c): return str(c).replace("|", "\\|").replace("\r", " ").replace("\n", " ").strip() def csv_to_markdown(data): """csv/tsv → Markdown 表格（csv.Sniffer 自动探测分隔符，兼容引号内逗号/换行）。""" import io text = decode_text(data) try: dialect = csv.Sniffer().sniff(text[:4096], delimiters=",\t;|") except csv.Error: dialect = csv.excel rows = list(csv.reader(io.StringIO(text), dialect)) if not rows: return "" total = len(rows) ncol = min(max((len(r) for r in rows), default=1), LOCAL_MAX_COLS) truncated = total > LOCAL_MAX_ROWS if truncated: rows = rows[:LOCAL_MAX_ROWS] header = rows[0] out = ["| " + " | ".join(_md_cell(c) for c in header[:ncol]) + " |", "|" + "---|" * len(header[:ncol])] for r in rows[1:]: cells = (list(r) + [""] * ncol)[:ncol] out.append("| " + " | ".join(_md_cell(c) for c in cells) + " |") if truncated: out.append(f"\n> 表格已截断：仅展示前 {LOCAL_MAX_ROWS} 行（原 {total} 行）") return "\n".join(out) class _HtmlToMarkdown(html.parser.HTMLParser): """stdlib html.parser → Markdown 提取器（无三方依赖）。 覆盖：标题/段落/列表/表格/链接/图片/代码块/分隔线；丢弃 script/style 等。 输出缓冲为单字符串：内联片段直接拼接（必要时补空格），块级结构显式换行。 JS 渲染页面提取结果会过短 → 由 local_parse 判定回退模型链。 """ _SKIP_TAGS = {"script", "style", "noscript", "template", "meta", "link", "iframe", "svg", "canvas"} _BLOCK_TAGS = {"p", "div", "section", "article", "header", "footer", "aside", "blockquote", "figure", "figcaption", "details", "summary", "address", "form", "fieldset", "main", "nav", "dl", "caption"} def __init__(self): super().__init__(convert_charrefs=True) self.buf = "" self.skip = 0 self.in_pre = False self.in_code = 0 self.in_title = False self.title_parts = [] self.table = None # {"header":[…], "rows":[…], "row":[…], "cell":str} self.list_stack = [] # ["ul"/"ol", …] self.link = None # (href, [text parts]) self._ws = re.compile(r"\s+") # ---- 缓冲原语 ---- _CJK = re.compile(r"[\u4e00-\u9fff\u3000-\u303f\uff00-\uffef]") def _push(self, s): """内联追加：前字符不是空白/标点且非 CJK 相邻时补一个空格。""" if not s: return if (self.buf and not self.buf[-1].isspace() and self.buf[-1] not in "([{`#|*_-" and not (self._CJK.match(self.buf[-1]) and self._CJK.match(s[0]))): self.buf += " " self.buf += s def _newline(self): if self.buf and not self.buf.endswith("\n"): self.buf += "\n" def _para(self): """块级开始：确保前有空行。""" self._newline() if not self.buf.endswith("\n\n"): self.buf += "\n" # ---- 事件 ---- def handle_starttag(self, tag, attrs): a = dict(attrs) if tag in self._SKIP_TAGS: self.skip += 1 return if tag == "title": self.in_title = True self.title_parts = [] return if tag == "pre": self._para() self.in_pre = True self.buf += "```\n" return if tag == "code" and not self.in_pre: self.in_code += 1 self._push("`") return if tag in ("h1", "h2", "h3", "h4", "h5", "h6"): self._para() self.buf += "#" * int(tag[1]) + " " return if tag == "br": self._newline() return if tag == "hr": self._para() self.buf += "---\n" return if tag == "li": self._newline() depth = len(self.list_stack) self.buf += " " * max(0, depth - 1) if self.list_stack and self.list_stack[-1][0] == "ol": self.list_stack[-1][1] += 1 self.buf += f"{self.list_stack[-1][1]}. " else: self.buf += "- " return if tag in ("ul", "ol"): # 嵌套列表内不额外加空行 (self._newline if self.list_stack else self._para)() self.list_stack.append([tag, 0]) return if tag == "a": self.link = (a.get("href", ""), []) return if tag == "img": alt, src = a.get("alt", ""), a.get("src", "") self._push(f"![{alt}]({src})" if src else alt) return if tag == "table": self._para() self.table = {"header": None, "rows": [], "row": None, "cell": ""} return if tag == "tr": if self.table is not None: self.table["row"] = [] return if tag in ("td", "th"): if self.table is not None and self.table["row"] is not None: self.table["cell"] = "" return if tag in self._BLOCK_TAGS: self._para() return def handle_endtag(self, tag): if tag in self._SKIP_TAGS: if self.skip > 0: self.skip -= 1 return if tag == "title" and self.in_title: self.in_title = False t = "".join(self.title_parts).strip() if t: self.buf = f"# {t}\n" + self.buf return if tag == "pre": self.in_pre = False if not self.buf.endswith("\n"): self.buf += "\n" self.buf += "```\n\n" return if tag == "code" and not self.in_pre: if self.in_code > 0: self.in_code -= 1 self.buf += "`" # 闭合反引号紧贴内容，不加空格 return if tag in ("h1", "h2", "h3", "h4", "h5", "h6"): self._newline() return if tag in ("ul", "ol"): if self.list_stack: self.list_stack.pop() self._newline() return if tag == "a" and self.link is not None: href, parts = self.link text = "".join(parts).strip() self.link = None if text: self._push(f"[{text}]({href})" if href else text) return if tag == "table" and self.table is not None: self.buf += self._render_table() + "\n\n" self.table = None return if tag == "tr" and self.table is not None and self.table["row"] is not None: cells = [self._ws.sub(" ", c).strip() for c in self.table["row"]] if self.table["header"] is None: self.table["header"] = cells else: self.table["rows"].append(cells) self.table["row"] = None return if tag in ("td", "th") and self.table is not None and self.table["row"] is not None: self.table["row"].append(self.table["cell"]) return if tag in self._BLOCK_TAGS: self._newline() return def handle_data(self, data): if self.in_title: self.title_parts.append(data) return if self.skip: return if self.link is not None: self.link[1].append(data) return if self.table is not None and self.table["row"] is not None: self.table["cell"] += data return if self.in_pre: self.buf += data return t = self._ws.sub(" ", data).strip() if t: self._push(t) def _render_table(self): t = self.table or {} header = t.get("header") or [] rows = t.get("rows") or [] if not header and not rows: return "" ncol = min(max(len(header), max((len(r) for r in rows), default=0)), LOCAL_MAX_COLS) trunc = len(rows) > LOCAL_MAX_ROWS pad = [""] * ncol out = ["| " + " | ".join(_md_cell((header + pad)[i]) for i in range(ncol)) + " |", "|" + "---|" * ncol] for r in rows[:LOCAL_MAX_ROWS]: cells = (list(r) + pad)[:ncol] out.append("| " + " | ".join(_md_cell(c) for c in cells) + " |") if trunc: out.append(f"\n> 表格已截断：仅展示前 {LOCAL_MAX_ROWS} 行") return "\n".join(out) def result(self): return re.sub(r"\n{3,}", "\n\n", self.buf).strip() def local_parse(kind, data, name=""): """按类型本地解析；返回 (text, note) 或 None（类型不走本地 / 结果不可用）。 text 组直接解码直读；csv 转 Markdown 表格；html 经 html.parser 提取为 Markdown。 提取结果过短（JS 渲染页等）→ 返回 None 由调用方回退模型链。 """ if kind == DOC_KIND_TEXT: ext = os.path.splitext((name or "").split("?")[0])[1].lower() if ext in (".csv", ".tsv"): md = csv_to_markdown(data) return (md, "CSV/TSV→Markdown 表格（本地解析，零配额）") if md.strip() else None if re.search(r"!\[.*?\]\([^)]*\)", decode_text(data[:8192])): _log("warn: 文本内嵌图片只能提取引用，图片内容需视觉模型（可用 --provider 调试指定）") text = decode_text(data) if not text.strip(): return None return text, "纯文本直读（本地解析，零配额）" if kind == DOC_KIND_HTML: try: p = _HtmlToMarkdown() p.feed(decode_text(data)) p.close() md = p.result() except Exception as e: _log(f"warn: HTML 本地解析失败（{e}），回退模型链") return None if len(md) < LOCAL_MIN_TEXT: _log(f"warn: HTML 本地提取过短（{len(md)} 字符，可能为 JS 渲染页），回退模型链") return None return md, "HTML→Markdown（本地解析，零配额）" return None # -------------------------------------------------------------------------- # 输出校验：三校验（字节/行数/单行长）+ token 估算 # 超限策略（已定）：全文不输出，落盘 UTF-8 文件并返回路径，由消费方 LLM 自行读取。 # -------------------------------------------------------------------------- _CJK_REMOVE_TBL = None def _cjk_remove_table(): """懒构建：仅 CJK 码点 → None 的小翻译表（~4 万条目，4.6MB），C 级表遍历。 勿用全表（111 万条目 ≈ 133MB 常驻）。基准：5MB 文本估算 ≈96ms。""" global _CJK_REMOVE_TBL if _CJK_REMOVE_TBL is None: cps = [] for lo, hi in ((0x4E00, 0x9FFF), (0x3400, 0x4DBF), (0xF900, 0xFAFF), (0x3040, 0x30FF), (0xAC00, 0xD7AF)): cps.extend(range(lo, hi + 1)) _CJK_REMOVE_TBL = str.maketrans({cp: None for cp in cps}) return _CJK_REMOVE_TBL def estimate_tokens(text): """轻量 token 估算：CJK ≈ 1 token/字，其余 ≈ 1/3.5 token/字符。 精确统计 CJK 字符数（translate 小表），误差仅来自换算系数；护栏用途足够。""" n = len(text) n_cjk = n - len(text.translate(_cjk_remove_table())) return int(n_cjk + (n - n_cjk) / 3.5) def validate_output(text): """三校验 + token 估算。返回统计 dict（全部为事实测量，供 meta/告警使用）： {chars, bytes, lines, max_line_bytes, tokens, over:{bytes,lines,line,tokens}} token 估算 fast path：token 数恒 ≤ 字符数（护栏方向安全），小文本不扫描。""" n = len(text) b = len(text.encode("utf-8")) lines = text.splitlines() max_line = max((len(l.encode("utf-8")) for l in lines), default=0) if n < LIMITS["max_tokens"]: tokens = n # 不扫描，字符数即保守上界 else: tokens = estimate_tokens(text) over = { "bytes": b > LIMITS["max_bytes"], "lines": len(lines) > LIMITS["max_lines"], "line": max_line > LIMITS["max_line_bytes"], "tokens": tokens > LIMITS["max_tokens"], } return {"chars": n, "bytes": b, "lines": len(lines), "max_line_bytes": max_line, "tokens": tokens, "over": over} def write_result_file(text, key_bytes, out=None): """超限策略：把解析结果落盘为 UTF-8 文本，返回路径。 --out 指定时用之；否则写 /export/.md。 返回 None 表示写盘失败（调用方回退原样输出）。""" path = out or os.path.join(EXPORT_DIR, sha256(key_bytes) + ".md") try: os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w", encoding="utf-8") as f: f.write(text) return path except OSError as e: _err(f"warn: 结果落盘失败（{e}），回退原样输出") return None def _doc_meta(kind, provider, model, usage): """返回给 LLM 的解析事实元信息（只含准确无误的基本事实，不做推断）。 - format: 检测到的文档类型分组 - mode: local（本地确定性解析）| model（解析模型厂商服务） - provider / model: 实际提供解析的厂商与模型名（local 时 provider="local"） - pages: 实际解析页数（仅厂商返回时给出，如 PaddleOCR） 置信度/质量判断由 LLM 依据这些事实自行决定（SKILL.md 有指引）。 """ meta = { "format": kind, "mode": "local" if provider == "local" else "model", "provider": provider, "model": model, } pages = (usage or {}).get("pages") if isinstance(pages, int): meta["pages"] = pages return meta def cmd_doc_parse(args): params = {"pages": args.pages, "language": args.language, "is_ocr": args.is_ocr, "precision": args.precision, "doc_prompt": args.prompt, "max_tokens": args.max_tokens, "timeout": args.timeout} use_cache = not args.no_cache # 输入硬上限预检：本地文件先 stat，避免大文件读入内存 if not is_url(args.file): if os.path.isfile(args.file) and os.path.getsize(args.file) > LIMITS["hard_max_bytes"]: raise MMError("usage", f"文件过大（{os.path.getsize(args.file) / 1048576:.1f}MB > 硬上限 " f"{LIMITS['hard_max_bytes'] // 1048576}MB）：请用 --pages 分片或先拆分文件" "（可在 config.json 的 limits.hard_max_bytes 调整）") # 读取/下载字节（缓存键与格式检测共用一次 IO） data = _read_or_download(args.file) if not data: raise MMError("usage", f"文件为空：{args.file}") if len(data) > LIMITS["hard_max_bytes"]: raise MMError("usage", f"输入过大（{len(data) / 1048576:.1f}MB > 硬上限 " f"{LIMITS['hard_max_bytes'] // 1048576}MB）：请先拆分或换小文件") # 缓存（§9.5）：本地文件/URL 均按内容 sha256 建 key cache_key_ = None if use_cache: cache_key_ = cache_key(data, "doc", args.provider or "auto", args.prompt or "", {k: v for k, v in params.items() if v}) hit = cache_get(cache_key_, "doc") if hit: meta = hit.get("meta") or {} if not meta: # 旧缓存条目（无 meta）：从 model 字段反向推导基本事实 m = hit.get("model") or "" meta = {"format": None, "mode": "local" if m == "local" else "model", "provider": m.split("/")[0] if "/" in m else (m or "doc"), "model": m.split("/", 1)[1] if "/" in m else None} if any((meta.get("over") or {}).values()): # 缓存命中且历史判定超限：同样只给落盘路径，不输出全文 if args.json: print(json.dumps({"cache": True, "provider": meta.get("provider"), "model": meta.get("model"), "meta": meta, "path": (meta.get("paths") or {}).get("result"), "text": None}, ensure_ascii=False)) else: print(f"\n") print("输出超限（历史判定），全文未输出，请用你的 read/grep 工具读取：") for k, p in (meta.get("paths") or {}).items(): print(f" {k}: {p}") return EXIT_OK if args.json: print(json.dumps({"cache": True, "provider": meta.get("provider"), "model": meta.get("model"), "meta": meta, "text": hit["result"]}, ensure_ascii=False)) else: print(f"\n") print(hit["result"]) return EXIT_OK provider, model, text, usage = None, None, None, {} kind = detect_doc_kind(data, args.file) # 仅作事实元信息；调试模式下不参与路由 if args.provider: # 手动指定 provider（调试用）：直连，不做格式路由 chain = [args.provider] provider, model, text, usage = run_chain( "doc", args.file, "", chain, model_hint=args.model, params=params) else: _log(f"[doc] 格式检测：{kind}（{sniff_mime(data)}）") local = local_parse(kind, data, args.file) if local is not None: text, note = local provider, model = "local", None _log(f"[doc] {note}") for flag, val in (("--pages", args.pages), ("--is-ocr", args.is_ocr), ("--precision", args.precision)): if val: _err(f"warn: {flag} 对本地解析无效，已忽略（本地解析无需模型）") else: chain = build_doc_chain(kind, CHAIN_DEFAULTS["doc"]) _log(f"[doc] 模型链：{' → '.join(chain)}") provider, model, text, usage = run_chain( "doc", args.file, "", chain, model_hint=args.model, params=params) meta = _doc_meta(kind, provider, model, usage) # 输出校验：三校验 + token 估算 stats = validate_output(text) meta["stats"] = {k: v for k, v in stats.items() if k != "over"} meta["over"] = stats["over"] over = any(stats["over"].values()) # 超限策略（已定）：全文落盘，只返回路径给 LLM，由 LLM 用自带 read/grep 工具取片段 if over: rpath = write_result_file(text, data, args.out) if rpath: meta["paths"] = {"result": rpath} if not is_url(args.file): meta["paths"]["source"] = args.file _err(f"warn: 输出超限 {json.dumps(stats['over'], ensure_ascii=False)}" f"（bytes={stats['bytes']} lines={stats['lines']}" f" max_line={stats['max_line_bytes']} tokens≈{stats['tokens']}）" f"—— 全文已落盘：{rpath}，交由 LLM 按需读取") if use_cache and cache_key_: cache_put(cache_key_, "doc", text, ttl=args.ttl or DOC_TTL_DEFAULT, usage=usage, model=f"{provider}/{model}" if provider != "local" else "local", meta=meta) if over and meta.get("paths", {}).get("result"): # 不输出全文；给出落盘路径（result=解析结果，source=原始本地文件） if args.json: print(json.dumps({"cache": False, "provider": provider, "model": model, "usage": usage, "meta": meta, "path": meta["paths"]["result"], "text": None}, ensure_ascii=False)) else: print(f"\n") print(f"输出超限（bytes={stats['bytes']} lines={stats['lines']}" f" tokens≈{stats['tokens']}），全文未输出，已落盘：") for k, p in meta["paths"].items(): print(f" {k}: {p}") print("请用你的 read/grep 工具按需读取片段。") return EXIT_OK if args.out: with open(args.out, "w", encoding="utf-8") as f: f.write(text) _log(f"结果已写入 {args.out}（{len(text)} 字符）") if args.json: print(json.dumps({"cache": False, "provider": provider, "model": model, "usage": usage, "meta": meta, "text": text}, ensure_ascii=False)) else: # 文本模式：头部附一行解析事实元信息（HTML 注释，LLM 可见、渲染不可见） print(f"\n") print(text) return EXIT_OK def cmd_image_ask(args): params = {"detail": args.detail, "max_tokens": args.max_tokens, "timeout": args.timeout} chain = [args.provider] if args.provider else list(CHAIN_DEFAULTS["image"]) chain = [p for p in chain if p in PROVIDERS] if not chain: raise MMError("usage", f"未知 provider：{args.provider}") if not args.question: raise MMError("usage", "image ask 需要问题参数（请用引号包裹）") use_cache = not args.no_cache cache_key_ = None if use_cache: data = _read_or_download(args.file) cache_key_ = cache_key(data, "image", args.provider or "auto", args.question, params) hit = cache_get(cache_key_, "image") if hit: if args.json: print(json.dumps({"cache": True, "provider": hit.get("model", "").split("/")[0], "model": hit.get("model"), "text": hit["result"]}, ensure_ascii=False)) else: print(hit["result"]) return EXIT_OK provider, model, text, usage = run_chain( "image", args.file, args.question, chain, model_hint=args.model, params=params) # 输出校验（与 doc parse 统一）；超限策略同 doc：落盘返回路径 stats = validate_output(text) meta = {"format": None, "mode": "model", "provider": provider, "model": model, "stats": {k: v for k, v in stats.items() if k != "over"}, "over": stats["over"]} over = any(stats["over"].values()) if over: rpath = write_result_file(text, text.encode("utf-8")) if rpath: meta["paths"] = {"result": rpath} if not is_url(args.file): meta["paths"]["source"] = args.file _err(f"warn: 回答超限 {json.dumps(stats['over'], ensure_ascii=False)}" f"（tokens≈{stats['tokens']}）—— 已落盘：{rpath}") if use_cache and cache_key_: cache_put(cache_key_, "image", text, ttl=args.ttl or IMAGE_TTL_DEFAULT, usage=usage, model=f"{provider}/{model}", meta=meta) if over and meta.get("paths", {}).get("result"): if args.json: print(json.dumps({"cache": False, "provider": provider, "model": model, "usage": usage, "meta": meta, "path": meta["paths"]["result"], "text": None}, ensure_ascii=False)) else: print(f"输出超限（tokens≈{stats['tokens']}），回答未输出，已落盘：") for k, p in meta["paths"].items(): print(f" {k}: {p}") print("请用你的 read/grep 工具按需读取。") return EXIT_OK if args.json: print(json.dumps({"cache": False, "provider": provider, "model": model, "usage": usage, "meta": meta, "text": text}, ensure_ascii=False)) else: print(text) return EXIT_OK def cmd_config(args): """config path|show|open — 配置管理（跨平台打开默认编辑器）""" if args.config_cmd == "path": # 返回当前生效的配置文件；不存在则返回默认新建位置 for p in CONFIG_PATHS: if p and os.path.isfile(p): print(p) return EXIT_OK print(CONFIG_DEFAULT_PATH) return EXIT_OK if args.config_cmd == "show": print(json.dumps(effective_config(), ensure_ascii=False, indent=2)) return EXIT_OK if args.config_cmd == "open": # 找已有配置，否则用模板生成默认配置再打开 path = None for p in CONFIG_PATHS: if p and os.path.isfile(p): path = p break if not path: path = CONFIG_DEFAULT_PATH try: os.makedirs(os.path.dirname(path), exist_ok=True) template = os.path.join(SKILL_ROOT, "config.example.json") if os.path.isfile(template): import shutil shutil.copyfile(template, path) else: with open(path, "w", encoding="utf-8") as f: f.write("{}\n") _log(f"config: 已创建默认配置 {path}") except OSError as e: raise MMError("usage", f"无法创建配置文件 {path}：{e}") return _open_with_default_app(path) raise MMError("usage", f"未知 config 子命令：{args.config_cmd}") def _open_with_default_app(path): """跨平台：用系统默认方式打开文件（编辑器）。 Windows: os.startfile / macOS: `open` / Linux: $VISUAL|$EDITOR 或 xdg-open（后台分离，不阻塞）。""" import subprocess if os.name == "nt": os.startfile(path) # type: ignore[attr-defined] _log(f"config: 已打开 {path}") return EXIT_OK if sys.platform == "darwin": subprocess.Popen(["open", path], start_new_session=True) _log(f"config: 已打开 {path}") return EXIT_OK # Linux/Unix for editor in (os.environ.get("VISUAL"), os.environ.get("EDITOR"), "xdg-open"): if not editor: continue cmd = editor.split() + [path] try: subprocess.Popen(cmd, start_new_session=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) _log(f"config: 已用 {cmd[0]} 打开 {path}") return EXIT_OK except OSError: continue _log(f"config: 未找到可用编辑器，请手动打开：{path}") return EXIT_OK def cmd_providers(args): """providers — 列出全部内置 provider、所属管线与模型候选（列表不可扩展，配置只可覆盖）""" rows = [] for name, cfg in sorted(PROVIDERS.items()): chains = [] if name in CHAIN_DEFAULTS["doc"]: chains.append("doc_chain") if name in CHAIN_DEFAULTS["image"]: chains.append("image_chain") models = [] for k in ("image_models", "doc_models", "ocr_models"): for m in cfg.get(k, []): models.append({"model": m, "pipeline": k.replace("_models", "")}) rows.append({ "provider": name, "kind": cfg["kind"], "key": bool(_get_key(name)), "chains": chains, "models": models, "formats": cfg.get("doc_formats", []), "note": cfg.get("note", ""), }) if args.json: print(json.dumps(rows, ensure_ascii=False, indent=2)) return EXIT_OK for r in rows: print(f"[{r['provider']}] kind={r['kind']} key={'✓' if r['key'] else '✗'} " f"chains={'+'.join(r['chains']) or '-'}") for m in r["models"]: print(f" {m['pipeline']:>5} {m['model']}") if r["formats"]: print(f" formats {', '.join(r['formats'])}") if r["note"]: print(f" note {r['note']}") print("\n提示：provider 列表由代码内置（各 provider 协议不同），不可在配置中新增；" "config.json 只能覆盖模型候选与默认链。") return EXIT_OK def cmd_cache(args): if args.cache_cmd == "stats": s = cache_stats() print(f"entries: {s['entries']} bytes: {s['bytes'] / 1048576:.1f} MB") if s["oldest"]: import datetime print(f"oldest: {datetime.datetime.fromtimestamp(s['oldest'])}") print(f"newest: {datetime.datetime.fromtimestamp(s['newest'])}") return EXIT_OK if args.cache_cmd == "clear": n = cache_clear() try: if os.path.isfile(PROBE_FILE): os.remove(PROBE_FILE) n += 1 except OSError: pass print(f"removed {n} cache files") return EXIT_OK raise MMError("usage", f"未知 cache 子命令：{args.cache_cmd}") def build_parser(): p = argparse.ArgumentParser( prog="mm_cli.py", description="multimodal-skill CLI（stdlib-only）：文档解析 / 图片理解") p.add_argument("--version", action="version", version=__version__) sub = p.add_subparsers(dest="command") d = sub.add_parser("doctor", help="各 provider 健康检查") d.add_argument("--provider", action="append", choices=sorted(PROVIDERS)) d.set_defaults(handler=lambda a: cmd_doctor(a.provider)) dp = sub.add_parser("doc", help="文档解析") dsub = dp.add_subparsers(dest="doc_cmd") dparse = dsub.add_parser("parse", help="文档/PDF/图片 → Markdown") dparse.add_argument("file", help="本地文件路径或 http(s) URL") dparse.add_argument("--provider", choices=sorted(PROVIDERS), help=f"指定 provider（调试用；默认按格式感知自动路由，链：{' → '.join(CHAIN_DEFAULTS['doc'])}）") dparse.add_argument("--model", help="指定模型（默认按候选列表）") dparse.add_argument("--pages", help="页码范围，如 1-20 或 2,4-6（PaddleOCR/MinerU）") dparse.add_argument("--language", default="ch", help="文档语言（MinerU），默认 ch") dparse.add_argument("--is-ocr", action="store_true", help="强制 OCR（扫描件）") dparse.add_argument("--precision", action="store_true", help="MinerU 用精准 v4 API（仅接受 URL，需 MINERU_API_TOKEN）") dparse.add_argument("--prompt", help="自定义文档转换指令（OpenAI 兼容 provider）") dparse.add_argument("--max-tokens", type=int) dparse.add_argument("--timeout", type=int, default=600, help="轮询总超时（秒）") dparse.add_argument("--out", help="结果写入文件") dparse.add_argument("--json", action="store_true", help="结构化输出") dparse.add_argument("--no-cache", action="store_true") dparse.add_argument("--ttl", type=int, help="缓存 TTL（秒），默认 30 天") dparse.set_defaults(handler=cmd_doc_parse) ip = sub.add_parser("image", help="图片理解") isub = ip.add_subparsers(dest="image_cmd") iask = isub.add_parser("ask", help="图片问答/描述") iask.add_argument("file", help="本地图片路径或 http(s) URL") iask.add_argument("question", help="问题（请用引号包裹）") iask.add_argument("--provider", choices=sorted(PROVIDERS), help=f"指定 provider（默认链：{' → '.join(CHAIN_DEFAULTS['image'])}）") iask.add_argument("--model", help="指定模型（默认按候选列表）") iask.add_argument("--detail", choices=["auto", "high", "low"], default=None) iask.add_argument("--max-tokens", type=int) iask.add_argument("--timeout", type=int, default=120) iask.add_argument("--json", action="store_true") iask.add_argument("--no-cache", action="store_true") iask.add_argument("--ttl", type=int, help="缓存 TTL（秒），默认 24h") iask.set_defaults(handler=cmd_image_ask) cp = sub.add_parser("cache", help="缓存管理") csub = cp.add_subparsers(dest="cache_cmd") csub.add_parser("stats", help="缓存统计").set_defaults(handler=cmd_cache) csub.add_parser("clear", help="清空缓存").set_defaults(handler=cmd_cache) cfp = sub.add_parser("config", help="配置管理（path/show/open）") cfsub = cfp.add_subparsers(dest="config_cmd") cfsub.add_parser("path", help="显示当前生效的配置文件路径").set_defaults(handler=cmd_config) cfsub.add_parser("show", help="显示合并后的生效配置").set_defaults(handler=cmd_config) cfsub.add_parser("open", help="用系统默认编辑器打开配置（不存在则先创建）").set_defaults(handler=cmd_config) pv = sub.add_parser("providers", help="列出内置 provider、管线与模型候选") pv.add_argument("--json", action="store_true") pv.set_defaults(handler=cmd_providers) return p def main(argv=None): _load_env() _load_config() parser = build_parser() args = parser.parse_args(argv) if not getattr(args, "handler", None): parser.print_help() return EXIT_USAGE if args.command == "doctor": return cmd_doctor(args.provider) if args.command == "cache": return cmd_cache(args) try: return args.handler(args) except MMError as e: _err(f"error[{e.kind}]: {e}") return e.exit_code except KeyboardInterrupt: _err("interrupted") return 130 if __name__ == "__main__": sys.exit(main())