From b42925ec81a451fedc708f12fe4e9fd2ddc76ae6 Mon Sep 17 00:00:00 2001 From: AlanAAG Date: Tue, 26 May 2026 17:49:58 -0600 Subject: [PATCH 1/4] Upgrade create_pdf action with themes, visual layout, and unicode safety - Replace deprecated FPDF/HTMLMixin pattern with direct write_html on FPDF 2.8 - Add gradient banner header rendered from the first H1 heading - Add 5 colour themes: default, corporate, minimal, warm, forest - Add tag_styles for h1-h5 with colour hierarchy, code blocks with background fill, and page number footer - Add unicode sanitizer (_CHAR_MAP) to prevent latin-1 crashes on em dashes, smart quotes, currency symbols, and other common characters - Add optional subtitle and page_numbers input fields - Return pages and size_bytes in success payload for output verification - Add parallelizable=False to prevent concurrent file write corruption - Remove FPDF from requirement list (was silently skipped by executor heuristic) - Add tiered error handling: PermissionError, OSError, and generic Exception with actionable messages instead of a single bare except - Auto-create missing parent directories instead of crashing - Validate .pdf extension upfront --- app/data/action/create_pdf.py | 364 +++++++++++++++++++++++++++++----- 1 file changed, 318 insertions(+), 46 deletions(-) diff --git a/app/data/action/create_pdf.py b/app/data/action/create_pdf.py index baaf4454..5553ffd7 100644 --- a/app/data/action/create_pdf.py +++ b/app/data/action/create_pdf.py @@ -1,96 +1,368 @@ from agent_core import action - - + + @action( name="create_pdf", - description="Creates a PDF file at the specified path using the provided Markdown content. Converts Markdown to PDF using fpdf2, preserving headings, paragraphs, lists, bold and italic formatting. Use absolute paths.", + description=( + "Creates a visually polished PDF from Markdown content. " + "Supports headings (# to #####), paragraphs, bullet and numbered lists, " + "bold, italic, inline code, fenced code blocks, tables, strikethrough, " + "blockquotes, and horizontal rules. " + "The first # heading is rendered as a gradient banner header. " + "Available themes: default (indigo), corporate (blue), minimal (grey), " + "warm (amber), forest (green). " + "Use absolute paths only." + ), mode="CLI", action_sets=["document_processing"], + parallelizable=False, input_schema={ "file_path": { "type": "string", "example": "C:/Users/user/Documents/my_file.pdf", - "description": "Absolute path where the new PDF file will be created. Use full absolute paths (e.g., C:/Users/user/file.pdf or /home/user/file.pdf). Ensure the directory exists and is writable.", + "description": ( + "Absolute path where the PDF will be saved. " + "Parent directories are created automatically if they do not exist. " + "Must end with .pdf." + ), }, "content": { "type": "string", - "example": "# My Title\n\nThis is a paragraph with **bold** text and a bullet list:\n- Item 1\n- Item 2", - "description": "The Markdown-formatted content to be converted into a PDF file. Supports headings (#, ##, etc.), paragraphs, bullet lists (-, *), numbered lists (1., 2.), bold (**text**), and italics (*text*).", + "example": ( + "# My Report\n\n## Summary\n\nThis is **bold** and *italic*.\n\n" + "- Item 1\n- Item 2\n\n```python\nprint('hello')\n```" + ), + "description": ( + "Markdown-formatted content to convert into a PDF. " + "The first # heading becomes the banner title. " + "Supports tables (pipe syntax), fenced code blocks (```lang), " + "and ~~strikethrough~~." + ), + }, + "theme": { + "type": "string", + "example": "default", + "description": ( + "Visual colour theme. One of: default (indigo gradient), " + "corporate (blue), minimal (grey), warm (amber/orange), " + "forest (green). Defaults to 'default'." + ), + }, + "subtitle": { + "type": "string", + "example": "Confidential - Internal Use Only", + "description": ( + "Optional subtitle line shown below the title in the banner. " + "Leave empty or omit to hide." + ), + }, + "page_numbers": { + "type": "boolean", + "example": True, + "description": "Show 'Page N of M' in the footer. Defaults to true.", }, }, output_schema={ "status": { "type": "string", "example": "success", - "description": "'success' if the file was created, 'error' otherwise.", + "description": "'success' or 'error'.", }, "path": { "type": "string", "example": "C:/Users/user/Documents/my_file.pdf", - "description": "The path to the newly created PDF file.", + "description": "Absolute path of the created PDF.", + }, + "pages": { + "type": "integer", + "example": 3, + "description": "Number of pages in the generated PDF. Only present on success.", + }, + "size_bytes": { + "type": "integer", + "example": 48230, + "description": "File size in bytes. Only present on success.", }, "message": { "type": "string", "example": "Permission denied.", - "description": "Error message, present only if status is 'error'.", + "description": "Human-readable error detail. Only present on error.", }, }, - requirement=["markdown2", "FPDF", "fpdf2"], + requirement=["markdown2", "fpdf2"], test_payload={ "file_path": "C:/Users/user/Documents/my_file.pdf", - "content": "# My Title\n\nThis is a paragraph with **bold** text and a bullet list:\n- Item 1\n- Item 2", + "content": ( + "# My Title\n\nThis is a paragraph with **bold** text and a bullet list:\n" + "- Item 1\n- Item 2" + ), "simulated_mode": True, }, ) def create_pdf_file(input_data: dict) -> dict: + # ── Input extraction ────────────────────────────────────────────────── + simulated_mode = bool(input_data.get("simulated_mode", False)) + file_path = str(input_data.get("file_path", "")).strip() + content = str(input_data.get("content", "")).strip() + theme = str(input_data.get("theme", "default")).strip().lower() + subtitle = str(input_data.get("subtitle", "")).strip() + page_numbers = bool(input_data.get("page_numbers", True)) + + # ── Validation ──────────────────────────────────────────────────────── + if not file_path: + return {"status": "error", "path": "", "message": "The 'file_path' field is required."} + if not content: + return {"status": "error", "path": "", "message": "The 'content' field is required."} + if not file_path.lower().endswith(".pdf"): + return {"status": "error", "path": "", "message": "'file_path' must end with .pdf."} + + if simulated_mode: + return {"status": "success", "path": file_path} + + # ── Imports (executor pre-installs via requirement=, this is a fallback) ── + import os + import re import sys import subprocess import importlib - - def _ensure(pkg): + from html import unescape + + def _ensure(pkg, import_as=None): try: - importlib.import_module(pkg) + importlib.import_module(import_as or pkg) except ImportError: subprocess.check_call( - [sys.executable, "-m", "pip", "install", pkg, "--quiet"] + [sys.executable, "-m", "pip", "install", pkg, "--quiet"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, ) - - [_ensure(p) for p in ("markdown2", "fpdf2")] + + _ensure("markdown2") + _ensure("fpdf2", "fpdf") + import markdown2 - from fpdf import FPDF, HTMLMixin - - class PDF(FPDF, HTMLMixin): - pass - - simulated_mode = input_data.get("simulated_mode", False) - - file_path = str(input_data.get("file_path", "")).strip() - content = str(input_data.get("content", "")).strip() - - if not file_path: + from fpdf import FPDF + from fpdf.fonts import TextStyle, FontFace + from fpdf.pattern import LinearGradient + + # ── Themes ──────────────────────────────────────────────────────────── + # Keys: hbg=gradient stop colours, accent=link/highlight colour, + # h2/h3=heading colours, body=body text, cbg/cc=code bg/fg, + # rule=accent rule below banner, htxt=banner text + _THEMES = { + "default": { + "hbg": [(30, 58, 138), (79, 70, 229)], + "accent": (79, 70, 229), + "h2": (30, 58, 138), + "h3": (55, 65, 81), + "body": (31, 41, 55), + "cbg": (243, 244, 246), + "cc": (17, 24, 39), + "rule": (199, 210, 254), + "htxt": (255, 255, 255), + }, + "corporate": { + "hbg": [(0, 72, 148), (0, 120, 212)], + "accent": (0, 120, 212), + "h2": (0, 72, 148), + "h3": (60, 60, 100), + "body": (31, 41, 55), + "cbg": (240, 247, 255), + "cc": (0, 72, 148), + "rule": (173, 216, 230), + "htxt": (255, 255, 255), + }, + "minimal": { + "hbg": [(50, 50, 50), (90, 90, 90)], + "accent": (80, 80, 80), + "h2": (40, 40, 40), + "h3": (80, 80, 80), + "body": (40, 40, 40), + "cbg": (245, 245, 245), + "cc": (30, 30, 30), + "rule": (200, 200, 200), + "htxt": (255, 255, 255), + }, + "warm": { + "hbg": [(120, 53, 15), (217, 119, 6)], + "accent": (180, 83, 9), + "h2": (120, 53, 15), + "h3": (92, 72, 44), + "body": (41, 37, 36), + "cbg": (255, 247, 237), + "cc": (120, 53, 15), + "rule": (253, 186, 116), + "htxt": (255, 255, 255), + }, + "forest": { + "hbg": [(20, 83, 45), (34, 197, 94)], + "accent": (22, 163, 74), + "h2": (20, 83, 45), + "h3": (55, 65, 55), + "body": (31, 41, 31), + "cbg": (240, 253, 244), + "cc": (20, 83, 45), + "rule": (134, 239, 172), + "htxt": (255, 255, 255), + }, + } + t = _THEMES.get(theme, _THEMES["default"]) + + # ── Unicode sanitizer ───────────────────────────────────────────────── + # fpdf2's built-in fonts (Helvetica, Courier, Times) only cover latin-1 + # (characters 0-255). Any unicode character above that range causes a + # crash at render time. This map converts the most common offenders to + # safe ASCII equivalents before the HTML reaches fpdf2's parser. + # Characters with no mapping are replaced with '?'. + _CHAR_MAP = { + "\u2014": "--", "\u2013": "-", "\u2012": "-", + "\u2018": "'", "\u2019": "'", "\u201a": ",", + "\u201c": '"', "\u201d": '"', "\u201e": '"', + "\u2026": "...", "\u00a0": " ", + "\u2022": "*", "\u2010": "-", "\u2011": "-", "\u2015": "--", + "\u2122": "TM", "\u00ae": "(R)", "\u00a9": "(C)", + "\u20ac": "EUR", "\u00a3": "GBP", "\u00a5": "JPY", + "\u2192": "->", "\u2190": "<-", "\u2191": "^", "\u2193": "v", + "\u2713": "[x]", "\u2714": "[x]", "\u2717": "[ ]", + "\u2610": "[ ]", "\u2611": "[x]", + "\u00b0": "deg", "\u2265": ">=", "\u2264": "<=", + "\u00d7": "x", "\u00f7": "/", "\u00b1": "+/-", + "\u2248": "~=", "\u2260": "!=", "\u00b2": "^2", "\u00b3": "^3", + } + + def _sanitize(text): + decoded = unescape(text) + out = [] + for ch in decoded: + rep = _CHAR_MAP.get(ch) + if rep is not None: + out.append(rep) + elif ord(ch) > 255: + out.append("?") + else: + out.append(ch) + return "".join(out) + + # ── Build PDF ───────────────────────────────────────────────────────── + try: + # Convert markdown to HTML. + # smarty-pants is intentionally excluded: it converts -- and "quotes" + # to unicode HTML entities that get unescaped inside fpdf2's parser + # AFTER our sanitizer has already run, causing a crash. + html = markdown2.markdown( + content, + extras=["fenced-code-blocks", "tables", "strike", "footnotes"], + ) + html = _sanitize(html) + + # Extract the first H1 to use as the banner title, then remove it + # from the body so it is not rendered twice. + title_match = re.search(r"]*>(.*?)", html, re.IGNORECASE | re.DOTALL) + doc_title = ( + re.sub(r"<[^>]+>", "", title_match.group(1)).strip() + if title_match else "" + ) + html_body = html.replace(title_match.group(0), "", 1) if title_match else html + + # FPDF setup + pdf = FPDF() + pdf.set_auto_page_break(auto=True, margin=22) + pdf.set_margins(left=20, top=15, right=20) + if doc_title: + pdf.set_title(doc_title) + pdf.set_creator("CraftBot") + pdf.add_page() + + pw = pdf.w - pdf.l_margin - pdf.r_margin # usable page width + lm = pdf.l_margin + y0 = 8 # banner top y-position + HH = 50 if subtitle else 40 # banner height + + # ── Gradient banner ─────────────────────────────────────────────── + grad = LinearGradient(lm, y0, lm + pw, y0, colors=t["hbg"]) + with pdf.use_pattern(grad): + pdf.rect(lm, y0, pw, HH, style="F") + + if doc_title: + pdf.set_font("Helvetica", "B", 20) + pdf.set_text_color(*t["htxt"]) + title_y = y0 + (HH - 20) / 2 - (5 if subtitle else 0) + pdf.set_xy(lm + 8, title_y) + pdf.cell(pw - 16, 12, doc_title[:72], align="L") + + if subtitle: + pdf.set_font("Helvetica", "I", 9) + pdf.set_text_color(200, 210, 240) + pdf.set_xy(lm + 8, y0 + HH - 14) + pdf.cell(pw - 16, 8, _sanitize(subtitle)[:100], align="L") + + # Thin accent rule below banner + pdf.set_draw_color(*t["rule"]) + pdf.set_line_width(0.8) + pdf.line(lm, y0 + HH + 1, lm + pw, y0 + HH + 1) + pdf.set_y(y0 + HH + 7) + + # ── Heading and code styles ─────────────────────────────────────── + tag_styles = { + "h1": TextStyle(font_family="Helvetica", font_style="B", font_size_pt=20, color=t["h2"], t_margin=10, b_margin=3), + "h2": TextStyle(font_family="Helvetica", font_style="B", font_size_pt=16, color=t["h2"], t_margin=8, b_margin=2), + "h3": TextStyle(font_family="Helvetica", font_style="B", font_size_pt=13, color=t["h3"], t_margin=6, b_margin=2), + "h4": TextStyle(font_family="Helvetica", font_style="BI", font_size_pt=11, color=t["h3"], t_margin=4, b_margin=1), + "h5": TextStyle(font_family="Helvetica", font_style="I", font_size_pt=10, color=t["h3"], t_margin=3, b_margin=1), + "code": TextStyle(font_family="Courier", font_size_pt=9, color=t["cc"], fill_color=t["cbg"]), + "pre": TextStyle(font_family="Courier", font_size_pt=9, color=t["cc"], fill_color=t["cbg"]), + "a": FontFace(color=t["accent"]), + } + + pdf.set_text_color(*t["body"]) + pdf.set_font("Helvetica", size=11) + pdf.write_html( + html_body, + font_family="Helvetica", + tag_styles=tag_styles, + table_line_separators=True, + ul_bullet_char="*", + ) + + # ── Page number footer ──────────────────────────────────────────── + n_pages = len(pdf.pages) + if page_numbers: + for pg in range(1, n_pages + 1): + pdf.page = pg + pdf.set_y(-12) + pdf.set_font("Helvetica", "I", 8) + pdf.set_text_color(150, 150, 150) + pdf.cell(0, 5, f"Page {pg} of {n_pages}", align="C") + + # ── Write to disk ───────────────────────────────────────────────── + abs_path = os.path.abspath(file_path) + parent = os.path.dirname(abs_path) + if parent: + os.makedirs(parent, exist_ok=True) + + pdf.output(abs_path) + return { + "status": "success", + "path": abs_path, + "pages": n_pages, + "size_bytes": os.path.getsize(abs_path), + } + + except PermissionError as exc: return { "status": "error", "path": "", - "message": "The 'file_path' field is required.", + "message": f"Permission denied writing to '{file_path}': {exc}", } - if not content: + except OSError as exc: return { "status": "error", "path": "", - "message": "The 'content' field is required.", + "message": f"File system error: {exc}", } - - if simulated_mode: - # Return mock result for testing - return {"status": "success", "path": file_path} - - try: - html_content = markdown2.markdown(content) - pdf = PDF() - pdf.set_auto_page_break(auto=True, margin=15) - pdf.add_page() - pdf.write_html(html_content) - pdf.output(file_path) - return {"status": "success", "path": file_path} - except Exception as e: - return {"status": "error", "path": "", "message": str(e)} + except Exception as exc: + return { + "status": "error", + "path": "", + "message": f"PDF generation failed: {type(exc).__name__}: {exc}", + } \ No newline at end of file From 1a370e50f3a10e97e3c26a095c9ccb3831bdc196 Mon Sep 17 00:00:00 2001 From: AlanAAG Date: Tue, 26 May 2026 21:38:58 -0600 Subject: [PATCH 2/4] Upgrade read_pdf with pdfplumber primary engine, text/layout modes, and page_range --- app/data/action/read_pdf.py | 695 ++++++++++++++++++++---------------- 1 file changed, 380 insertions(+), 315 deletions(-) diff --git a/app/data/action/read_pdf.py b/app/data/action/read_pdf.py index 5f75373b..a83df236 100644 --- a/app/data/action/read_pdf.py +++ b/app/data/action/read_pdf.py @@ -1,265 +1,171 @@ from agent_core import action - - + + @action( name="read_pdf", - description="Securely reads a PDF with Docling and returns compact, layout-aware JSON including page sizes, bboxes, text, and form-field candidates. Implements a robust fallback using pypdfium2 and pdfminer.six if Docling cannot determine page sizes or extract text.", + description=( + "Reads a PDF and returns its content. " + "mode='text' (default): returns plain text and tables — use for summarising, " + "Q&A, and content extraction. Fast, minimal tokens. " + "mode='layout': returns per-word bounding boxes (BOTTOMLEFT origin) — use when " + "edit_pdf or form-filling needs spatial coordinates. " + "page_range limits which pages are read (e.g. '1', '1-3', '2,4'). " + "Digital PDFs use pdfplumber. Scanned/image PDFs fall back to Docling automatically." + ), mode="CLI", action_sets=["document_processing"], platforms=["windows", "linux", "darwin"], input_schema={ "file_path": { "type": "string", - "example": "C:/path/to/form.pdf", - "description": "Local path to the PDF to read.", - } + "example": "C:/path/to/document.pdf", + "description": "Absolute path to the PDF file to read.", + }, + "mode": { + "type": "string", + "example": "text", + "description": ( + "Output mode. 'text' (default): plain text + tables, minimal tokens. " + "'layout': per-word bbox coordinates for spatial tasks like edit_pdf or form-filling." + ), + }, + "page_range": { + "type": "string", + "example": "1-3", + "description": ( + "Pages to read. Omit to read all pages. " + "Formats: '1' (single), '1-3' (range), '1,3,5' (list)." + ), + }, }, output_schema={ - "status": {"type": "string", "example": "success"}, + "status": { + "type": "string", + "example": "success", + "description": "'success' or 'error'.", + }, "content": { "type": "object", - "description": "Layout-aware PDF extraction output.", + "description": ( + "Extraction result. Always contains document_metadata and pages. " + "text mode adds 'text' (string) and 'tables' (list, if any). " + "layout mode adds 'elements' (list of words with bbox_abs, bbox_norm, " + "is_form_field_candidate — same shape as v1 for backward compatibility)." + ), "example": { "document_metadata": { - "file_name": "sample.pdf", + "file_name": "invoice.pdf", "mimetype": "application/pdf", - "docling_version": "1.7.0", + "page_count": 2, + "engine": "pdfplumber", }, - "pages": [{"page_number": 1, "width": 595.44, "height": 841.68}], - "elements": [ - { - "page_number": 1, - "element_type": "text", - "text": "Name:", - "bbox_abs": { - "x0": 10, - "y0": 20, - "x1": 100, - "y1": 40, - "coord_origin": "BOTTOMLEFT", - }, - "bbox_norm": {"x0": 0.05, "y0": 0.02, "x1": 0.2, "y1": 0.05}, - "is_form_field_candidate": False, - } + "pages": [{"page_number": 1, "width": 595.28, "height": 841.89}], + "text": "Invoice #1042\nBill To: John Smith", + "tables": [ + [["Description", "Amount"], ["Web Dev", "$1,500.00"]] ], }, }, "message": { "type": "string", - "example": "File not found.", - "description": "Only set if status = error.", + "example": "File does not exist.", + "description": "Human-readable error detail. Only present on error.", }, }, - requirement=[ - "Any", - "DocumentConverter", - "pypdfium2", - "extract_text", - "docling", - "pdfminer", - ], - test_payload={"file_path": "C:/path/to/form.pdf", "simulated_mode": True}, + requirement=["pdfplumber", "pypdfium2", "docling", "pdfminer.six"], + test_payload={ + "file_path": "C:/path/to/form.pdf", + "simulated_mode": True, + }, ) def read_pdf_file(input_data: dict) -> dict: - #!/usr/bin/env python3 - import sys import os import re - import importlib + import sys import subprocess - - simulated_mode = input_data.get("simulated_mode", False) - - if simulated_mode: - # Return mock result for testing - return { - "status": "success", - "message": "", - "content": { - "document_metadata": { - "file_name": "test.pdf", - "mimetype": "application/pdf", - "docling_version": "1.7.0", - }, - "pages": [{"page_number": 1, "width": 595.44, "height": 841.68}], - "elements": [ - { - "page_number": 1, - "element_type": "text", - "text": "Test PDF content", - "bbox_abs": { - "x0": 10, - "y0": 20, - "x1": 100, - "y1": 40, - "coord_origin": "BOTTOMLEFT", - }, - "bbox_norm": {"x0": 0.05, "y0": 0.02, "x1": 0.2, "y1": 0.05}, - "is_form_field_candidate": False, - } - ], - }, - } - - # ------------------- - # Safe dependency install - # ------------------- - def _ensure(pkg: str) -> None: - try: - importlib.import_module(pkg) - except ImportError: - subprocess.check_call( - [sys.executable, "-m", "pip", "install", pkg, "--quiet"] - ) - - for _pkg in ("docling", "pypdfium2", "pdfminer.six", "Pillow"): - _ensure(_pkg) - - from docling.document_converter import DocumentConverter - import pypdfium2 - from pdfminer.high_level import extract_text - - SAFE_EXTS = {".pdf"} - MAX_FILE_SIZE_MB = 50 - _FIELD_RE = re.compile(r"(?:_{4,}|\.{4,}|—{3,}|–{3,})") - - # Helper functions + import importlib + + # ── Helpers ─────────────────────────────────────────────────────────── def _json(status, message="", content=None): return {"status": status, "message": message, "content": content or ""} - + + _FIELD_RE = re.compile(r"(?:_{4,}|\.{4,}|—{3,}|–{3,})") + def _is_form_blank(text): - if not text: - return False - return bool(_FIELD_RE.search(text.strip())) - - def _to_safe_int(v): - if isinstance(v, int): - if v > 9223372036854775807 or v < -9223372036854775808: - return str(v) - return v - - def _deep_sanitize(o): - if isinstance(o, dict): - return {k: _deep_sanitize(_to_safe_int(v)) for k, v in o.items()} - if isinstance(o, list): - return [_deep_sanitize(_to_safe_int(v)) for v in o] - return _to_safe_int(o) - - # Page extraction with fallback - def _extract_pages(raw, src): - pages = raw.get("pages") if raw else [] + return bool(text and _FIELD_RE.search(text.strip())) + + def _parse_page_range(pr, total): + """ + Parse '1', '1-3', '2,4,6' into a sorted list of 1-based page numbers. + Returns None on invalid input so the caller can surface a clear error. + """ + if not pr: + return list(range(1, total + 1)) + pages = set() + try: + for part in str(pr).split(","): + part = part.strip() + if not part: + continue + if "-" in part: + s, e = part.split("-", 1) + start = max(1, int(s.strip())) + end = min(total, int(e.strip())) + if start > end: + # e.g. '5-2' — reversed range, treat as invalid + return None + pages.update(range(start, end + 1)) + else: + p = int(part.strip()) + if 1 <= p <= total: + pages.add(p) + except (ValueError, AttributeError): + return None + return sorted(pages) + + def _words_to_elements(words, page_num, pw, ph): + """ + Convert pdfplumber word list to v1-compatible element format. + pdfplumber uses TOPLEFT origin (top = distance from page top). + We convert to BOTTOMLEFT so edit_pdf coordinates stay consistent + with what v1 and docling always produced. + """ out = [] - - if isinstance(pages, list): - for p in pages: - size = p.get("size") or {} - out.append( - { - "page_number": p.get("page_no") or p.get("number"), - "width": size.get("width"), - "height": size.get("height"), - } - ) - elif isinstance(pages, dict): - for k in sorted( - pages.keys(), key=lambda x: int(x) if str(x).isdigit() else str(x) - ): - p = pages[k] - size = p.get("size") or {} - out.append( - { - "page_number": p.get("page_no") or p.get("number") or int(k), - "width": size.get("width"), - "height": size.get("height"), - } - ) - - needs_fallback = ( - any(p["width"] in (None, 0) or p["height"] in (None, 0) for p in out) - or not out - ) - if needs_fallback: - try: - pdf = pypdfium2.PdfDocument(src) - if not out: - out = [ - {"page_number": i + 1, "width": None, "height": None} - for i in range(len(pdf)) - ] - for idx, p in enumerate(out): - page = pdf.get_page(idx) - w, h = page.get_size() - if not p["width"]: - p["width"] = float(w) - if not p["height"]: - p["height"] = float(h) - except Exception: - out = [ - {"page_number": i + 1, "width": 612, "height": 792} - for i in range(len(out) or 1) - ] + for w in words: + x0, x1 = float(w["x0"]), float(w["x1"]) + # TOPLEFT → BOTTOMLEFT: flip y axis + y0_bl = round(ph - float(w["bottom"]), 2) + y1_bl = round(ph - float(w["top"]), 2) + abs_bbox = { + "x0": round(x0, 2), + "y0": y0_bl, + "x1": round(x1, 2), + "y1": y1_bl, + "coord_origin": "BOTTOMLEFT", + } + norm_bbox = { + "x0": round(max(0.0, min(1.0, x0 / pw)), 4), + "y0": round(max(0.0, min(1.0, y0_bl / ph)), 4), + "x1": round(max(0.0, min(1.0, x1 / pw)), 4), + "y1": round(max(0.0, min(1.0, y1_bl / ph)), 4), + } + out.append({ + "page_number": page_num, + "element_type": "text", + "text": w["text"], + "bbox_abs": abs_bbox, + "bbox_norm": norm_bbox, + "is_form_field_candidate": _is_form_blank(w["text"]), + }) return out - - # Map page dimensions - def _page_dims_map(pages): - out = {} - for p in pages: - pn = p.get("page_number") - if isinstance(pn, int) and p.get("width") and p.get("height"): - out[pn] = {"w": float(p["width"]), "h": float(p["height"])} - return out - - # Bbox helpers - def _bbox_abs_from_docling(bbox): - return { - "x0": float(bbox.get("l", 0)), - "y0": float(bbox.get("b", 0)), - "x1": float(bbox.get("r", 0)), - "y1": float(bbox.get("t", 0)), - "coord_origin": str(bbox.get("coord_origin") or "BOTTOMLEFT"), - } - - def _norm_bbox(absb, w, h): - return { - "x0": max(0, min(1, absb["x0"] / w)), - "y0": max(0, min(1, absb["y0"] / h)), - "x1": max(0, min(1, absb["x1"] / w)), - "y1": max(0, min(1, absb["y1"] / h)), - } - - # Extract elements - def _extract_elements(raw, dims, src): + + def _docling_to_elements(raw, page_dims): + """ + Convert docling export_to_dict() output to v1-compatible element list. + Preserves the exact parsing logic from v1 for the fallback path. + """ out = [] texts = raw.get("texts") if raw else [] - - if not texts: - try: - full_text = extract_text(src) - for i, line in enumerate(full_text.splitlines()): - page_no = 1 - w, h = dims[page_no]["w"], dims[page_no]["h"] - abs_bbox = { - "x0": 0, - "y0": i * 10, - "x1": w, - "y1": (i + 1) * 10, - "coord_origin": "BOTTOMLEFT", - } - norm = _norm_bbox(abs_bbox, w, h) - out.append( - { - "page_number": page_no, - "element_type": "text", - "text": line.strip(), - "bbox_abs": abs_bbox, - "bbox_norm": norm, - "is_form_field_candidate": _is_form_blank(line), - } - ) - except Exception: - pass - return out - for t in texts: text_val = t.get("text") or t.get("orig") label = t.get("label") or t.get("type") or "text" @@ -269,100 +175,259 @@ def _extract_elements(raw, dims, src): p0 = prov[0] page_no = p0.get("page_no") bbox = p0.get("bbox") - if ( - page_no is None - or not isinstance(bbox, dict) - or int(page_no) not in dims - ): + if page_no is None or not isinstance(bbox, dict): continue - d = dims[int(page_no)] - abs_bbox = _bbox_abs_from_docling(bbox) - norm = _norm_bbox(abs_bbox, d["w"], d["h"]) - out.append( - { - "page_number": int(page_no), - "element_type": label, - "text": text_val, - "bbox_abs": abs_bbox, - "bbox_norm": norm, - "is_form_field_candidate": _is_form_blank(text_val), - } - ) + pn = int(page_no) + if pn not in page_dims: + continue + d = page_dims[pn] + pw, ph = d["w"], d["h"] + abs_bbox = { + "x0": float(bbox.get("l", 0)), + "y0": float(bbox.get("b", 0)), + "x1": float(bbox.get("r", 0)), + "y1": float(bbox.get("t", 0)), + "coord_origin": str(bbox.get("coord_origin") or "BOTTOMLEFT"), + } + norm_bbox = { + "x0": round(max(0.0, min(1.0, abs_bbox["x0"] / pw)), 4), + "y0": round(max(0.0, min(1.0, abs_bbox["y0"] / ph)), 4), + "x1": round(max(0.0, min(1.0, abs_bbox["x1"] / pw)), 4), + "y1": round(max(0.0, min(1.0, abs_bbox["y1"] / ph)), 4), + } + out.append({ + "page_number": pn, + "element_type": label, + "text": text_val, + "bbox_abs": abs_bbox, + "bbox_norm": norm_bbox, + "is_form_field_candidate": _is_form_blank(text_val), + }) return out - - simulated_mode = input_data.get("simulated_mode", False) - + + # ── Input extraction ────────────────────────────────────────────────── + simulated_mode = bool(input_data.get("simulated_mode", False)) + file_path = str(input_data.get("file_path", "")).strip() + mode = str(input_data.get("mode", "text")).strip().lower() + page_range = str(input_data.get("page_range", "")).strip() + + if mode not in ("text", "layout"): + mode = "text" + + # ── Simulated mode ──────────────────────────────────────────────────── if simulated_mode: - # Return mock result for testing - return { - "status": "success", - "content": { - "document_metadata": { - "file_name": os.path.basename( - input_data.get("file_path", "test.pdf") - ), - "mimetype": "application/pdf", - "docling_version": "1.7.0", - }, - "pages": [{"page_number": 1, "width": 595.44, "height": 841.68}], - "elements": [ - { - "page_number": 1, - "element_type": "text", - "text": "Test PDF content", - } - ], + base_content = { + "document_metadata": { + "file_name": os.path.basename(file_path) if file_path else "test.pdf", + "mimetype": "application/pdf", + "page_count": 1, + "engine": "simulated", }, - "message": "", + "pages": [{"page_number": 1, "width": 595.28, "height": 841.89}], } - - # Main execution - try: - src = str(input_data.get("file_path", "")).strip() - if not src: - return _json("error", "'file_path' is required.") - if ".." in src.replace("\\", "/"): - return _json("error", "Invalid file path.") - if not os.path.isfile(src): - return _json("error", "File does not exist.") - if not os.access(src, os.R_OK): - return _json("error", "File is not readable.") - ext = os.path.splitext(src)[1].lower() - if ext not in SAFE_EXTS: - return _json("error", f"Unsupported file type '{ext}'. Only PDF allowed.") - size_mb = os.path.getsize(src) / (1024 * 1024) - if size_mb > MAX_FILE_SIZE_MB: - return _json( - "error", - f"File too large ({size_mb:.1f} MB). Max {MAX_FILE_SIZE_MB} MB.", - ) - - raw = None + if mode == "layout": + base_content["elements"] = [{ + "page_number": 1, + "element_type": "text", + "text": "Test PDF content", + "bbox_abs": {"x0": 10.0, "y0": 20.0, "x1": 100.0, "y1": 40.0, "coord_origin": "BOTTOMLEFT"}, + "bbox_norm": {"x0": 0.05, "y0": 0.02, "x1": 0.2, "y1": 0.05}, + "is_form_field_candidate": False, + }] + else: + base_content["text"] = "Test PDF content" + return _json("success", "", base_content) + + # ── Dependency bootstrap (executor pre-installs via requirement=) ───── + def _ensure(pkg, import_as=None): try: - conv = DocumentConverter() - result = conv.convert(src) - if result.status == "success": - raw = result.document.export_to_dict() - except Exception: - pass - - pages = _extract_pages(raw, src) - dims = _page_dims_map(pages) - elements = _extract_elements(raw, dims, src) - - origin = raw.get("origin") if raw else {} + importlib.import_module(import_as or pkg) + except ImportError: + try: + subprocess.check_call( + [sys.executable, "-m", "pip", "install", pkg, "--quiet"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + except Exception: + pass # executor pre-installs via requirement=; failure here is non-fatal + + _ensure("pdfplumber") + _ensure("pypdfium2") + _ensure("docling") + + import pdfplumber + import pypdfium2 + + # ── Validation ──────────────────────────────────────────────────────── + if not file_path: + return _json("error", "'file_path' is required.") + if ".." in file_path.replace("\\", "/"): + return _json("error", "Invalid file path.") + if not os.path.isfile(file_path): + return _json("error", "File does not exist.") + if not os.access(file_path, os.R_OK): + return _json("error", "File is not readable.") + if not file_path.lower().endswith(".pdf"): + return _json("error", "Only .pdf files are supported.") + size_mb = os.path.getsize(file_path) / (1024 * 1024) + if size_mb > 100: + return _json( + "error", + f"File too large ({size_mb:.1f} MB). Max 100 MB. " + "If the PDF is a scanned document, consider splitting it into smaller " + "sections first using a tool like qpdf, then read each part separately " + "using the page_range parameter.", + ) + + # ── Primary extraction: pdfplumber ──────────────────────────────────── + try: + pages_out = [] + text_parts = [] + all_elements = [] + all_tables = [] + scanned_page_nums = [] # pages where pdfplumber found no text + + with pdfplumber.open(file_path) as doc: + total_pages = len(doc.pages) + target_pages = _parse_page_range(page_range, total_pages) + + if target_pages is None: + return _json("error", f"Invalid page_range format: '{page_range}'. Use '1', '1-3', or '2,4,6'.") + + for i, page in enumerate(doc.pages): + pn = i + 1 + if pn not in target_pages: + continue + + pw = page.width + ph = page.height + pages_out.append({ + "page_number": pn, + "width": round(pw, 2), + "height": round(ph, 2), + }) + + page_text = page.extract_text() or "" + + if page_text.strip(): + # Digital page — pdfplumber can handle it + if mode == "text": + text_parts.append(page_text) + tables = page.extract_tables() + if tables: + all_tables.extend(tables) + else: + # layout mode: word-level bbox + words = page.extract_words() + all_elements.extend(_words_to_elements(words, pn, pw, ph)) + else: + # No extractable text on this page — could be blank or scanned. + # We record it but only trigger the docling fallback if EVERY + # target page is empty. A single blank page in a digital PDF + # should not cause a full docling run. + scanned_page_nums.append(pn) + + engine = "pdfplumber" + engine_warning = "" + + # ── Fallback: docling for scanned pages ─────────────────────────── + # Only triggered when ALL target pages have no extractable text, + # which reliably signals a scanned or image-only PDF. + # A digital PDF with occasional blank pages will have text_parts + # populated and will NOT reach this block. + all_text_empty = not text_parts and not all_elements + if scanned_page_nums and all_text_empty: + try: + from docling.document_converter import DocumentConverter + from docling.datamodel.base_models import ConversionStatus + + conv = DocumentConverter() + result = conv.convert(file_path) + + if result.status in (ConversionStatus.SUCCESS, ConversionStatus.PARTIAL_SUCCESS): + engine = "docling" + + if mode == "text": + # export_to_markdown gives clean, LLM-ready text + fallback_text = result.document.export_to_markdown() or "" + if fallback_text.strip(): + text_parts.append(fallback_text) + else: + # layout mode: use docling's bbox data + raw = result.document.export_to_dict() + + # Rebuild page dims map from the pages we extracted + page_dims = { + p["page_number"]: {"w": p["width"], "h": p["height"]} + for p in pages_out + } + + # If pages_out is empty (fully scanned, pdfplumber got nothing) + # pull page dimensions from pypdfium2 + if not pages_out: + pdf2 = pypdfium2.PdfDocument(file_path) + target_pages_set = set( + _parse_page_range(page_range, len(pdf2)) or range(1, len(pdf2) + 1) + ) + for idx in range(len(pdf2)): + pn = idx + 1 + if pn not in target_pages_set: + continue + pg = pdf2.get_page(idx) + w, h = pg.get_size() + pages_out.append({ + "page_number": pn, + "width": round(float(w), 2), + "height": round(float(h), 2), + }) + page_dims[pn] = {"w": float(w), "h": float(h)} + + docling_elements = _docling_to_elements(raw, page_dims) + # Filter to target pages only — use the set already computed + # at extraction time, which holds original 1-based page numbers. + # Do NOT re-parse against len(pages_out): that would be the + # count of target pages, not total_pages, and would clip the + # range for any page_range that doesn't start at 1. + target_set = set(target_pages) + all_elements.extend( + e for e in docling_elements + if e["page_number"] in target_set + ) + else: + engine_warning = "Scanned pages detected but OCR extraction returned no content." + + except Exception as exc: + # docling unavailable or failed — surface what pdfplumber got + # (empty for scanned PDFs) and warn via metadata. + engine_warning = f"Scanned pages detected but OCR fallback failed: {type(exc).__name__}." + + # ── Build output ────────────────────────────────────────────────── meta = { - "file_name": origin.get("filename") or os.path.basename(src), - "mimetype": origin.get("mimetype") or "application/pdf", - "docling_version": raw.get("version") if raw else None, - } - - payload = { - "document_metadata": _deep_sanitize(meta), - "pages": _deep_sanitize(pages), - "elements": _deep_sanitize(elements), + "file_name": os.path.basename(file_path), + "mimetype": "application/pdf", + "page_count": total_pages, + "engine": engine, } - - return _json("success", "PDF extracted successfully.", payload) - except Exception as e: - return _json("error", str(e)) + if engine_warning: + meta["engine_warning"] = engine_warning + + if mode == "text": + content = { + "document_metadata": meta, + "pages": pages_out, + "text": "\n\n".join(text_parts), + } + if all_tables: + content["tables"] = all_tables + else: + content = { + "document_metadata": meta, + "pages": pages_out, + "elements": all_elements, + } + + return _json("success", "", content) + + except Exception as exc: + return _json("error", f"{type(exc).__name__}: {exc}") \ No newline at end of file From faca9ae00f83aa96e707648db8d97c931756ad9a Mon Sep 17 00:00:00 2001 From: AlanAAG Date: Tue, 26 May 2026 23:35:52 -0600 Subject: [PATCH 3/4] Add edit_pdf action and update requirements with PDF processing deps --- app/data/action/edit_pdf.py | 581 ++++++++++++++++++++++++++++++++++++ requirements.txt | 5 + 2 files changed, 586 insertions(+) create mode 100644 app/data/action/edit_pdf.py diff --git a/app/data/action/edit_pdf.py b/app/data/action/edit_pdf.py new file mode 100644 index 00000000..b2edfcfd --- /dev/null +++ b/app/data/action/edit_pdf.py @@ -0,0 +1,581 @@ +from agent_core import action + +@action( + name="edit_pdf", + description=( + "Edits an existing PDF by applying a list of operations and saves the result " + "to a new file. Accepts two input types per operation: " + "coord-based (x0/y0/x1/y1 in BOTTOMLEFT origin — direct from read_pdf layout mode) " + "and pattern-based (text string searched internally — no read_pdf call needed). " + "Operations: add_text, redact, highlight, underline, strikeout (coord or pattern), " + "replace_text (find + font-matched reinsert), add_text_near (fill after a label), " + "watermark, rotate_page, fill_field (AcroForm). " + "For tasks that require text reflow (rephrasing paragraphs, inserting new sections, " + "reformatting layout): use create_pdf to rebuild the document with changes applied — " + "the user receives the same output path with a clean result. " + "Use absolute paths only." + ), + mode="CLI", + action_sets=["document_processing"], + parallelizable=False, + platforms=["windows", "linux", "darwin"], + input_schema={ + "file_path": { + "type": "string", + "example": "C:/path/to/document.pdf", + "description": "Absolute path to the source PDF to edit.", + }, + "output_path": { + "type": "string", + "example": "C:/path/to/document_edited.pdf", + "description": ( + "Absolute path for the output PDF. " + "Parent directories are created automatically. " + "Must end with .pdf." + ), + }, + "operations": { + "type": "array", + "description": ( + "Ordered list of edit operations to apply. Each operation is an object " + "with a 'type' field plus type-specific fields. " + "Coord fields (x0/y0/x1/y1) use BOTTOMLEFT origin (direct from read_pdf layout mode). " + "Pattern fields search the PDF internally — no read_pdf call needed.\n" + "Types:\n" + " add_text: insert text. Fields: page(int), text(str), x(float), y(float, BOTTOMLEFT), " + "font_size(float,default 12), color(hex,default '#000000')\n" + " redact: true redaction — removes content from stream. Fields: page(int), " + "x0/y0/x1/y1(float, BOTTOMLEFT) OR pattern(str) + page('all' or int)\n" + " highlight: highlight annotation. Fields: page(int), " + "x0/y0/x1/y1(float, BOTTOMLEFT) OR pattern(str) + page, color(hex,default '#ffff00')\n" + " underline: underline annotation. Fields: page(int), x0/y0/x1/y1 OR pattern + page\n" + " strikeout: strikeout annotation. Fields: page(int), x0/y0/x1/y1 OR pattern + page\n" + " replace_text: find text and replace with font-matched new text. " + "Fields: pattern(str), replacement(str), page('all' or int)\n" + " add_text_near: find a label and insert value after it (form fill). " + "Fields: label(str), value(str), page(int), " + "offset_x(float,default 5), font_size(float,default 11), color(hex,default '#000000')\n" + " watermark: add diagonal text watermark to all pages. " + "Fields: text(str), font_size(float,default 52), color(hex,default '#bbbbbb'), opacity(0-1,default 0.25)\n" + " rotate_page: rotate a page. Fields: page(int), degrees(90/180/270)\n" + " fill_field: fill an AcroForm field. Fields: field_name(str), value(str)" + ), + "example": [ + {"type": "highlight", "pattern": "Invoice", "page": "all", "color": "#fef08a"}, + {"type": "add_text", "page": 1, "text": "PAID", "x": 200, "y": 400, + "font_size": 36, "color": "#16a34a"}, + {"type": "redact", "page": 1, "x0": 31.2, "y0": 791.3, "x1": 86.3, "y1": 807.3}, + {"type": "replace_text", "pattern": "Q3", "replacement": "Q4", "page": "all"}, + {"type": "watermark", "text": "DRAFT", "color": "#bbbbbb", "opacity": 0.2}, + ], + }, + }, + output_schema={ + "status": { + "type": "string", + "example": "success", + "description": "'success' or 'error'.", + }, + "output_path": { + "type": "string", + "example": "C:/path/to/document_edited.pdf", + "description": "Absolute path of the edited PDF.", + }, + "operations_applied": { + "type": "integer", + "example": 3, + "description": "Number of operations successfully applied.", + }, + "warnings": { + "type": "array", + "example": ["replace_text 'Q3': 0 matches found on page 2"], + "description": "Non-fatal issues (zero matches, font fallbacks, etc.).", + }, + "message": { + "type": "string", + "example": "File does not exist.", + "description": "Human-readable error detail. Only present on error.", + }, + }, + requirement=["pymupdf", "pypdf"], + test_payload={ + "file_path": "C:/path/to/document.pdf", + "output_path": "C:/path/to/document_edited.pdf", + "operations": [{"type": "watermark", "text": "DRAFT"}], + "simulated_mode": True, + }, +) +def edit_pdf_file(input_data: dict) -> dict: + import os + import sys + import re + import subprocess + import importlib + + # ── Helpers ─────────────────────────────────────────────────────────── + def _json(status, message="", output_path="", ops_applied=0, warnings=None): + result = { + "status": status, + "output_path": output_path, + "operations_applied": ops_applied, + "warnings": warnings or [], + } + if message: + result["message"] = message + return result + + def _ensure(pkg, import_as=None): + try: + importlib.import_module(import_as or pkg) + except ImportError: + try: + subprocess.check_call( + [sys.executable, "-m", "pip", "install", pkg, "--quiet"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + except Exception: + pass + + def _hex_to_rgb(hex_color): + """Convert '#rrggbb' or '#rgb' to (r,g,b) float tuple 0-1.""" + h = str(hex_color).lstrip("#") + if len(h) == 3: + h = "".join(c * 2 for c in h) + if len(h) != 6: + return (0.0, 0.0, 0.0) + return tuple(int(h[i:i+2], 16) / 255.0 for i in (0, 2, 4)) + + def _color_int_to_rgb(color_int): + """Convert PyMuPDF integer color to (r,g,b) float tuple.""" + r = ((color_int >> 16) & 0xFF) / 255.0 + g = ((color_int >> 8) & 0xFF) / 255.0 + b = ( color_int & 0xFF) / 255.0 + return (r, g, b) + + # Base14 fonts always available in PyMuPDF — no embedding needed + _BASE14 = { + "Helvetica": "helv", + "Helvetica-Bold": "hebo", + "Helvetica-BoldOblique": "hebi", + "Times-Roman": "tiro", + "Times-Bold": "tibo", + "Times-Italic": "tiit", + "Times-BoldItalic": "tibi", + "Courier": "cour", + "Courier-Bold": "cobo", + "Courier-Oblique": "coit", + "Courier-BoldOblique": "cobi", + } + + def _best_font(pdf_font_name): + """ + Map any PDF font name to the nearest valid PyMuPDF Base14 shortcode. + Returns (shortcode, was_fallback). + """ + if pdf_font_name in _BASE14: + return _BASE14[pdf_font_name], False + n = pdf_font_name.lower() + is_bold = any(x in n for x in ("bold", "demi", "black", "heavy")) + is_italic = any(x in n for x in ("italic", "oblique", "slant")) + if "times" in n or "roman" in n or "serif" in n: + if is_bold and is_italic: return "tibi", True + if is_bold: return "tibo", True + if is_italic: return "tiit", True + return "tiro", True + if "courier" in n or "mono" in n or "typewriter" in n: + if is_bold and is_italic: return "cobi", True + if is_bold: return "cobo", True + if is_italic: return "coit", True + return "cour", True + # Default: Helvetica family + if is_bold and is_italic: return "hebi", True + if is_bold: return "hebo", True + return "helv", True + + def _bl_to_rect(x0, y0, x1, y1, ph): + """ + Convert BOTTOMLEFT bbox (from read_pdf) to PyMuPDF TOPLEFT Rect. + PyMuPDF: y=0 at top, y=ph at bottom. + BOTTOMLEFT: y=0 at bottom, y=ph at top. + Conversion: y_mupdf = ph - y_bottomleft + """ + import fitz + return fitz.Rect(x0, ph - y1, x1, ph - y0) + + def _resolve_pages(page_spec, total_pages): + """ + Resolve page spec to list of 0-based indices. + 'all' → all pages. int → single page (1-based). 'all' default. + """ + if page_spec == "all" or page_spec is None or page_spec == "": + return list(range(total_pages)) + try: + p = int(page_spec) + if 1 <= p <= total_pages: + return [p - 1] + return [] + except (ValueError, TypeError): + return list(range(total_pages)) + + def _get_span_at_rect(page, target_rect): + """ + Find the text span whose bbox best overlaps target_rect. + Returns the span dict or None. + """ + import fitz + best_span = None + best_area = 0.0 + for block in page.get_text("dict")["blocks"]: + if block.get("type") != 0: + continue + for line in block.get("lines", []): + for span in line.get("spans", []): + sr = fitz.Rect(span["bbox"]) + overlap = sr & target_rect + if overlap.is_valid and overlap.get_area() > best_area: + best_area = overlap.get_area() + best_span = span + return best_span + + # ── Input extraction ────────────────────────────────────────────────── + simulated_mode = bool(input_data.get("simulated_mode", False)) + file_path = str(input_data.get("file_path", "")).strip() + output_path = str(input_data.get("output_path", "")).strip() + operations = input_data.get("operations", []) + + if not isinstance(operations, list): + operations = [] + + # ── Simulated mode ──────────────────────────────────────────────────── + if simulated_mode: + return _json("success", output_path=output_path, ops_applied=len(operations)) + + # ── Dependency bootstrap ────────────────────────────────────────────── + _ensure("pymupdf", "fitz") + _ensure("pypdf", "pypdf") + + import fitz + import pypdf + + # ── Validation ──────────────────────────────────────────────────────── + if not file_path: + return _json("error", "'file_path' is required.") + if not output_path: + return _json("error", "'output_path' is required.") + if not file_path.lower().endswith(".pdf"): + return _json("error", "Only .pdf files are supported.") + if not output_path.lower().endswith(".pdf"): + return _json("error", "'output_path' must end with .pdf.") + if ".." in file_path.replace("\\", "/"): + return _json("error", "Invalid file_path.") + if ".." in output_path.replace("\\", "/"): + return _json("error", "Invalid output_path.") + if not os.path.isfile(file_path): + return _json("error", "File does not exist.") + if not os.access(file_path, os.R_OK): + return _json("error", "File is not readable.") + size_mb = os.path.getsize(file_path) / (1024 * 1024) + if size_mb > 100: + return _json("error", f"File too large ({size_mb:.1f} MB). Max 100 MB.") + if not operations: + return _json("error", "'operations' list is required and must not be empty.") + + # Detect reflow operations — these require create_pdf routing + _REFLOW_OPS = {"rephrase_text", "insert_section", "insert_table", "reformat", "reflow"} + reflow_ops = [op.get("type") for op in operations if op.get("type") in _REFLOW_OPS] + if reflow_ops: + return _json( + "error", + f"Operation(s) {reflow_ops} require text reflow which PDF does not support. " + "Use create_pdf to rebuild the document with the desired changes applied. " + "Read the original with read_pdf (text mode), apply changes to the text content, " + "then pass the updated content to create_pdf at the same output_path.", + ) + + # ── Apply operations ────────────────────────────────────────────────── + try: + doc = fitz.open(file_path) + total_pgs = len(doc) + warnings = [] + ops_done = 0 + + for i, op in enumerate(operations): + op_type = str(op.get("type", "")).strip().lower() + op_tag = f"op[{i}] '{op_type}'" + + try: + + # ── add_text ───────────────────────────────────────────── + if op_type == "add_text": + page_idx = int(op.get("page", 1)) - 1 + if not (0 <= page_idx < total_pgs): + warnings.append(f"{op_tag}: page {op.get('page')} out of range.") + continue + page = doc[page_idx] + ph = page.rect.height + # x,y are BOTTOMLEFT — convert y to TOPLEFT + x = float(op.get("x", 0)) + y_bl = float(op.get("y", 0)) + y_tl = ph - y_bl + text = str(op.get("text", "")) + font_size = float(op.get("font_size", 12)) + color = _hex_to_rgb(op.get("color", "#000000")) + font_code, _ = _best_font(op.get("font", "Helvetica")) + page.insert_text( + fitz.Point(x, y_tl), text, + fontname=font_code, fontsize=font_size, + color=color, + ) + ops_done += 1 + + # ── redact (coord or pattern) ───────────────────────────── + elif op_type == "redact": + fill_color = _hex_to_rgb(op.get("fill_color", "#000000")) + if "pattern" in op: + pattern = str(op["pattern"]) + page_idxs = _resolve_pages(op.get("page", "all"), total_pgs) + hit_count = 0 + for pi in page_idxs: + page = doc[pi] + hits = page.search_for(pattern) + for h in hits: + page.add_redact_annot(h, fill=fill_color) + hit_count += 1 + for pi in page_idxs: + doc[pi].apply_redactions() + if hit_count == 0: + warnings.append(f"{op_tag}: pattern '{pattern}' not found.") + ops_done += 1 + else: + page_idx = int(op.get("page", 1)) - 1 + if not (0 <= page_idx < total_pgs): + warnings.append(f"{op_tag}: page out of range."); continue + page = doc[page_idx] + ph = page.rect.height + r = _bl_to_rect( + float(op["x0"]), float(op["y0"]), + float(op["x1"]), float(op["y1"]), ph + ) + page.add_redact_annot(r, fill=fill_color) + page.apply_redactions() + ops_done += 1 + + # ── highlight / underline / strikeout ───────────────────── + elif op_type in ("highlight", "underline", "strikeout"): + annot_fns = { + "highlight": "add_highlight_annot", + "underline": "add_underline_annot", + "strikeout": "add_strikeout_annot", + } + fn_name = annot_fns[op_type] + color = _hex_to_rgb(op.get("color", "#ffff00")) if op_type == "highlight" else None + + if "pattern" in op: + pattern = str(op["pattern"]) + page_idxs = _resolve_pages(op.get("page", "all"), total_pgs) + hit_count = 0 + for pi in page_idxs: + page = doc[pi] + hits = page.search_for(pattern) + for h in hits: + annot = getattr(page, fn_name)(h) + if color: + annot.set_colors(stroke=color) + annot.update() + hit_count += 1 + if hit_count == 0: + warnings.append(f"{op_tag}: pattern '{pattern}' not found.") + ops_done += 1 + else: + page_idx = int(op.get("page", 1)) - 1 + if not (0 <= page_idx < total_pgs): + warnings.append(f"{op_tag}: page out of range."); continue + page = doc[page_idx] + ph = page.rect.height + r = _bl_to_rect( + float(op["x0"]), float(op["y0"]), + float(op["x1"]), float(op["y1"]), ph + ) + annot = getattr(page, fn_name)(r) + if color: + annot.set_colors(stroke=color) + annot.update() + ops_done += 1 + + # ── replace_text (font-matched seamless replacement) ────── + elif op_type == "replace_text": + pattern = str(op.get("pattern", "")) + replacement = str(op.get("replacement", "")) + page_idxs = _resolve_pages(op.get("page", "all"), total_pgs) + if not pattern: + warnings.append(f"{op_tag}: 'pattern' is required."); continue + hit_count = 0 + for pi in page_idxs: + page = doc[pi] + hits = page.search_for(pattern) + for h in hits: + span = _get_span_at_rect(page, h) + if span: + font_name = span["font"] + font_size = span["size"] + color_rgb = _color_int_to_rgb(span["color"]) + font_code, was_fb = _best_font(font_name) + if was_fb: + warnings.append( + f"{op_tag}: font '{font_name}' not in Base14, " + f"using '{font_code}' as fallback." + ) + else: + font_code = "helv" + font_size = 11.0 + color_rgb = (0.0, 0.0, 0.0) + # Redact original text + page.add_redact_annot(h) + page.apply_redactions() + # Reinsert with same properties + insert_pt = fitz.Point(h.x0, h.y1 - 2) + page.insert_text( + insert_pt, replacement, + fontname=font_code, + fontsize=font_size, + color=color_rgb, + ) + hit_count += 1 + if hit_count == 0: + warnings.append(f"{op_tag}: pattern '{pattern}' not found.") + ops_done += 1 + + # ── add_text_near (form fill: insert value after label) ──── + elif op_type == "add_text_near": + label = str(op.get("label", "")) + value = str(op.get("value", "")) + page_idx = int(op.get("page", 1)) - 1 + offset_x = float(op.get("offset_x", 5)) + font_size = float(op.get("font_size", 11)) + color = _hex_to_rgb(op.get("color", "#000000")) + font_code, _ = _best_font(op.get("font", "Helvetica")) + if not label: + warnings.append(f"{op_tag}: 'label' is required."); continue + if not (0 <= page_idx < total_pgs): + warnings.append(f"{op_tag}: page out of range."); continue + page = doc[page_idx] + hits = page.search_for(label) + if not hits: + warnings.append(f"{op_tag}: label '{label}' not found on page {page_idx+1}.") + continue + label_rect = hits[0] + insert_pt = fitz.Point( + label_rect.x1 + offset_x, + label_rect.y1 - 2, # baseline approx + ) + page.insert_text( + insert_pt, value, + fontname=font_code, + fontsize=font_size, + color=color, + ) + ops_done += 1 + + # ── watermark ───────────────────────────────────────────── + elif op_type == "watermark": + text = str(op.get("text", "CONFIDENTIAL")) + font_size = float(op.get("font_size", 52)) + color = _hex_to_rgb(op.get("color", "#bbbbbb")) + # opacity in insert_textbox: blend into color lightness + opacity = float(op.get("opacity", 0.25)) + # Blend color toward white to simulate opacity + blended = tuple(c + (1.0 - c) * (1.0 - opacity) for c in color) + for page in doc: + pw, ph = page.rect.width, page.rect.height + rect = fitz.Rect(40, ph / 2 - 60, pw - 40, ph / 2 + 60) + page.insert_textbox( + rect, text, + fontsize=font_size, + color=blended, + align=fitz.TEXT_ALIGN_CENTER, + ) + ops_done += 1 + + # ── rotate_page ─────────────────────────────────────────── + elif op_type == "rotate_page": + page_idx = int(op.get("page", 1)) - 1 + degrees = int(op.get("degrees", 90)) + if not (0 <= page_idx < total_pgs): + warnings.append(f"{op_tag}: page out of range."); continue + if degrees not in (90, 180, 270, -90, -180, -270): + warnings.append(f"{op_tag}: degrees must be 90, 180, or 270."); continue + # Normalize to 0-360 + degrees = degrees % 360 + current = doc[page_idx].rotation + doc[page_idx].set_rotation((current + degrees) % 360) + ops_done += 1 + + # ── fill_field (AcroForm via pypdf) ─────────────────────── + elif op_type == "fill_field": + # Defer all fill_field ops to after PyMuPDF saves + # (pypdf needs to open the saved file) + # We flag these for post-processing below + pass # handled in post-processing step + + else: + warnings.append(f"{op_tag}: unknown operation type '{op_type}'.") + + except KeyError as e: + warnings.append(f"{op_tag}: missing required field {e}.") + except Exception as e: + warnings.append(f"{op_tag}: {type(e).__name__}: {e}.") + + # ── Write PyMuPDF output ────────────────────────────────────────── + abs_output = os.path.abspath(output_path) + parent = os.path.dirname(abs_output) + if parent: + os.makedirs(parent, exist_ok=True) + + doc.save(abs_output, garbage=4, deflate=True) + doc.close() + + # ── Post-process: AcroForm fill_field via pypdf ─────────────────── + acroform_ops = [op for op in operations if str(op.get("type", "")).lower() == "fill_field"] + if acroform_ops: + try: + reader = pypdf.PdfReader(abs_output) + writer = pypdf.PdfWriter() + writer.append(reader) + existing_fields = reader.get_fields() or {} + for op in acroform_ops: + op_tag = f"op[fill_field]" + field_name = str(op.get("field_name", "")) + value = str(op.get("value", "")) + if not field_name: + warnings.append(f"{op_tag}: 'field_name' is required.") + continue + if field_name not in existing_fields: + warnings.append( + f"{op_tag}: field '{field_name}' not found in AcroForm. " + f"Available fields: {list(existing_fields.keys())[:10]}." + ) + continue + for page_obj in writer.pages: + writer.update_page_form_field_values( + page_obj, {field_name: value} + ) + ops_done += 1 + with open(abs_output, "wb") as f: + writer.write(f) + except Exception as e: + warnings.append(f"AcroForm fill failed: {type(e).__name__}: {e}.") + + return _json( + "success", + output_path=abs_output, + ops_applied=ops_done, + warnings=warnings, + ) + + except PermissionError as exc: + return _json("error", f"Permission denied: {exc}") + except OSError as exc: + return _json("error", f"File system error: {exc}") + except Exception as exc: + return _json("error", f"{type(exc).__name__}: {exc}") diff --git a/requirements.txt b/requirements.txt index 31de103e..870d76be 100644 --- a/requirements.txt +++ b/requirements.txt @@ -47,4 +47,9 @@ playwright # WhatsApp Web browser automation opencv-python-headless # Video analysis keyframe extraction babel>=2.14.0 # Language list for onboarding boto3>=1.34.0 # AWS Bedrock provider (LLM/VLM/embedding via Converse + Titan) +pdfplumber # PDF text, table, and bbox extraction (read_pdf primary engine) +pypdfium2 # PDF page dimensions for scanned document fallback (read_pdf) +pdfminer.six # PDF text extraction fallback (read_pdf) +pymupdf # PDF editing engine — redaction, annotations, replace (edit_pdf) +pypdf # AcroForm field filling and PDF merging (edit_pdf) From 1628dafdbbb60a336ab54f960581a533a4c92e66 Mon Sep 17 00:00:00 2001 From: AlanAAG Date: Wed, 27 May 2026 00:34:38 -0600 Subject: [PATCH 4/4] Update create_pdf theme guidance and add theme_used to edit_pdf description --- app/data/action/create_pdf.py | 22 ++++++++++++++++++---- app/data/action/edit_pdf.py | 3 +++ 2 files changed, 21 insertions(+), 4 deletions(-) diff --git a/app/data/action/create_pdf.py b/app/data/action/create_pdf.py index 5553ffd7..1a08c9d8 100644 --- a/app/data/action/create_pdf.py +++ b/app/data/action/create_pdf.py @@ -43,9 +43,13 @@ "type": "string", "example": "default", "description": ( - "Visual colour theme. One of: default (indigo gradient), " - "corporate (blue), minimal (grey), warm (amber/orange), " - "forest (green). Defaults to 'default'." + "Visual colour theme. One of: " + "default (indigo) — general use; " + "corporate (blue) — business, finance, formal reports; " + "minimal (grey) — academic, technical, low-decoration; " + "warm (amber) — creative, personal, informal; " + "forest (green) — sustainability, nature, environmental. " + "Defaults to 'default'." ), }, "subtitle": { @@ -83,6 +87,14 @@ "example": 48230, "description": "File size in bytes. Only present on success.", }, + "theme_used": { + "type": "string", + "example": "corporate", + "description": ( + "The theme that was applied. Useful for downstream actions " + "(e.g. edit_pdf) that need to match colours to the document style." + ), + }, "message": { "type": "string", "example": "Permission denied.", @@ -207,7 +219,8 @@ def _ensure(pkg, import_as=None): }, } t = _THEMES.get(theme, _THEMES["default"]) - + theme = theme if theme in _THEMES else "default" # resolve fallback for theme_used + # ── Unicode sanitizer ───────────────────────────────────────────────── # fpdf2's built-in fonts (Helvetica, Courier, Times) only cover latin-1 # (characters 0-255). Any unicode character above that range causes a @@ -346,6 +359,7 @@ def _sanitize(text): "path": abs_path, "pages": n_pages, "size_bytes": os.path.getsize(abs_path), + "theme_used": theme, } except PermissionError as exc: diff --git a/app/data/action/edit_pdf.py b/app/data/action/edit_pdf.py index b2edfcfd..73473171 100644 --- a/app/data/action/edit_pdf.py +++ b/app/data/action/edit_pdf.py @@ -13,6 +13,9 @@ "For tasks that require text reflow (rephrasing paragraphs, inserting new sections, " "reformatting layout): use create_pdf to rebuild the document with changes applied — " "the user receives the same output path with a clean result. " + "When editing a PDF created by create_pdf, use the theme_used value from that call " + "to pick matching accent colours: default=#4f46e5, corporate=#0078d4, " + "minimal=#505050, warm=#b45309, forest=#16a34a. " "Use absolute paths only." ), mode="CLI",