feat(tools): 2.0 phase 5 wave 2 - pdf depth (21 tools) (#220)

2026-08-03 07:46:42 +02:00 · 2026-06-13 10:18:55 +08:00
parent ae1337901d
commit 2f39e38162
128 changed files with 11381 additions and 778 deletions
@@ -63,6 +63,12 @@ DISPATCHER_PROFILE = os.environ.get("DISPATCHER_PROFILE", "ai")
 DOCS_SCRIPTS = {
    "doc_pagecount",
    "doc_health",
+    "doc_flatten",
+    "doc_redact",
+    "doc_text",
+    "doc_to_word",
+    "doc_metadata",
+    "doc_html_pdf",
 }

 if DISPATCHER_PROFILE == "docs":
@@ -0,0 +1,30 @@
+"""Flatten forms/annotations into page content via PyMuPDF bake.
+Args: {"path": in, "out": out}. Prints {"ok": true}."""
+import json
+import sys
+
+
+def main():
+    args = json.loads(sys.argv[1]) if len(sys.argv) > 1 else {}
+    path, out = args.get("path"), args.get("out")
+    if not path or not out:
+        print(json.dumps({"error": "missing path/out"}))
+        sys.exit(1)
+    try:
+        import fitz
+    except ImportError:
+        print(json.dumps({"error": "PyMuPDF not installed"}))
+        sys.exit(1)
+    try:
+        doc = fitz.open(path)
+        doc.bake()  # widgets + annotations become page content
+        doc.save(out)
+        doc.close()
+        print(json.dumps({"ok": True}))
+    except Exception as exc:  # noqa: BLE001
+        print(json.dumps({"error": str(exc)}))
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()
@@ -0,0 +1,81 @@
+"""HTML or Markdown to PDF via WeasyPrint (no-phone-home posture).
+Remote references (src/href/action/url()) are REJECTED before conversion
+with a clear error listing up to 5 offending URLs. The url_fetcher remains
+as a defense-in-depth backstop: it blocks any reference that slips past
+the scan, raising ValueError so WeasyPrint omits the resource with zero
+outbound requests.
+Args: {"path": in, "out": o, "mode": "html"|"markdown"}. Prints {"ok": true}."""
+import json
+import re
+import sys
+
+_REMOTE_REF_RE = re.compile(
+    r'(?:src|href|action)\s*=\s*["\']?\s*(https?://[^\s"\'>\)]{1,200})',
+    re.IGNORECASE,
+)
+_REMOTE_CSS_URL_RE = re.compile(
+    r'url\s*\(\s*["\']?\s*(https?://[^\s"\'>\)]{1,200})',
+    re.IGNORECASE,
+)
+
+
+def _find_remote_refs(source):
+    """Return up to 5 remote URLs found in HTML/CSS source."""
+    refs = []
+    for pattern in (_REMOTE_REF_RE, _REMOTE_CSS_URL_RE):
+        for m in pattern.finditer(source):
+            refs.append(m.group(1)[:120])
+            if len(refs) >= 5:
+                return refs
+    return refs
+
+
+def main():
+    args = json.loads(sys.argv[1]) if len(sys.argv) > 1 else {}
+    path, out, mode = args.get("path"), args.get("out"), args.get("mode", "html")
+    if not path or not out or mode not in ("html", "markdown"):
+        print(json.dumps({"error": "missing path/out or bad mode"}))
+        sys.exit(1)
+    try:
+        from weasyprint import HTML
+        from weasyprint.urls import default_url_fetcher
+    except ImportError:
+        print(json.dumps({"error": "weasyprint not installed"}))
+        sys.exit(1)
+
+    def no_remote_fetcher(url, *fargs, **kwargs):
+        if url.startswith("data:"):
+            return default_url_fetcher(url, *fargs, **kwargs)
+        raise ValueError(f"remote resources are disabled: {url[:120]}")
+
+    try:
+        with open(path, "r", encoding="utf-8", errors="replace") as fh:
+            source = fh.read()
+        if mode == "markdown":
+            try:
+                import markdown as md
+            except ImportError:
+                print(json.dumps({"error": "markdown not installed"}))
+                sys.exit(1)
+            body = md.markdown(source, extensions=["tables", "fenced_code"])
+            source = (
+                "<!doctype html><html><head><meta charset=\"utf-8\">"
+                "<style>body{font-family:sans-serif;max-width:46em;margin:2em auto;}"
+                "code,pre{background:#f4f4f4;}table,td,th{border:1px solid #999;border-collapse:collapse;padding:4px;}</style>"
+                f"</head><body>{body}</body></html>"
+            )
+        remote_refs = _find_remote_refs(source)
+        if remote_refs:
+            print(json.dumps({"error": f"remote resources are disabled: {', '.join(remote_refs)}"}))
+            sys.exit(1)
+        HTML(string=source, url_fetcher=no_remote_fetcher, base_url=None).write_pdf(out)
+        print(json.dumps({"ok": True}))
+    except SystemExit:
+        raise
+    except Exception as exc:  # noqa: BLE001
+        print(json.dumps({"error": str(exc)}))
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()
@@ -0,0 +1,55 @@
+"""Read or write PDF document metadata via pikepdf docinfo.
+Args: {"path": in, "mode": "get"} -> {"metadata": {...}}
+      {"path": in, "out": o, "mode": "set", "metadata": {"Title": "..", ...}} -> {"ok": true}
+Settable keys: Title, Author, Subject, Keywords, Creator, Producer."""
+import json
+import sys
+
+ALLOWED_KEYS = {"Title", "Author", "Subject", "Keywords", "Creator", "Producer"}
+
+
+def main():
+    args = json.loads(sys.argv[1]) if len(sys.argv) > 1 else {}
+    path, mode = args.get("path"), args.get("mode", "get")
+    if not path:
+        print(json.dumps({"error": "missing path"}))
+        sys.exit(1)
+    try:
+        import pikepdf
+    except ImportError:
+        print(json.dumps({"error": "pikepdf not installed"}))
+        sys.exit(1)
+    try:
+        if mode == "get":
+            with pikepdf.open(path) as pdf:
+                info = {}
+                if pdf.docinfo is not None:
+                    for k, v in pdf.docinfo.items():
+                        info[str(k).lstrip("/")] = str(v)
+                print(json.dumps({"metadata": info}))
+            return
+        out = args.get("out")
+        meta = args.get("metadata") or {}
+        if not out or not isinstance(meta, dict):
+            print(json.dumps({"error": "missing out/metadata for set"}))
+            sys.exit(1)
+        with pikepdf.open(path) as pdf:
+            for key, value in meta.items():
+                if key not in ALLOWED_KEYS:
+                    continue
+                if value is None or value == "":
+                    if f"/{key}" in pdf.docinfo:
+                        del pdf.docinfo[f"/{key}"]
+                else:
+                    pdf.docinfo[f"/{key}"] = str(value)[:500]
+            pdf.save(out)
+        print(json.dumps({"ok": True}))
+    except SystemExit:
+        raise
+    except Exception as exc:  # noqa: BLE001
+        print(json.dumps({"error": str(exc)}))
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()
@@ -0,0 +1,64 @@
+"""True redaction: remove every occurrence of the given terms, then VERIFY
+none remain extractable. Args: {"path": in, "out": out, "terms": [".."],
+"caseSensitive": false}. Prints {"found": N, "verified": true}.
+
+Case-sensitivity note (PyMuPDF 1.27.2): fitz.Page.search_for is ALWAYS
+case-insensitive. When caseSensitive=true is requested, the search phase
+still finds all case variants (over-redaction, which is the safe direction
+for a legal redaction tool). The verification pass then enforces exact-case
+matching, so a caseSensitive=true request only reports leakage when the
+exact-case term survives. This is the sanctioned fallback documented in
+the wave-2 plan."""
+import json
+import sys
+
+
+def main():
+    args = json.loads(sys.argv[1]) if len(sys.argv) > 1 else {}
+    path, out, terms = args.get("path"), args.get("out"), args.get("terms") or []
+    case_sensitive = bool(args.get("caseSensitive", False))
+    if not path or not out or not isinstance(terms, list) or not terms:
+        print(json.dumps({"error": "missing path/out/terms"}))
+        sys.exit(1)
+    try:
+        import fitz
+    except ImportError:
+        print(json.dumps({"error": "PyMuPDF not installed"}))
+        sys.exit(1)
+    try:
+        doc = fitz.open(path)
+        flags = fitz.TEXT_DEHYPHENATE
+        found = 0
+        for page in doc:
+            for term in terms:
+                quads = page.search_for(term, quads=True, flags=flags) if case_sensitive else page.search_for(term, quads=True)
+                for quad in quads:
+                    page.add_redact_annot(quad, fill=(0, 0, 0))
+                    found += 1
+            page.apply_redactions()
+        doc.save(out, garbage=4, deflate=True)
+        doc.close()
+        # Verification pass: reopen and prove no term is extractable anymore.
+        check = fitz.open(out)
+        leaked = []
+        for page in check:
+            text = page.get_text()
+            haystack = text if case_sensitive else text.lower()
+            for term in terms:
+                needle = term if case_sensitive else term.lower()
+                if needle and needle in haystack:
+                    leaked.append(term)
+        check.close()
+        if leaked:
+            print(json.dumps({"error": f"verification failed: terms still extractable: {sorted(set(leaked))}"}))
+            sys.exit(1)
+        print(json.dumps({"found": found, "verified": True}))
+    except SystemExit:
+        raise
+    except Exception as exc:  # noqa: BLE001
+        print(json.dumps({"error": str(exc)}))
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()
@@ -0,0 +1,31 @@
+"""Extract plain text. Args: {"path": in, "out": out-txt-path}. Prints {"chars": N}."""
+import json
+import sys
+
+
+def main():
+    args = json.loads(sys.argv[1]) if len(sys.argv) > 1 else {}
+    path, out = args.get("path"), args.get("out")
+    if not path or not out:
+        print(json.dumps({"error": "missing path/out"}))
+        sys.exit(1)
+    try:
+        import fitz
+    except ImportError:
+        print(json.dumps({"error": "PyMuPDF not installed"}))
+        sys.exit(1)
+    try:
+        doc = fitz.open(path)
+        parts = [page.get_text() for page in doc]
+        doc.close()
+        text = "\n".join(parts)
+        with open(out, "w", encoding="utf-8") as fh:
+            fh.write(text)
+        print(json.dumps({"chars": len(text)}))
+    except Exception as exc:  # noqa: BLE001
+        print(json.dumps({"error": str(exc)}))
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()
@@ -0,0 +1,30 @@
+"""PDF to DOCX via pdf2docx (text PDFs; scanned PDFs produce poor output by
+design: documented fidelity caveat). Args: {"path": in, "out": out}.
+Prints {"ok": true}."""
+import json
+import sys
+
+
+def main():
+    args = json.loads(sys.argv[1]) if len(sys.argv) > 1 else {}
+    path, out = args.get("path"), args.get("out")
+    if not path or not out:
+        print(json.dumps({"error": "missing path/out"}))
+        sys.exit(1)
+    try:
+        from pdf2docx import Converter
+    except ImportError:
+        print(json.dumps({"error": "pdf2docx not installed"}))
+        sys.exit(1)
+    try:
+        cv = Converter(path)
+        cv.convert(out)
+        cv.close()
+        print(json.dumps({"ok": True}))
+    except Exception as exc:  # noqa: BLE001
+        print(json.dumps({"error": str(exc)}))
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()