#!/usr/bin/env python3 # Author: Tom Sapletta ยท https://tom.sapletta.com # Part of the ifURI solution. """Small URI document previewer for invoice review panels. The connector returns stable preview cards and can materialize cached thumbnails: PDF first pages through ``pdftoppm`` and text-like files as SVG text previews. Heavy OCR or image analysis stays in ``ocr://``; binary access stays in ``fs://``. """ from __future__ import annotations import base64 import hashlib import html import mimetypes import os import shutil import subprocess from pathlib import Path from typing import Any import urirun CONNECTOR_ID = "document-previewer" conn = urirun.connector(CONNECTOR_ID, scheme="preview") IMAGE_EXTS = {"png", "jpg", "jpeg", "webp", "gif", "bmp", "tif", "tiff"} PDF_THUMB_EXTS = {"pdf"} TEXT_THUMB_EXTS = {"txt", "csv", "json", "md", "log", "xml", "html", "htm"} THUMB_EXTS = PDF_THUMB_EXTS | TEXT_THUMB_EXTS DEFAULT_THUMB_WIDTH = 576 DEFAULT_THUMB_HEIGHT = 432 TYPE_RULES: dict[str, dict[str, str]] = { "pdf": {"kind": "pdf", "label": "PDF", "title": "PDF document", "color": "#b42318", "bg": "#fff1f0"}, "csv": {"kind": "sheet", "label": "CSV", "title": "CSV spreadsheet", "color": "#087443", "bg": "#ecfdf3"}, "xls": {"kind": "sheet", "label": "XLS", "title": "Spreadsheet", "color": "#087443", "bg": "#ecfdf3"}, "xlsx": {"kind": "sheet", "label": "XLSX", "title": "Spreadsheet", "color": "#087443", "bg": "#ecfdf3"}, "ods": {"kind": "sheet", "label": "ODS", "title": "Spreadsheet", "color": "#087443", "bg": "#ecfdf3"}, "doc": {"kind": "doc", "label": "DOC", "title": "Text document", "color": "#175cd3", "bg": "#eff8ff"}, "docx": {"kind": "doc", "label": "DOCX", "title": "Text document", "color": "#175cd3", "bg": "#eff8ff"}, "odt": {"kind": "doc", "label": "ODT", "title": "Text document", "color": "#175cd3", "bg": "#eff8ff"}, "rtf": {"kind": "doc", "label": "RTF", "title": "Rich text document", "color": "#175cd3", "bg": "#eff8ff"}, "txt": {"kind": "text", "label": "TXT", "title": "Text file", "color": "#344054", "bg": "#f2f4f7"}, "md": {"kind": "text", "label": "MD", "title": "Markdown file", "color": "#344054", "bg": "#f2f4f7"}, "json": {"kind": "code", "label": "JSON", "title": "JSON file", "color": "#7a2e0e", "bg": "#fff6ed"}, "xml": {"kind": "code", "label": "XML", "title": "XML file", "color": "#7a2e0e", "bg": "#fff6ed"}, "html": {"kind": "code", "label": "HTML", "title": "HTML file", "color": "#7a2e0e", "bg": "#fff6ed"}, "htm": {"kind": "code", "label": "HTML", "title": "HTML file", "color": "#7a2e0e", "bg": "#fff6ed"}, "zip": {"kind": "archive", "label": "ZIP", "title": "Archive", "color": "#475467", "bg": "#f2f4f7"}, "rar": {"kind": "archive", "label": "RAR", "title": "Archive", "color": "#475467", "bg": "#f2f4f7"}, "7z": {"kind": "archive", "label": "7Z", "title": "Archive", "color": "#475467", "bg": "#f2f4f7"}, "gz": {"kind": "archive", "label": "GZ", "title": "Archive", "color": "#475467", "bg": "#f2f4f7"}, "eml": {"kind": "mail", "label": "EML", "title": "Email message", "color": "#0f766e", "bg": "#f0fdfa"}, "msg": {"kind": "mail", "label": "MSG", "title": "Email message", "color": "#0f766e", "bg": "#f0fdfa"}, } DEFAULT_RULE = {"kind": "file", "label": "FILE", "title": "File", "color": "#344054", "bg": "#f8fafc"} def _root(path: str = "") -> Path: configured = path or os.environ.get("IFURI_PREVIEW_ROOT") or os.environ.get("INVOICE_ROOT") or os.environ.get("IFURI_FS_ROOT") or "." return Path(configured).expanduser().resolve() def _resolve(root: Path, rel: str) -> Path | None: rel_path = Path(rel.replace("\\", "/")) if rel_path.is_absolute() or ".." in rel_path.parts: return None target = (root / rel_path).resolve() return target if target == root or root in target.parents else None def file_ext(path: str) -> str: return Path(path).suffix.lower().lstrip(".") def preview_rule(ext: str) -> dict[str, str]: normalized = ext.lower().lstrip(".") if normalized in IMAGE_EXTS: return {"kind": "image", "label": normalized.upper() or "IMG", "title": "Image", "color": "#5925dc", "bg": "#f4f3ff"} return TYPE_RULES.get(normalized, {**DEFAULT_RULE, "label": (normalized.upper()[:5] if normalized else "FILE")}) def svg_icon(label: str, color: str, bg: str) -> str: safe_label = "".join(ch for ch in label.upper() if ch.isalnum())[:5] or "FILE" return ( '' f'' f'' f'' f'' f'{safe_label}' "" ) def data_uri(svg: str) -> str: encoded = base64.b64encode(svg.encode("utf-8")).decode("ascii") return "data:image/svg+xml;base64," + encoded def cache_dir(root: Path) -> Path: return root / ".state" / "previews" def cache_key(path: str, target: Path, width: int, height: int) -> str: try: stat = target.stat() raw = f"{path}\0{stat.st_mtime_ns}\0{stat.st_size}\0{width}x{height}" except OSError: raw = f"{path}\0missing\0{width}x{height}" return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:24] def text_lines(path: Path, max_lines: int = 8, max_chars: int = 30) -> list[str]: raw = path.read_bytes()[:4096] text = raw.decode("utf-8", "replace") lines: list[str] = [] for line in text.replace("\t", " ").splitlines(): collapsed = " ".join(line.split()) if not collapsed: continue lines.append(collapsed[:max_chars]) if len(lines) >= max_lines: break return lines or ["(empty)"] def text_thumbnail_svg(path: Path, rel: str, ext: str, width: int, height: int) -> str: rule = preview_rule(ext) title = html.escape(Path(rel).name[:42]) escaped_lines = [html.escape(line) for line in text_lines(path)] y = 48 rows = [] for line in escaped_lines: rows.append(f'{line}') y += 13 label = html.escape(rule["label"]) return ( f'' f'' f'' f'' f'{label}' f'{title}' f'{"".join(rows)}' "" ) def icon_thumbnail_svg(ext: str, width: int, height: int) -> str: rule = preview_rule(ext) return svg_icon(rule["label"], rule["color"], rule["bg"]).replace('width="192" height="144"', f'width="{width}" height="{height}"').replace("viewBox=\"0 0 192 144\"", f'viewBox="0 0 {width} {height}"') def write_text_thumbnail(target: Path, thumb_path: Path, rel: str, ext: str, width: int, height: int) -> dict[str, Any]: svg = text_thumbnail_svg(target, rel, ext, width, height) thumb_path.write_text(svg, encoding="utf-8") return {"ok": True, "thumbnail": str(thumb_path), "mime": "image/svg+xml", "generated": True, "backend": "svg-text"} def write_pdf_thumbnail(target: Path, thumb_path: Path, width: int, height: int, timeout: int) -> dict[str, Any]: pdftoppm = shutil.which("pdftoppm") if not pdftoppm: return {"ok": False, "error": "pdftoppm is not installed", "backend": "pdftoppm"} tmp_prefix = thumb_path.with_suffix("") tmp_file = tmp_prefix.with_suffix(".png") if tmp_file.exists(): tmp_file.unlink() cmd = [ pdftoppm, "-f", "1", "-l", "1", "-singlefile", "-png", "-scale-to", str(max(width, height)), str(target), str(tmp_prefix), ] proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, check=False) if proc.returncode != 0 or not tmp_file.is_file(): if tmp_file.exists(): tmp_file.unlink() return { "ok": False, "error": (proc.stderr or proc.stdout or f"pdftoppm exited {proc.returncode}").strip(), "backend": "pdftoppm", "command": cmd, } tmp_file.replace(thumb_path) return {"ok": True, "thumbnail": str(thumb_path), "mime": "image/png", "generated": True, "backend": "pdftoppm"} def thumbnail_for_path( root: str, path: str, width: int = DEFAULT_THUMB_WIDTH, height: int = DEFAULT_THUMB_HEIGHT, force: bool = False, timeout: int = 30, ) -> dict[str, Any]: base = _root(root) target = _resolve(base, path) if target is None: return {"ok": False, "connector": CONNECTOR_ID, "path": path, "error": "path escapes preview root"} if not target.is_file(): return {"ok": False, "connector": CONNECTOR_ID, "path": path, "error": "file does not exist"} ext = file_ext(path) if ext not in THUMB_EXTS: return {"ok": False, "connector": CONNECTOR_ID, "path": path, "error": f"thumbnail is not supported for .{ext}"} out_dir = cache_dir(base) out_dir.mkdir(parents=True, exist_ok=True) suffix = ".png" if ext in PDF_THUMB_EXTS else ".svg" key = cache_key(path, target, width, height) thumb_path = out_dir / f"{key}{suffix}" if thumb_path.exists() and not force: mime = "image/png" if suffix == ".png" else "image/svg+xml" return { "ok": True, "connector": CONNECTOR_ID, "root": str(base), "path": path, "thumbnail": str(thumb_path), "thumbnailRel": str(thumb_path.relative_to(base)), "mime": mime, "generated": False, "cached": True, "backend": "cache", } result = ( write_pdf_thumbnail(target, thumb_path, width, height, timeout) if ext in PDF_THUMB_EXTS else write_text_thumbnail(target, thumb_path, path, ext, width, height) ) if not result.get("ok"): fallback_path = out_dir / f"{key}.svg" fallback_path.write_text(icon_thumbnail_svg(ext, width, height), encoding="utf-8") result = { "ok": True, "thumbnail": str(fallback_path), "mime": "image/svg+xml", "generated": True, "backend": "fallback-svg", "warning": result.get("error", ""), } thumb_path = fallback_path return { "connector": CONNECTOR_ID, "root": str(base), "path": path, "thumbnailRel": str(thumb_path.relative_to(base)), "cached": False, **result, } def preview_card_for_path(root: str, path: str, include_svg: bool = True) -> dict[str, Any]: base = _root(root) target = _resolve(base, path) if target is None: return {"ok": False, "connector": CONNECTOR_ID, "path": path, "error": "path escapes preview root"} exists = target.is_file() stat = target.stat() if exists else None ext = file_ext(path) rule = preview_rule(ext) mime, _ = mimetypes.guess_type(str(target)) svg = svg_icon(rule["label"], rule["color"], rule["bg"]) if include_svg else "" return { "ok": True, "connector": CONNECTOR_ID, "root": str(base), "path": path, "name": Path(path).name, "exists": exists, "size": stat.st_size if stat else None, "mtime": int(stat.st_mtime) if stat else None, "ext": ext, "mime": mime or "application/octet-stream", "kind": rule["kind"], "label": rule["label"], "title": rule["title"], "color": rule["color"], "background": rule["bg"], "nativePreview": rule["kind"] == "image", "thumbnailSupported": ext in THUMB_EXTS or rule["kind"] == "image", "svg": svg, "dataUri": data_uri(svg) if svg else "", } @conn.handler("document/query/card", isolated=True, meta={"label": "Build a document preview card"}) def preview_card(root: str = "", path: str = "", include_svg: bool = True) -> dict[str, Any]: if not path: return {"ok": False, "connector": CONNECTOR_ID, "path": path, "error": "path is required"} return preview_card_for_path(root, path, include_svg=include_svg) @conn.handler("document/query/batch", isolated=True, meta={"label": "Build document preview cards"}) def preview_batch(root: str = "", paths: list[str] | None = None, include_svg: bool = False, max_files: int = 1000) -> dict[str, Any]: items = list(paths or [])[: max(0, int(max_files))] cards = [preview_card_for_path(root, str(path), include_svg=include_svg) for path in items] return { "ok": all(card.get("ok") for card in cards), "connector": CONNECTOR_ID, "root": str(_root(root)), "count": len(cards), "truncated": bool(paths and len(paths) > len(items)), "cards": cards, } @conn.handler("document/command/thumbnail", isolated=True, external=True, meta={"label": "Generate a cached document thumbnail"}) def preview_thumbnail( root: str = "", path: str = "", width: int = DEFAULT_THUMB_WIDTH, height: int = DEFAULT_THUMB_HEIGHT, force: bool = False, timeout: int = 30, ) -> dict[str, Any]: if not path: return {"ok": False, "connector": CONNECTOR_ID, "path": path, "error": "path is required"} return thumbnail_for_path(root, path, width=width, height=height, force=force, timeout=timeout) @conn.handler("document/command/thumbnails", isolated=True, external=True, meta={"label": "Generate cached document thumbnails"}) def preview_thumbnails( root: str = "", paths: list[str] | None = None, width: int = DEFAULT_THUMB_WIDTH, height: int = DEFAULT_THUMB_HEIGHT, force: bool = False, timeout: int = 30, max_files: int = 1000, ) -> dict[str, Any]: items = list(paths or [])[: max(0, int(max_files))] thumbnails = [ thumbnail_for_path(root, str(path), width=width, height=height, force=force, timeout=timeout) for path in items ] return { "ok": all(item.get("ok") for item in thumbnails), "connector": CONNECTOR_ID, "root": str(_root(root)), "count": len(thumbnails), "truncated": bool(paths and len(paths) > len(items)), "thumbnails": thumbnails, } def urirun_bindings() -> dict[str, Any]: return conn.bindings() def main(argv: list[str] | None = None) -> int: return conn.cli(argv) if __name__ == "__main__": raise SystemExit(main())