Spaces:
Running on Zero
Running on Zero
| from __future__ import annotations | |
| class HTTPException(Exception): | |
| """Small FastAPI-compatible exception used by the reusable engine.""" | |
| def __init__(self, status_code: int, detail: str): | |
| self.status_code = status_code | |
| self.detail = detail | |
| super().__init__(detail) | |
| import json | |
| import math | |
| import mimetypes | |
| import os | |
| import re | |
| import shutil | |
| import subprocess | |
| import tempfile | |
| import uuid | |
| from io import BytesIO | |
| from pathlib import Path | |
| from threading import Lock | |
| from functools import lru_cache | |
| from typing import Any, Literal | |
| import pymupdf as fitz | |
| from app.font_matching import build_font_catalog, compatible_document_fonts, rank_font_candidates | |
| from app.google_fonts import catalog as google_font_catalog, ensure_font as ensure_google_font | |
| from pydantic import BaseModel, Field | |
| BASE_DIR = Path(__file__).resolve().parent | |
| DATA_DIR = Path(os.getenv('DATA_DIR', str(BASE_DIR.parent / 'data'))) | |
| DATA_DIR.mkdir(parents=True, exist_ok=True) | |
| MAX_UPLOAD_BYTES = int(os.getenv('MAX_UPLOAD_MB', '25')) * 1024 * 1024 | |
| MAX_EDITOR_IMAGE_BYTES = 12 * 1024 * 1024 | |
| MAX_EDITOR_IMAGE_PIXELS = 40000000 | |
| MAX_PAGES = int(os.getenv('MAX_PAGES', '20')) | |
| os.environ.setdefault('PADDLE_PDX_CACHE_HOME', str(DATA_DIR / 'model-cache')) | |
| _ocr_instance: Any | None = None | |
| _ocr_lock = Lock() | |
| _layout_instance: Any | None = None | |
| _layout_lock = Lock() | |
| _fontid_runtime: tuple[Any, Any, dict[str, dict[str, str]], Any] | None = None | |
| _fontid_lock = Lock() | |
| _document_storage_lock = Lock() | |
| FONTID_MODEL_ID = 'issai/FontID' | |
| FONTID_REVISION = 'bd6f54235e84e3b06a829158fa6fad829a74233d' | |
| class EditObject(BaseModel): | |
| id: str | |
| page: int = Field(ge=0) | |
| type: Literal['text', 'image'] = 'text' | |
| text: str = '' | |
| x: float = Field(ge=0) | |
| y: float = Field(ge=0) | |
| width: float = Field(gt=0) | |
| height: float = Field(gt=0) | |
| fontSize: float = Field(gt=0, le=144) | |
| rotation: float = Field(default=0, ge=-360, le=360) | |
| font: str = '' | |
| fontFamily: str = 'original' | |
| assetId: str | None = None | |
| assetExt: str | None = None | |
| color: str = Field(pattern='^#[0-9a-fA-F]{6}$') | |
| bold: bool = False | |
| italic: bool = False | |
| deleted: bool = False | |
| isNew: bool = False | |
| isBackground: bool = False | |
| class TableGroupEdit(BaseModel): | |
| id: str | |
| page: int = Field(ge=0) | |
| x: float = Field(ge=0) | |
| y: float = Field(ge=0) | |
| width: float = Field(gt=0) | |
| height: float = Field(gt=0) | |
| class ExportRequest(BaseModel): | |
| objects: list[EditObject] | |
| tableGroups: list[TableGroupEdit] = Field(default_factory=list) | |
| def _is_document_directory(path: Path) -> bool: | |
| return path.is_dir() and bool(re.fullmatch('[0-9a-f]{32}', path.name)) | |
| def purge_document_storage(keep_id: str | None=None) -> int: | |
| """Delete uploaded-document folders, preserving only the active document. | |
| Model/font caches are deliberately stored outside these 32-hex document folders, | |
| so this never removes reusable application caches. | |
| """ | |
| removed = 0 | |
| for path in DATA_DIR.iterdir(): | |
| if not _is_document_directory(path) or (keep_id and path.name == keep_id): | |
| continue | |
| shutil.rmtree(path, ignore_errors=True) | |
| removed += 1 | |
| return removed | |
| def purge_all_document_storage() -> int: | |
| with _document_storage_lock: | |
| return purge_document_storage() | |
| purge_all_document_storage() | |
| def document_dir(document_id: str) -> Path: | |
| if not re.fullmatch('[0-9a-f]{32}', document_id): | |
| raise HTTPException(status_code=404, detail='Document not found') | |
| path = DATA_DIR / document_id | |
| if not path.is_dir(): | |
| raise HTTPException(status_code=404, detail='Document not found') | |
| return path | |
| def normalized_font_name(value: str) -> str: | |
| name = value.split('+')[-1].lower() | |
| return re.sub('[^a-z0-9]', '', name) | |
| def normalized_rotation(value: Any) -> float: | |
| """Normalize an editor text rotation to the range [0, 360).""" | |
| try: | |
| angle = float(value) % 360.0 | |
| except (TypeError, ValueError): | |
| return 0.0 | |
| if abs(angle) < 0.0005 or abs(angle - 360.0) < 0.0005: | |
| return 0.0 | |
| return round(angle, 3) | |
| def pdf_text_rotation(direction: Any) -> float: | |
| """Convert PyMuPDF's line direction vector to a clockwise page angle.""" | |
| try: | |
| dx, dy = (float(direction[0]), float(direction[1])) | |
| if abs(dx) + abs(dy) < 1e-06: | |
| return 0.0 | |
| return normalized_rotation(math.degrees(math.atan2(-dy, dx))) | |
| except (TypeError, ValueError, IndexError): | |
| return 0.0 | |
| STANDARD_PDF_FONT_ALIASES = {'courier': 'cour', 'courierbold': 'cobo', 'courieroblique': 'coit', 'courierboldoblique': 'cobi', 'helvetica': 'helv', 'helveticabold': 'hebo', 'helveticaoblique': 'heit', 'helveticaboldoblique': 'hebi', 'timesroman': 'tiro', 'timesbold': 'tibo', 'timesitalic': 'tiit', 'timesbolditalic': 'tibi', 'symbol': 'symb', 'zapfdingbats': 'zapfdingbats'} | |
| def standard_pdf_font_alias(font_name: str, bold: bool=False, italic: bool=False) -> str | None: | |
| key = normalized_font_name(font_name) | |
| if key in {'helv', 'hebo', 'heit', 'hebi', 'cour', 'cobo', 'coit', 'cobi', 'tiro', 'tibo', 'tiit', 'tibi', 'symb', 'zapfdingbats'}: | |
| return key | |
| if key in {'times', 'timesnewroman'}: | |
| key = 'timesroman' | |
| alias = STANDARD_PDF_FONT_ALIASES.get(key) | |
| if alias: | |
| return alias | |
| base = key.replace('bold', '').replace('italic', '').replace('oblique', '') | |
| base_alias = STANDARD_PDF_FONT_ALIASES.get(base) | |
| if base_alias and base_alias in {'helv', 'cour', 'tiro'}: | |
| if base_alias == 'helv': | |
| return 'hebi' if bold and italic else 'hebo' if bold else 'heit' if italic else 'helv' | |
| if base_alias == 'cour': | |
| return 'cobi' if bold and italic else 'cobo' if bold else 'coit' if italic else 'cour' | |
| if base_alias == 'tiro': | |
| return 'tibi' if bold and italic else 'tibo' if bold else 'tiit' if italic else 'tiro' | |
| return None | |
| def font_xref_lookup(page: fitz.Page) -> dict[str, dict[str, Any]]: | |
| lookup: dict[str, dict[str, Any]] = {} | |
| try: | |
| for font in page.get_fonts(full=True): | |
| if len(font) < 5: | |
| continue | |
| record = {'xref': int(font[0]), 'type': str(font[2]), 'baseName': str(font[3]), 'resource': str(font[4]), 'encoding': str(font[5]) if len(font) > 5 else ''} | |
| for raw_name in (font[3], font[4]): | |
| key = normalized_font_name(str(raw_name)) | |
| if key: | |
| lookup[key] = record | |
| except Exception: | |
| return lookup | |
| return lookup | |
| def extract_embedded_fonts(pdf: fitz.Document, page: fitz.Page, folder: Path) -> dict[str, tuple[str, str]]: | |
| assets: dict[str, tuple[str, str]] = {} | |
| font_folder = folder / 'fonts' | |
| try: | |
| page_fonts = page.get_fonts(full=True) | |
| except Exception: | |
| return assets | |
| for font in page_fonts: | |
| xref = int(font[0]) | |
| if xref <= 0: | |
| continue | |
| try: | |
| base_name, extension, _font_type, data = pdf.extract_font(xref) | |
| except Exception: | |
| continue | |
| extension = str(extension).lower().lstrip('.') | |
| if extension not in {'ttf', 'otf', 'ttc', 'pfb', 'pfa'} or not data: | |
| continue | |
| asset_id = uuid.uuid4().hex | |
| font_folder.mkdir(exist_ok=True) | |
| (font_folder / f'{asset_id}.{extension}').write_bytes(data) | |
| keys = {normalized_font_name(str(base_name)), normalized_font_name(str(font[3]))} | |
| if len(font) > 4: | |
| keys.add(normalized_font_name(str(font[4]))) | |
| for key in keys: | |
| if key: | |
| assets[key] = (asset_id, extension) | |
| return assets | |
| def find_embedded_font(font_name: str, assets: dict[str, tuple[str, str]]) -> tuple[str, str] | None: | |
| key = normalized_font_name(font_name) | |
| if not key: | |
| return None | |
| if key in assets: | |
| return assets[key] | |
| matches = [asset for candidate, asset in assets.items() if len(candidate) >= 5 and (candidate in key or key in candidate)] | |
| return matches[0] if len(set(matches)) == 1 else None | |
| def span_glyph_advances(span: dict[str, Any], text: str, direction: Any | None=None) -> list[float]: | |
| """Return source-face glyph advances in PDF points for UI line reflow.""" | |
| chars = span.get('chars') or [] | |
| if len(chars) != len(text) or not chars: | |
| return [] | |
| direction = direction or span.get('dir') or (1.0, 0.0) | |
| try: | |
| dx, dy = (float(direction[0]), float(direction[1])) | |
| projected = [float(char['origin'][0]) * dx + float(char['origin'][1]) * dy for char in chars] | |
| advances: list[float] = [] | |
| for index, char in enumerate(chars): | |
| if index + 1 < len(chars): | |
| advance = abs(projected[index + 1] - projected[index]) | |
| else: | |
| bbox = [float(value) for value in char.get('bbox', ())] | |
| advance = max(0.0, bbox[2] - bbox[0]) if len(bbox) == 4 else 0.0 | |
| advances.append(round(advance, 4)) | |
| return advances | |
| except (KeyError, TypeError, ValueError, IndexError): | |
| return [] | |
| def extract_native_objects(pdf: fitz.Document, folder: Path) -> list[dict[str, Any]]: | |
| objects: list[dict[str, Any]] = [] | |
| for page_index, page in enumerate(pdf): | |
| blocks = page.get_text('rawdict').get('blocks', []) | |
| try: | |
| image_occurrences = page.get_image_info(xrefs=True) | |
| except Exception: | |
| image_occurrences = [] | |
| used_image_occurrences: set[int] = set() | |
| font_assets = extract_embedded_fonts(pdf, page, folder) | |
| font_xrefs = font_xref_lookup(page) | |
| image_folder = folder / 'images' | |
| for block_index, block in enumerate(blocks): | |
| block_id = uuid.uuid4().hex | |
| if block.get('type') == 1 and block.get('bbox') and block.get('image'): | |
| x0, y0, x1, y1 = [float(value) for value in block['bbox']] | |
| if x1 <= x0 or y1 <= y0: | |
| continue | |
| asset_id = uuid.uuid4().hex | |
| extension = re.sub('[^a-z0-9]', '', str(block.get('ext', 'png')).lower()) or 'png' | |
| image_folder.mkdir(exist_ok=True) | |
| (image_folder / f'{asset_id}.{extension}').write_bytes(block['image']) | |
| block_bbox = [x0, y0, x1, y1] | |
| matching_images = sorted(((sum((abs(float(actual) - expected) for actual, expected in zip(info.get('bbox', ()), block_bbox))), index, info) for index, info in enumerate(image_occurrences) if index not in used_image_occurrences and len(info.get('bbox', ())) == 4), key=lambda item: item[0]) | |
| image_xref = int(matching_images[0][2].get('xref', 0) or 0) if matching_images and matching_images[0][0] <= 2.0 else 0 | |
| if image_xref: | |
| used_image_occurrences.add(matching_images[0][1]) | |
| page_area = max(1.0, float(page.rect.width) * float(page.rect.height)) | |
| image_area = (x1 - x0) * (y1 - y0) | |
| is_background = image_area / page_area >= 0.62 and (x1 - x0) / max(1.0, page.rect.width) >= 0.86 and ((y1 - y0) / max(1.0, page.rect.height) >= 0.76) | |
| objects.append({'id': asset_id, 'page': page_index, 'type': 'image', 'text': '', 'originalText': '', 'x': x0, 'y': y0, 'width': x1 - x0, 'height': y1 - y0, 'originalX': x0, 'originalY': y0, 'originalWidth': x1 - x0, 'originalHeight': y1 - y0, 'fontSize': 1.0, 'font': '', 'fontFamily': 'original', 'fontFamilyHint': 'sans', 'fontSizeSource': 'not applicable', 'color': '#000000', 'bold': False, 'italic': False, 'deleted': False, 'source': 'native', 'confidence': 1.0, 'regionId': None, 'blockId': block_id, 'assetId': asset_id, 'assetExt': extension, 'imageXref': image_xref, 'isBackground': is_background}) | |
| continue | |
| lines = block.get('lines', []) | |
| line_tops = [float(line.get('bbox', [0, 0, 0, 0])[1]) for line in lines] | |
| for line_index, line in enumerate(lines): | |
| line_id = uuid.uuid4().hex | |
| line_bbox = line.get('bbox', [0, 0, 0, 0]) | |
| line_top = float(line_bbox[1]) | |
| line_height = max(1.0, float(line_bbox[3]) - line_top) | |
| next_top = line_tops[line_index + 1] if line_index + 1 < len(line_tops) else line_top + line_height * 1.2 | |
| line_step = max(line_height, next_top - line_top) | |
| line_rotation = pdf_text_rotation(line.get('dir')) | |
| for span in line.get('spans', []): | |
| chars = span.get('chars') or [] | |
| text = ''.join((str(char.get('c', '')) for char in chars)) if chars else str(span.get('text', '')) | |
| bbox = span.get('bbox') | |
| if not text.strip() or not bbox: | |
| continue | |
| x0, y0, x1, y1 = [float(value) for value in bbox] | |
| if x1 <= x0 or y1 <= y0: | |
| continue | |
| flags = int(span.get('flags', 0)) | |
| color = int(span.get('color', 0)) | |
| font_name = str(span.get('font', '')) | |
| style_name = font_name.lower() | |
| font_size = float(span.get('size', 11)) | |
| font_color = f'#{color >> 16 & 255:02x}{color >> 8 & 255:02x}{color & 255:02x}' | |
| font_bold = bool(flags & 16) or any((marker in style_name for marker in ('bold', 'heavy', 'black'))) | |
| font_italic = bool(flags & 2) or any((marker in style_name for marker in ('italic', 'oblique'))) | |
| font_family_hint = 'mono' if flags & 8 else 'serif' if flags & 4 else 'sans' | |
| embedded_font = find_embedded_font(font_name, font_assets) | |
| font_record = font_xrefs.get(normalized_font_name(font_name), {}) | |
| exact_candidate = {'family': font_name, 'value': 'original', 'confidence': 1.0, 'visualScore': 1.0, 'bold': font_bold, 'italic': font_italic, 'source': 'exact', 'assetId': embedded_font[0] if embedded_font else None} | |
| objects.append({'id': uuid.uuid4().hex, 'page': page_index, 'type': 'text', 'text': text, 'originalText': text, 'x': x0, 'y': y0, 'width': x1 - x0, 'height': y1 - y0, 'originalX': x0, 'originalY': y0, 'originalWidth': x1 - x0, 'originalHeight': y1 - y0, 'fontSize': font_size, 'rotation': line_rotation, 'originalRotation': line_rotation, 'originalFontSize': font_size, 'glyphAdvances': span_glyph_advances(span, text, line.get('dir')), 'font': font_name, 'fontFamily': 'original', 'originalFontFamily': 'original', 'fontFamilyHint': font_family_hint, 'fontSizeSource': 'PDF font metadata · exact source face', 'fontCandidates': [exact_candidate], 'fontGuess': font_name, 'fontGuessConfidence': 1.0, 'fontMatchSource': 'exact', 'fontRenderName': font_name, 'fontRenderNote': 'Exact face recorded in the PDF content stream', 'fontStyleSource': 'PDF font flags + content metadata', 'fontStyleConfidence': 1.0, 'fontAsset': embedded_font[0] if embedded_font else None, 'fontAssetExt': embedded_font[1] if embedded_font else None, 'fontXref': int(font_record.get('xref', 0) or 0), 'fontType': str(font_record.get('type', '')), 'fontResource': str(font_record.get('resource', '')), 'color': font_color, 'originalColor': font_color, 'bold': font_bold, 'originalBold': font_bold, 'italic': font_italic, 'originalItalic': font_italic, 'deleted': False, 'source': 'native', 'confidence': 1.0, 'regionId': None, 'blockId': block_id, 'blockOrder': block_index, 'lineId': line_id, 'lineIndex': line_index, 'lineTop': line_top, 'lineStep': line_step}) | |
| return objects | |
| def get_ocr() -> Any: | |
| global _ocr_instance | |
| if _ocr_instance is None: | |
| with _ocr_lock: | |
| if _ocr_instance is None: | |
| try: | |
| Path(os.environ['PADDLE_PDX_CACHE_HOME']).mkdir(parents=True, exist_ok=True) | |
| from paddleocr import PaddleOCR | |
| except Exception as exc: | |
| raise HTTPException(status_code=503, detail=f'OCR dependencies are unavailable: {exc}') from exc | |
| try: | |
| _ocr_instance = PaddleOCR(ocr_version='PP-OCRv6', use_doc_orientation_classify=False, use_doc_unwarping=False, use_textline_orientation=False, device='cpu', enable_mkldnn=False) | |
| except Exception as exc: | |
| raise HTTPException(status_code=503, detail=f'Could not load the local PaddleOCR model: {exc}') from exc | |
| return _ocr_instance | |
| def get_layout_detector() -> Any: | |
| global _layout_instance | |
| if _layout_instance is None: | |
| with _layout_lock: | |
| if _layout_instance is None: | |
| try: | |
| from paddleocr import LayoutDetection | |
| _layout_instance = LayoutDetection(model_name='PP-DocLayout-L', device='cpu', engine='paddle_static', enable_mkldnn=False) | |
| except Exception as exc: | |
| raise HTTPException(status_code=503, detail=f'Could not load the local layout model: {exc}') from exc | |
| return _layout_instance | |
| def get_fontid_runtime() -> tuple[Any, Any, dict[str, dict[str, str]], Any]: | |
| global _fontid_runtime | |
| if _fontid_runtime is not None: | |
| return _fontid_runtime | |
| with _fontid_lock: | |
| if _fontid_runtime is not None: | |
| return _fontid_runtime | |
| try: | |
| import torch | |
| from huggingface_hub import snapshot_download | |
| from transformers import Siglip2ImageProcessor | |
| from app.fontid_model import FontIDModel | |
| except Exception as exc: | |
| raise RuntimeError(f'FontID runtime dependencies are unavailable: {exc}') from exc | |
| model_folder = Path(os.getenv('FONTID_MODEL_DIR', '')) | |
| required_files = ('config.json', 'model.safetensors', 'class_mappings.json', 'preprocessor_config.json') | |
| if not model_folder.is_dir() or not all(((model_folder / name).is_file() for name in required_files)): | |
| try: | |
| model_folder = Path(snapshot_download(repo_id=FONTID_MODEL_ID, revision=FONTID_REVISION, cache_dir=os.getenv('HF_HOME', str(DATA_DIR / 'fontid-cache')), allow_patterns=[*required_files])) | |
| except Exception as exc: | |
| raise RuntimeError(f'Could not download local FontID model weights: {exc}') from exc | |
| try: | |
| torch.set_num_threads(max(1, min(4, (os.cpu_count() or 2) // 2))) | |
| processor = Siglip2ImageProcessor.from_pretrained(str(model_folder)) | |
| model = FontIDModel.from_pretrained(model_folder) | |
| mappings = json.loads((model_folder / 'class_mappings.json').read_text(encoding='utf-8')) | |
| except Exception as exc: | |
| raise RuntimeError(f'Could not load the local FontID model: {exc}') from exc | |
| _fontid_runtime = (processor, model, mappings, torch) | |
| return _fontid_runtime | |
| def predict_font_crops(crops: list[tuple[str, Any]], batch_size: int=4) -> dict[str, dict[str, Any]]: | |
| processor, model, mappings, torch = get_fontid_runtime() | |
| predictions: dict[str, dict[str, Any]] = {} | |
| for offset in range(0, len(crops), batch_size): | |
| batch = crops[offset:offset + batch_size] | |
| inputs = processor(images=[crop for _, crop in batch], max_num_patches=256, return_tensors='pt') | |
| with _fontid_lock, torch.inference_mode(): | |
| output = model(**inputs) | |
| for batch_index, (object_id, _crop) in enumerate(batch): | |
| result: dict[str, Any] = {} | |
| for head in ('font', 'style', 'lang'): | |
| probabilities = torch.softmax(output[head][batch_index], dim=-1) | |
| count = 3 if head == 'font' else 1 | |
| values, indices = torch.topk(probabilities, k=count) | |
| classes = mappings.get(head, {}) | |
| result[head] = [{'value': str(classes.get(str(int(index)), 'unknown')), 'confidence': float(score)} for score, index in zip(values.tolist(), indices.tolist())] | |
| predictions[object_id] = result | |
| return predictions | |
| def font_family_hint(family: str) -> str: | |
| name = re.sub('[^a-z0-9]', '', family.lower()) | |
| if any((token in name for token in ('mono', 'courier', 'cascadia', 'code', 'console'))): | |
| return 'mono' | |
| if any((token in name for token in ('serif', 'garamond', 'amiri', 'mincho', 'palatino', 'times', 'lora', 'merriweather', 'ruqaa'))): | |
| return 'serif' | |
| return 'sans' | |
| FONT_FILE_EXTENSIONS = {'.ttf', '.otf', '.ttc', '.pfb', '.pfa'} | |
| def _type1_font_index() -> dict[str, Path]: | |
| """Index bundled/system PostScript Type 1 fonts by PostScript filename. | |
| PDFs often reference classic TeX/Computer Modern faces such as CMBX12 by | |
| PostScript name. Those faces may be Type 1 rather than TTF/OTF, so | |
| fontconfig alone cannot reliably find them. | |
| """ | |
| roots: list[Path] = [] | |
| configured = os.getenv('PDF_EDITOR_FONT_DIRS', '') | |
| roots.extend((Path(value) for value in configured.split(os.pathsep) if value)) | |
| roots.extend([BASE_DIR.parent / 'runtime-fonts', Path('/usr/share/texlive/texmf-dist/fonts/type1'), Path('/usr/share/fonts/type1'), Path('/usr/local/share/fonts/type1'), Path('/usr/share/fonts/X11/Type1')]) | |
| index: dict[str, Path] = {} | |
| seen: set[Path] = set() | |
| for root in roots: | |
| if not root.is_dir(): | |
| continue | |
| try: | |
| files = root.rglob('*.pfb') | |
| except OSError: | |
| continue | |
| for path in files: | |
| try: | |
| resolved = path.resolve() | |
| except OSError: | |
| resolved = path | |
| if resolved in seen or not path.is_file(): | |
| continue | |
| seen.add(resolved) | |
| key = normalized_font_name(path.stem) | |
| if key and key not in index: | |
| index[key] = path | |
| return index | |
| def _font_supports_text(path: Path, text: str) -> bool: | |
| try: | |
| font = fitz.Font(fontfile=str(path)) | |
| return not text or all((character.isspace() or font.has_glyph(ord(character)) for character in text)) | |
| except Exception: | |
| return False | |
| def font_file_for_family(family: str, bold: bool, italic: bool, text: str='') -> Path | None: | |
| if not family or not re.fullmatch('[\\w .-]{1,120}', family, re.UNICODE): | |
| return None | |
| type1_index = _type1_font_index() | |
| family_key = normalized_font_name(family) | |
| exact_type1 = type1_index.get(family_key) | |
| if exact_type1 and _font_supports_text(exact_type1, text): | |
| return exact_type1 | |
| if shutil.which('fc-match'): | |
| style = 'Bold Italic' if bold and italic else 'Bold' if bold else 'Italic' if italic else 'Regular' | |
| patterns = [f'{family}:style={style}', family] | |
| for pattern in patterns: | |
| try: | |
| result = subprocess.run(['fc-match', '--format=%{file}', pattern], check=False, capture_output=True, text=True, timeout=2) | |
| candidate = Path(result.stdout.strip()) | |
| except (OSError, subprocess.TimeoutExpired): | |
| continue | |
| if result.returncode != 0 or candidate.suffix.lower() not in FONT_FILE_EXTENSIONS or (not candidate.is_file()): | |
| continue | |
| if _font_supports_text(candidate, text): | |
| return candidate | |
| return None | |
| def pdf_font_alias(path: Path, face_name: str='') -> str: | |
| alias = re.sub('[^A-Za-z0-9_]', '', face_name)[:48] | |
| return alias or re.sub('[^A-Za-z0-9_]', '', path.stem)[:48] or 'CustomFont' | |
| def crop_ink_metrics(image: Any, box: tuple[int, int, int, int]) -> dict[str, Any] | None: | |
| """Measure visible glyph pixels and foreground color inside an OCR box.""" | |
| left, top, right, bottom = box | |
| left = max(0, min(image.width, left)) | |
| right = max(left, min(image.width, right)) | |
| top = max(0, min(image.height, top)) | |
| bottom = max(top, min(image.height, bottom)) | |
| if right - left < 2 or bottom - top < 2: | |
| return None | |
| crop = image.crop((left, top, right, bottom)).convert('RGB') | |
| pixels = crop.load() | |
| width, height = crop.size | |
| border = [] | |
| stride = max(1, max(width, height) // 200) | |
| for x in range(0, width, stride): | |
| border.append(pixels[x, 0]) | |
| border.append(pixels[x, height - 1]) | |
| for y in range(stride, height - 1, stride): | |
| border.append(pixels[0, y]) | |
| border.append(pixels[width - 1, y]) | |
| background = tuple((sorted((pixel[channel] for pixel in border))[len(border) // 2] for channel in range(3))) | |
| def inspect(threshold: int) -> tuple[int, int, int, int, list[tuple[int, tuple[int, int, int]]]]: | |
| min_x, min_y, max_x, max_y = (width, height, -1, -1) | |
| ink_count = 0 | |
| foreground: list[tuple[int, tuple[int, int, int]]] = [] | |
| sample_stride = max(1, width * height // 12000) | |
| for y in range(height): | |
| for x in range(width): | |
| color = pixels[x, y] | |
| contrast = max((abs(color[channel] - background[channel]) for channel in range(3))) | |
| if contrast < threshold: | |
| continue | |
| ink_count += 1 | |
| min_x = min(min_x, x) | |
| min_y = min(min_y, y) | |
| max_x = max(max_x, x) | |
| max_y = max(max_y, y) | |
| if contrast >= max(threshold, 40) and (y * width + x) % sample_stride == 0: | |
| foreground.append((contrast, color)) | |
| return (min_x, min_y, max_x, max_y, foreground) | |
| min_x, min_y, max_x, max_y, foreground = inspect(26) | |
| if max_x < min_x or max_y < min_y: | |
| min_x, min_y, max_x, max_y, foreground = inspect(12) | |
| if max_x < min_x or max_y < min_y: | |
| return None | |
| if foreground: | |
| cutoff = max(26, int(max((value[0] for value in foreground)) * 0.65)) | |
| solid = [color for contrast, color in foreground if contrast >= cutoff] | |
| foreground_color = tuple((sorted((color[channel] for color in solid))[len(solid) // 2] for channel in range(3))) | |
| else: | |
| foreground_color = (17, 17, 17) | |
| return {'width': max_x - min_x + 1, 'height': max_y - min_y + 1, 'color': '#' + ''.join((f'{value:02x}' for value in foreground_color))} | |
| def estimate_ocr_font_size(text: str, ink: dict[str, Any] | None, scale_x: float, scale_y: float, font: fitz.Font, fallback_box_height: float) -> tuple[float, float]: | |
| width_estimate = height_estimate = None | |
| try: | |
| glyph_width = font.text_length(text.strip(), fontsize=1) | |
| if ink and glyph_width > 0: | |
| width_estimate = ink['width'] * scale_x / glyph_width | |
| except Exception: | |
| pass | |
| try: | |
| glyph_bounds = [font.glyph_bbox(ord(character)) for character in text if not character.isspace()] | |
| glyph_height = max((bounds.y1 for bounds in glyph_bounds), default=0) - min((bounds.y0 for bounds in glyph_bounds), default=0) | |
| if ink and glyph_height > 0: | |
| height_estimate = ink['height'] * scale_y / glyph_height | |
| except Exception: | |
| pass | |
| estimates = [value for value in (width_estimate, height_estimate) if value and math.isfinite(value) and (value > 0)] | |
| if not estimates: | |
| return (max(1.0, min(144.0, fallback_box_height / 0.72)), 0.28) | |
| if len(estimates) == 1: | |
| return (max(1.0, min(144.0, estimates[0])), 0.5) | |
| agreement = min(estimates) / max(estimates) | |
| combined = math.exp(0.68 * math.log(width_estimate) + 0.32 * math.log(height_estimate)) | |
| return (max(1.0, min(144.0, combined)), max(0.2, min(0.98, 0.35 + 0.6 * agreement))) | |
| def annotate_ocr_fonts(image_path: Path, objects: list[dict[str, Any]], scale_x: float, scale_y: float, document_objects: list[dict[str, Any]] | None=None, folder: Path | None=None) -> tuple[str, str | None]: | |
| """Match OCR fonts against this PDF and open fonts; use FontID as a hint.""" | |
| from PIL import Image | |
| supported = [item for item in objects if item.get('text', '').strip()] | |
| if not supported: | |
| return ('FontID not needed', None) | |
| fontid_error = None | |
| try: | |
| scripts_supported = any((any((character.isascii() and character.isalpha() or 1024 <= ord(character) <= 1327 or 1536 <= ord(character) <= 2303 or (19968 <= ord(character) <= 40959) for character in item['text'])) for item in supported)) | |
| if scripts_supported: | |
| get_fontid_runtime() | |
| except Exception as exc: | |
| fontid_error = str(exc) | |
| predictions: dict[str, dict[str, Any]] = {} | |
| if fontid_error is None: | |
| with Image.open(image_path) as opened: | |
| page_image = opened.convert('RGB') | |
| font_crops: list[tuple[str, Any]] = [] | |
| for item in supported: | |
| box = item.get('_ocrBoxPx') | |
| if not box: | |
| continue | |
| ink = crop_ink_metrics(page_image, tuple(box)) | |
| item['_inkMetrics'] = ink | |
| chars_supported = any((character.isascii() and character.isalpha() or 1024 <= ord(character) <= 1327 or 1536 <= ord(character) <= 2303 or (19968 <= ord(character) <= 40959) for character in item['text'])) | |
| if chars_supported: | |
| font_crops.append((item['id'], page_image.crop(tuple(box)))) | |
| if font_crops: | |
| try: | |
| predictions = predict_font_crops(font_crops) | |
| except Exception as exc: | |
| fontid_error = str(exc) | |
| font_catalog = build_font_catalog(document_objects or [], folder) if folder else [] | |
| font_cache: dict[tuple[str, bool, bool, str, str], tuple[Path | None, fitz.Font]] = {} | |
| with Image.open(image_path) as opened: | |
| page_image = opened.convert('RGB') | |
| for item in supported: | |
| predicted = predictions.get(item['id'], {}) | |
| font_matches = predicted.get('font', []) | |
| pixel_box = item.get('_ocrBoxPx') | |
| ranked = rank_font_candidates(page_image, tuple(pixel_box), item['text'], font_catalog, font_matches) if pixel_box else [] | |
| best_match = ranked[0] if ranked else None | |
| document_profiles = compatible_document_fonts(item['text'], font_catalog) if not best_match else [] | |
| font_reference = best_match or (document_profiles[0] if document_profiles else None) | |
| family = font_reference['family'] if font_reference else font_matches[0]['value'] if font_matches else '' | |
| style = predicted.get('style', [{}])[0].get('value', 'Regular') | |
| bold = font_reference['bold'] if font_reference else style in {'Bold', 'BoldItalic'} | |
| italic = font_reference['italic'] if font_reference else style in {'Italic', 'BoldItalic'} | |
| hint = font_family_hint(family) if family else 'sans' | |
| preferred_path = None | |
| if font_reference and font_reference.get('assetId') and folder: | |
| preferred_path = next((folder / 'fonts' / f"{font_reference['assetId']}.{ext}" for ext in ('ttf', 'otf') if (folder / 'fonts' / f"{font_reference['assetId']}.{ext}").is_file()), None) | |
| cache_key = (family or hint, bold, italic, item['text'], str(preferred_path or '')) | |
| if cache_key not in font_cache: | |
| font_path = preferred_path or font_file_for_family(family or {'sans': 'DejaVu Sans', 'serif': 'DejaVu Serif', 'mono': 'DejaVu Sans Mono'}[hint], bold, italic, item['text']) | |
| try: | |
| font_object = fitz.Font(fontfile=str(font_path)) if font_path else fitz.Font(fontname={'sans': 'helv', 'serif': 'tiro', 'mono': 'cour'}[hint]) | |
| except Exception: | |
| font_path = None | |
| font_object = fitz.Font(fontname='helv') | |
| font_cache[cache_key] = (font_path, font_object) | |
| font_path, font_object = font_cache[cache_key] | |
| box = item.get('_ocrBoxPx') | |
| if box is None: | |
| box = (round(item['x'] / scale_x), round(item['y'] / scale_y), round((item['x'] + item['width']) / scale_x), round((item['y'] + item['height']) / scale_y)) | |
| ink = item.pop('_inkMetrics', None) or crop_ink_metrics(page_image, tuple(box)) | |
| size, size_confidence = estimate_ocr_font_size(item['text'], ink, scale_x, scale_y, font_object, float(item['originalHeight'])) | |
| if best_match: | |
| candidates = ranked | |
| source_label = 'PDF font' if best_match['source'] == 'document' else 'installed open font' | |
| font_label = f'Best rendered-glyph match: {family} ({source_label})' | |
| family_confidence = best_match['visualScore'] | |
| elif document_profiles: | |
| candidates = document_profiles | |
| font_label = f'Embedded PDF font supports this text: {family}' | |
| family_confidence = None | |
| elif font_matches: | |
| candidates = [{'family': entry['value'], 'confidence': entry['confidence'], 'value': f"fontid:{entry['value']}", 'source': 'classifier', 'bold': bold, 'italic': italic} for entry in font_matches] | |
| font_label = f'FontID category suggestion: {family}' | |
| family_confidence = None | |
| else: | |
| candidates = [] | |
| family_confidence = None | |
| font_label = 'Font family not identified · choose a font manually' | |
| if ink: | |
| color = ink['color'] | |
| color_source = 'Sampled from scanned glyph pixels' | |
| else: | |
| color = '#111111' | |
| color_source = 'Dark text fallback' | |
| item.update({'fontSize': size, 'originalFontSize': size, 'glyphAdvances': [round(font_object.text_length(character, fontsize=size), 4) for character in item['text']], 'font': font_label, 'fontFamily': 'original', 'originalFontFamily': 'original', 'fontFamilyHint': hint, 'fontGuess': family or None, 'fontCandidates': candidates, 'fontGuessConfidence': family_confidence, 'fontMatchSource': best_match['source'] if best_match else 'document-profile' if document_profiles else 'classifier' if font_matches else 'fallback', 'fontSizeSource': 'OCR glyph width/height fitted to font metrics', 'fontSizeConfidence': size_confidence, 'fontColorSource': color_source, 'color': color, 'originalColor': color, 'bold': bold, 'originalBold': bold, 'italic': italic, 'originalItalic': italic, 'fontSystemPath': str(font_path) if font_path else None, 'fontRenderName': font_object.name, 'fontRenderNote': '' if not family or (font_object.name and normalized_font_name(family) in normalized_font_name(font_object.name)) else f'Closest installed open font: {font_object.name}', 'fontStyleSource': 'Rendered-glyph face match' if best_match else 'PDF font metadata' if document_profiles else 'FontID visual estimate' if predicted.get('style') else 'DejaVu/Noto style fallback', 'fontStyleConfidence': None if font_reference else predicted.get('style', [{}])[0].get('confidence')}) | |
| item.pop('_ocrBoxPx', None) | |
| matched = sum((bool(item.get('fontCandidates')) and item.get('fontMatchSource') in {'document', 'installed', 'document-profile'} for item in supported)) | |
| if predictions and matched: | |
| return (f'FontID + open-set glyph matching · {matched} font matches', fontid_error) | |
| if matched: | |
| return (f'PDF/open-font glyph matching · {matched} font matches', fontid_error) | |
| if predictions: | |
| return (f'FontID category suggestions · {len(predictions)} samples', fontid_error) | |
| return ('OCR glyph-metric estimate' if fontid_error else 'Font matching unavailable', fontid_error) | |
| def read_model_result(result: Any, destination: Path) -> dict[str, Any]: | |
| result.save_to_json(str(destination)) | |
| payload = json.loads(destination.read_text(encoding='utf-8')) | |
| return payload.get('res', payload) | |
| def recognize_page(image_path: Path, page_index: int, scale_x: float, scale_y: float) -> list[dict[str, Any]]: | |
| detected: list[dict[str, Any]] = [] | |
| result_path = image_path.with_name('ocr-result.json') | |
| for result in get_ocr().predict(str(image_path)): | |
| detail = read_model_result(result, result_path) | |
| texts = detail.get('rec_texts', []) | |
| scores = detail.get('rec_scores', []) | |
| boxes = detail.get('rec_boxes', []) | |
| polygons = detail.get('rec_polys', []) or detail.get('rec_polygons', []) or [] | |
| for i, text in enumerate(texts): | |
| if not str(text).strip() or i >= len(boxes): | |
| continue | |
| x0, y0, x1, y1 = [float(value) for value in boxes[i]] | |
| rotation = 0.0 | |
| if i < len(polygons): | |
| try: | |
| points = polygons[i] | |
| if len(points) >= 2: | |
| p0, p1 = (points[0], points[1]) | |
| rotation = normalized_rotation(math.degrees(math.atan2(-(float(p1[1]) - float(p0[1])), float(p1[0]) - float(p0[0])))) | |
| except (TypeError, ValueError, IndexError): | |
| rotation = 0.0 | |
| estimated_height = max(1.0, (y1 - y0) * scale_y / 0.72) | |
| detected.append({'id': uuid.uuid4().hex, 'page': page_index, 'text': str(text), 'originalText': str(text), 'x': x0 * scale_x, 'y': y0 * scale_y, 'width': max(1.0, (x1 - x0) * scale_x), 'height': max(1.0, (y1 - y0) * scale_y), 'fontSize': max(1.0, min(144.0, estimated_height)), 'originalFontSize': max(1.0, min(144.0, estimated_height)), 'font': 'FontID visual match pending', 'fontFamily': 'original', 'originalFontFamily': 'original', 'fontFamilyHint': 'sans', 'fontSizeSource': 'OCR glyph-height estimate', 'fontSizeConfidence': 0.3, 'color': '#111111', 'originalColor': '#111111', 'bold': False, 'originalBold': False, 'italic': False, 'originalItalic': False, 'deleted': False, 'source': 'ocr', 'confidence': float(scores[i]) if i < len(scores) else None, 'regionId': None, 'type': 'text', 'originalX': x0 * scale_x, 'originalY': y0 * scale_y, 'originalWidth': max(1.0, (x1 - x0) * scale_x), 'originalHeight': max(1.0, (y1 - y0) * scale_y), 'blockId': None, 'lineId': None, '_ocrBoxPx': [x0, y0, x1, y1]}) | |
| return detected | |
| def remove_native_text_duplicates(ocr_objects: list[dict[str, Any]], native_objects: list[dict[str, Any]]) -> list[dict[str, Any]]: | |
| """Keep OCR detections for scanned content while avoiding duplicate selectable PDF text.""" | |
| native_text = [item for item in native_objects if item.get('type', 'text') == 'text'] | |
| supplemental: list[dict[str, Any]] = [] | |
| for ocr in ocr_objects: | |
| ox0, oy0 = (float(ocr['x']), float(ocr['y'])) | |
| ox1, oy1 = (ox0 + float(ocr['width']), oy0 + float(ocr['height'])) | |
| ocr_area = max(1.0, float(ocr['width']) * float(ocr['height'])) | |
| center_x, center_y = ((ox0 + ox1) / 2, (oy0 + oy1) / 2) | |
| duplicate = False | |
| for native in native_text: | |
| nx0, ny0 = (float(native['x']), float(native['y'])) | |
| nx1, ny1 = (nx0 + float(native['width']), ny0 + float(native['height'])) | |
| intersection = max(0.0, min(ox1, nx1) - max(ox0, nx0)) * max(0.0, min(oy1, ny1) - max(oy0, ny0)) | |
| contains_center = nx0 <= center_x <= nx1 and ny0 <= center_y <= ny1 | |
| if intersection / ocr_area >= 0.45 or contains_center: | |
| duplicate = True | |
| break | |
| if not duplicate: | |
| supplemental.append(ocr) | |
| return supplemental | |
| def detect_layout(image_path: Path, page_index: int, scale_x: float, scale_y: float) -> list[dict[str, Any]]: | |
| regions: list[dict[str, Any]] = [] | |
| result_path = image_path.with_name('layout-result.json') | |
| for result in get_layout_detector().predict(str(image_path), batch_size=1, layout_nms=True): | |
| detail = read_model_result(result, result_path) | |
| for index, box in enumerate(detail.get('boxes', [])): | |
| coords = box.get('coordinate') or [] | |
| if len(coords) != 4: | |
| continue | |
| x0, y0, x1, y1 = [float(value) for value in coords] | |
| if x1 <= x0 or y1 <= y0: | |
| continue | |
| regions.append({'id': uuid.uuid4().hex, 'page': page_index, 'label': str(box.get('label', 'content')), 'confidence': float(box.get('score', 0)), 'order': index, 'x': x0 * scale_x, 'y': y0 * scale_y, 'width': (x1 - x0) * scale_x, 'height': (y1 - y0) * scale_y}) | |
| return regions | |
| def assign_regions(objects: list[dict[str, Any]], regions: list[dict[str, Any]], page_index: int) -> None: | |
| for obj in objects: | |
| if obj.get('page') != page_index: | |
| continue | |
| cx = float(obj['x']) + float(obj['width']) / 2 | |
| cy = float(obj['y']) + float(obj['height']) / 2 | |
| candidates = [r for r in regions if r['x'] <= cx <= r['x'] + r['width'] and r['y'] <= cy <= r['y'] + r['height']] | |
| if candidates: | |
| smallest = min(candidates, key=lambda r: r['width'] * r['height']) | |
| obj['regionId'] = smallest['id'] | |
| else: | |
| obj['regionId'] = None | |
| ocr_regions: dict[str, list[dict[str, Any]]] = {} | |
| for obj in objects: | |
| if obj.get('page') == page_index and obj.get('source') == 'ocr': | |
| key = obj.get('regionId') or 'unassigned' | |
| ocr_regions.setdefault(key, []).append(obj) | |
| for region_id, region_objects in ocr_regions.items(): | |
| lines: list[list[dict[str, Any]]] = [] | |
| for obj in sorted(region_objects, key=lambda item: (item['y'], item['x'])): | |
| center_y = obj['y'] + obj['height'] / 2 | |
| line = next((candidate for candidate in reversed(lines) if abs(center_y - (candidate[0]['y'] + candidate[0]['height'] / 2)) <= max(2.5, min(obj['height'], candidate[0]['height']) * 0.45)), None) | |
| if line is None: | |
| lines.append([obj]) | |
| else: | |
| line.append(obj) | |
| lines.sort(key=lambda line: min((item['y'] for item in line))) | |
| tops = [min((item['y'] for item in line)) for line in lines] | |
| for index, line in enumerate(lines): | |
| top = tops[index] | |
| line_step = tops[index + 1] - top if index + 1 < len(tops) else max((item['height'] for item in line)) * 1.35 | |
| line_id = uuid.uuid4().hex | |
| for obj in line: | |
| obj['blockId'] = f'ocr-{region_id}' if region_id != 'unassigned' else None | |
| obj['lineId'] = line_id | |
| obj['lineIndex'] = index | |
| obj['lineTop'] = top | |
| obj['lineStep'] = max(line_step, obj['height']) | |
| def render_analysis_image(pdf: fitz.Document, page_index: int, output: Path, scale: float=2.0) -> tuple[float, float]: | |
| page = pdf[page_index] | |
| pixmap = page.get_pixmap(matrix=fitz.Matrix(scale, scale), alpha=False) | |
| pixmap.save(output) | |
| return (page.rect.width / pixmap.width, page.rect.height / pixmap.height) | |
| def save_document_model(folder: Path, model: dict[str, Any]) -> None: | |
| model_path = folder / 'document.json' | |
| temp_path = folder / 'document.json.tmp' | |
| temp_path.write_text(json.dumps(model, ensure_ascii=False, indent=2), encoding='utf-8') | |
| temp_path.replace(model_path) | |
| def analyze_page(document_id: str, page_number: int, refresh: bool=False) -> dict[str, Any]: | |
| folder = document_dir(document_id) | |
| model_path = folder / 'document.json' | |
| document = json.loads(model_path.read_text(encoding='utf-8')) | |
| if page_number < 0 or page_number >= len(document['pages']): | |
| raise HTTPException(status_code=404, detail='Page not found') | |
| ocr_detection_count = 0 | |
| font_model_status = 'No scanned text' | |
| font_model_error: str | None = None | |
| try: | |
| pdf = fitz.open(folder / 'original.pdf') | |
| with tempfile.TemporaryDirectory() as tmp: | |
| image_path = Path(tmp) / 'page.png' | |
| scale_x, scale_y = render_analysis_image(pdf, page_number, image_path) | |
| page_objects = [obj for obj in document['objects'] if obj['page'] == page_number] | |
| native_text = any((obj.get('source') == 'native' and obj.get('type', 'text') == 'text' for obj in page_objects)) | |
| cached_ocr = bool(document['pages'][page_number].get('ocrAnalyzed')) or any((obj.get('source') == 'ocr' for obj in page_objects)) | |
| ocr_ran = refresh or not cached_ocr | |
| recognized = recognize_page(image_path, page_number, scale_x, scale_y) if ocr_ran else [] | |
| ocr_detection_count = len(recognized) if ocr_ran else sum((obj.get('page') == page_number and obj.get('source') == 'ocr' for obj in page_objects)) | |
| native_objects = [obj for obj in page_objects if obj.get('source') == 'native'] | |
| new_objects = remove_native_text_duplicates(recognized, native_objects) | |
| regions = detect_layout(image_path, page_number, scale_x, scale_y) | |
| ocr_objects = new_objects if ocr_ran else [obj for obj in page_objects if obj.get('source') == 'ocr' and obj.get('type') == 'text'] | |
| font_model_status, font_model_error = annotate_ocr_fonts(image_path, ocr_objects, scale_x, scale_y, document.get('objects', []), folder) | |
| pdf.close() | |
| except HTTPException: | |
| raise | |
| except Exception as exc: | |
| raise HTTPException(status_code=500, detail=f'Page analysis failed: {exc}') from exc | |
| if ocr_ran: | |
| document['objects'] = [obj for obj in document['objects'] if not (obj['page'] == page_number and obj.get('source') == 'ocr')] | |
| document['objects'].extend(new_objects) | |
| assign_regions(document['objects'], regions, page_number) | |
| document['pages'][page_number]['regions'] = regions | |
| document['pages'][page_number]['analyzed'] = True | |
| document['pages'][page_number]['layoutModel'] = 'PP-DocLayout-L' | |
| document['pages'][page_number]['ocrAnalyzed'] = True | |
| document['pages'][page_number]['hasText'] = native_text or any((obj.get('page') == page_number and obj.get('source') == 'ocr' for obj in document['objects'])) | |
| save_document_model(folder, document) | |
| return {'layoutModel': 'PP-DocLayout-L', 'ocrModel': 'PaddleOCR PP-OCRv6 + PDF text extraction', 'ocrUsed': True, 'ocrRan': ocr_ran, 'ocrDetections': ocr_detection_count, 'fontModel': font_model_status, 'fontModelError': font_model_error[:500] if font_model_error else None, 'objects': [obj for obj in document['objects'] if obj['page'] == page_number], 'regions': regions} | |
| def ocr_page(document_id: str, page_number: int) -> dict[str, Any]: | |
| """Compatibility endpoint: OCR only. Prefer /analyze for OCR plus layout regions.""" | |
| folder = document_dir(document_id) | |
| model_path = folder / 'document.json' | |
| document = json.loads(model_path.read_text(encoding='utf-8')) | |
| if page_number < 0 or page_number >= len(document['pages']): | |
| raise HTTPException(status_code=404, detail='Page not found') | |
| try: | |
| pdf = fitz.open(folder / 'original.pdf') | |
| with tempfile.TemporaryDirectory() as tmp: | |
| image_path = Path(tmp) / 'page.png' | |
| scale_x, scale_y = render_analysis_image(pdf, page_number, image_path) | |
| detected = recognize_page(image_path, page_number, scale_x, scale_y) | |
| font_model_status, font_model_error = annotate_ocr_fonts(image_path, detected, scale_x, scale_y, document.get('objects', []), folder) | |
| pdf.close() | |
| except HTTPException: | |
| raise | |
| except Exception as exc: | |
| raise HTTPException(status_code=500, detail=f'OCR failed: {exc}') from exc | |
| document['objects'].extend(detected) | |
| document['pages'][page_number]['hasText'] = bool(detected) or document['pages'][page_number]['hasText'] | |
| save_document_model(folder, document) | |
| return {'model': 'PaddleOCR PP-OCRv6', 'fontModel': font_model_status, 'fontModelError': font_model_error[:500] if font_model_error else None, 'objects': detected} | |
| def insertion_font(item: EditObject, source_object: dict[str, Any], folder: Path) -> tuple[str, str | None]: | |
| source_font = str(source_object.get('font', '')) | |
| source_is_native = source_object.get('source') == 'native' | |
| font_asset = source_object.get('fontAsset') | |
| extension = source_object.get('fontAssetExt') | |
| family = item.fontFamily | |
| exact_original_style = family == 'original' and item.bold == source_object.get('bold', False) and (item.italic == source_object.get('italic', False)) | |
| if exact_original_style and font_asset and (extension in {'ttf', 'otf'}): | |
| path = folder / 'fonts' / f'{font_asset}.{extension}' | |
| if path.is_file(): | |
| try: | |
| embedded = fitz.Font(fontfile=str(path)) | |
| if all((character.isspace() or embedded.has_glyph(ord(character)) for character in item.text)): | |
| return (f'source_{font_asset}', str(path)) | |
| if source_is_native: | |
| raise HTTPException(status_code=422, detail=f'The exact PDF font "{source_font or embedded.name}" does not contain all replacement characters. Choose another font or keep the original text.') | |
| except HTTPException: | |
| raise | |
| except Exception: | |
| pass | |
| if family.startswith('document:'): | |
| asset_id = family.removeprefix('document:') | |
| match = next((candidate for candidate in source_object.get('fontCandidates', []) if candidate.get('value') == family), None) | |
| if re.fullmatch('[0-9a-f]{32}', asset_id): | |
| for asset_ext in ('ttf', 'otf'): | |
| path = folder / 'fonts' / f'{asset_id}.{asset_ext}' | |
| if not path.is_file(): | |
| continue | |
| try: | |
| embedded = fitz.Font(fontfile=str(path)) | |
| compatible_style = match is None or (item.bold, item.italic) == (match.get('bold'), match.get('italic')) | |
| if compatible_style and all((character.isspace() or embedded.has_glyph(ord(character)) for character in item.text)): | |
| return (f'source_{asset_id}', str(path)) | |
| except Exception: | |
| pass | |
| if match: | |
| font_path = font_file_for_family(match.get('family', ''), item.bold, item.italic, item.text) | |
| if font_path: | |
| try: | |
| font = fitz.Font(fontfile=str(font_path)) | |
| return (pdf_font_alias(font_path, font.name), str(font_path)) | |
| except Exception: | |
| pass | |
| if family.startswith('fontid:'): | |
| recognized_family = family.removeprefix('fontid:') | |
| font_path = font_file_for_family(recognized_family, item.bold, item.italic, item.text) | |
| if font_path: | |
| try: | |
| font = fitz.Font(fontfile=str(font_path)) | |
| return (pdf_font_alias(font_path, font.name), str(font_path)) | |
| except Exception: | |
| pass | |
| if family.startswith('google:'): | |
| requested_family = family.removeprefix('google:') | |
| font_path = ensure_google_font(requested_family, item.bold, item.italic) | |
| if font_path and _font_supports_text(font_path, item.text): | |
| try: | |
| font = fitz.Font(fontfile=str(font_path)) | |
| return (pdf_font_alias(font_path, font.name), str(font_path)) | |
| except Exception: | |
| pass | |
| if font_path: | |
| raise HTTPException(status_code=422, detail=f'Google Font "{requested_family}" does not contain all replacement characters.') | |
| if family.startswith('installed:'): | |
| requested_family = family.removeprefix('installed:') | |
| font_path = font_file_for_family(requested_family, item.bold, item.italic, item.text) | |
| if font_path: | |
| try: | |
| font = fitz.Font(fontfile=str(font_path)) | |
| return (pdf_font_alias(font_path, font.name), str(font_path)) | |
| except Exception: | |
| pass | |
| if exact_original_style: | |
| standard_alias = standard_pdf_font_alias(source_font, item.bold, item.italic) | |
| if standard_alias: | |
| try: | |
| exact_font = fitz.Font(fontname=standard_alias) | |
| if all((character.isspace() or exact_font.has_glyph(ord(character)) for character in item.text)): | |
| return (standard_alias, None) | |
| except Exception: | |
| pass | |
| system_path = source_object.get('fontSystemPath') | |
| if system_path: | |
| path = Path(system_path) | |
| if path.suffix.lower() in FONT_FILE_EXTENSIONS and path.is_file(): | |
| try: | |
| visual_font = fitz.Font(fontfile=str(path)) | |
| if all((character.isspace() or visual_font.has_glyph(ord(character)) for character in item.text)): | |
| return (pdf_font_alias(path, visual_font.name), str(path)) | |
| except Exception: | |
| pass | |
| source_family = source_font.split('+')[-1].replace('_', ' ').strip() | |
| if source_family: | |
| source_path = font_file_for_family(source_family, item.bold, item.italic, item.text) | |
| if source_path: | |
| try: | |
| resolved_font = fitz.Font(fontfile=str(source_path)) | |
| resolved_key = normalized_font_name(resolved_font.name) | |
| source_key = normalized_font_name(source_family) | |
| path_key = normalized_font_name(source_path.stem) | |
| if path_key == source_key: | |
| return (pdf_font_alias(source_path, source_family), str(source_path)) | |
| if not source_is_native or source_key in resolved_key or resolved_key in source_key or (len(source_key) >= 6 and source_key[:6] in resolved_key): | |
| return (pdf_font_alias(source_path, resolved_font.name), str(source_path)) | |
| except Exception: | |
| pass | |
| hint = (source_object.get('fontFamilyHint') or source_font).lower() | |
| if family == 'original': | |
| family = 'mono' if any((key in hint for key in ('courier', 'mono', 'consol'))) else 'serif' if any((key in hint for key in ('times', 'serif', 'garamond', 'bookman'))) else 'sans' | |
| prefix = {'serif': 'DejaVuSerif', 'mono': 'DejaVuSansMono'}.get(family, 'DejaVuSans') | |
| italic_suffix = 'Italic' if prefix == 'DejaVuSerif' else 'Oblique' | |
| suffix = f'-Bold{italic_suffix}' if item.bold and item.italic else '-Bold' if item.bold else f'-{italic_suffix}' if item.italic else '' | |
| font_path = Path('/usr/share/fonts/truetype/dejavu') / f'{prefix}{suffix}.ttf' | |
| if font_path.is_file(): | |
| try: | |
| fallback_font = fitz.Font(fontfile=str(font_path)) | |
| if all((character.isspace() or fallback_font.has_glyph(ord(character)) for character in item.text)): | |
| return (pdf_font_alias(font_path, font_path.stem), str(font_path)) | |
| except Exception: | |
| pass | |
| script_fonts = [('NotoSansDevanagari-Regular.ttf', range(2304, 2432)), ('NotoSansBengali-Regular.ttf', range(2432, 2560)), ('NotoSansGurmukhi-Regular.ttf', range(2560, 2688)), ('NotoSansGujarati-Regular.ttf', range(2688, 2816)), ('NotoSansOriya-Regular.ttf', range(2816, 2944)), ('NotoSansTamil-Regular.ttf', range(2944, 3072)), ('NotoSansTelugu-Regular.ttf', range(3072, 3200)), ('NotoSansKannada-Regular.ttf', range(3200, 3328)), ('NotoSansMalayalam-Regular.ttf', range(3328, 3456)), ('NotoSansArabic-Regular.ttf', range(1536, 1792)), ('NotoSansThai-Regular.ttf', range(3584, 3712))] | |
| chars = [ord(character) for character in item.text] | |
| for filename, codepoints in script_fonts: | |
| if not any((codepoint in codepoints for codepoint in chars)): | |
| continue | |
| candidate = Path('/usr/share/fonts/truetype/noto') / filename | |
| if candidate.is_file(): | |
| try: | |
| script_font = fitz.Font(fontfile=str(candidate)) | |
| if all((chr(codepoint).isspace() or script_font.has_glyph(codepoint) for codepoint in chars)): | |
| return (pdf_font_alias(candidate, candidate.stem), str(candidate)) | |
| except Exception: | |
| pass | |
| cjk_path = Path('/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc') | |
| if any((11904 <= codepoint <= 40959 or 63744 <= codepoint <= 64255 for codepoint in chars)) and cjk_path.is_file(): | |
| try: | |
| cjk_font = fitz.Font(fontfile=str(cjk_path)) | |
| if all((chr(codepoint).isspace() or cjk_font.has_glyph(codepoint) for codepoint in chars)): | |
| return ('NotoSansCJK', str(cjk_path)) | |
| except Exception: | |
| pass | |
| fallback = {(False, False): 'helv', (True, False): 'hebo', (False, True): 'heit', (True, True): 'hebi'} | |
| return (fallback[item.bold, item.italic], None) | |
| def insert_editor_text(page: fitz.Page, edit: EditObject, font_name: str, font_file: str | None, color: tuple[float, float, float], overlay: bool=True) -> None: | |
| """Insert vector text at its inspector geometry and rotate around the box center.""" | |
| try: | |
| font = fitz.Font(fontfile=font_file) if font_file else fitz.Font(fontname=font_name) | |
| ascender = min(1.5, max(0.5, float(font.ascender))) | |
| except Exception: | |
| ascender = 0.9 | |
| angle = normalized_rotation(edit.rotation) | |
| point = fitz.Point(edit.x, edit.y + edit.fontSize * ascender) | |
| if abs(angle) < 0.0005: | |
| page.insert_text(point, edit.text, fontsize=edit.fontSize, fontname=font_name, fontfile=font_file, color=color, overlay=overlay) | |
| return | |
| center = fitz.Point(edit.x + edit.width / 2.0, edit.y + edit.height / 2.0) | |
| matrix = fitz.Matrix(1, 1).prerotate(angle) | |
| page.insert_text(point, edit.text, fontsize=edit.fontSize, fontname=font_name, fontfile=font_file, color=color, overlay=overlay, morph=(center, matrix)) | |
| def sample_background_color(page: fitz.Page, rect: fitz.Rect) -> tuple[float, float, float]: | |
| margin = max(2.0, min(6.0, rect.height * 0.18)) | |
| page_rect = page.rect | |
| clip = fitz.Rect(max(page_rect.x0, rect.x0 - margin), max(page_rect.y0, rect.y0 - margin), min(page_rect.x1, rect.x1 + margin), min(page_rect.y1, rect.y1 + margin)) | |
| if clip.is_empty or clip.width <= 0 or clip.height <= 0: | |
| return (1.0, 1.0, 1.0) | |
| try: | |
| pixmap = page.get_pixmap(matrix=fitz.Matrix(2, 2), clip=clip, alpha=False) | |
| except Exception: | |
| return (1.0, 1.0, 1.0) | |
| if pixmap.width < 3 or pixmap.height < 3 or pixmap.n < 3: | |
| return (1.0, 1.0, 1.0) | |
| stride = pixmap.n | |
| pixels = pixmap.samples | |
| samples: list[tuple[int, int, int]] = [] | |
| for x in range(pixmap.width): | |
| for y in (0, pixmap.height - 1): | |
| offset = (y * pixmap.width + x) * stride | |
| samples.append((pixels[offset], pixels[offset + 1], pixels[offset + 2])) | |
| for y in range(1, pixmap.height - 1): | |
| for x in (0, pixmap.width - 1): | |
| offset = (y * pixmap.width + x) * stride | |
| samples.append((pixels[offset], pixels[offset + 1], pixels[offset + 2])) | |
| samples.sort(key=lambda pixel: pixel[0] + pixel[1] + pixel[2]) | |
| channel_values = [] | |
| for channel in range(3): | |
| values = sorted((pixel[channel] for pixel in samples)) | |
| channel_values.append(values[len(values) // 2] / 255) | |
| return tuple(channel_values) | |
| def patch_background_raster_text(folder: Path, image_object: dict[str, Any], edits: list[tuple[EditObject, dict[str, Any]]], source_by_id: dict[str, dict[str, Any]]) -> bytes: | |
| """Edit OCR text in its backing image so every foreground PDF layer stays intact.""" | |
| from PIL import Image, ImageDraw, ImageFont | |
| asset_id = str(image_object.get('assetId', '')) | |
| extension = str(image_object.get('assetExt', '')) | |
| if not re.fullmatch('[0-9a-f]{32}', asset_id) or not re.fullmatch('png|jpe?g|webp', extension): | |
| raise HTTPException(status_code=409, detail='The detected background image cannot be edited independently') | |
| image_path = folder / 'images' / f'{asset_id}.{extension}' | |
| if not image_path.is_file(): | |
| raise HTTPException(status_code=400, detail='The source background image asset is missing') | |
| try: | |
| with Image.open(image_path) as opened: | |
| image = opened.convert('RGBA') | |
| except Exception as exc: | |
| raise HTTPException(status_code=415, detail='The source background image could not be decoded') from exc | |
| px_per_point_x = image.width / max(0.01, float(image_object['width'])) | |
| px_per_point_y = image.height / max(0.01, float(image_object['height'])) | |
| image_x = float(image_object['x']) | |
| image_y = float(image_object['y']) | |
| source_rectangles: list[tuple[int, int, int, int]] = [] | |
| for _edit, original in edits: | |
| pad_x = max(1.0, float(original['height']) * px_per_point_x * 0.1) | |
| pad_y = max(1.0, float(original['height']) * px_per_point_y * 0.1) | |
| x0 = max(0, math.floor((float(original['x']) - image_x) * px_per_point_x - pad_x)) | |
| y0 = max(0, math.floor((float(original['y']) - image_y) * px_per_point_y - pad_y)) | |
| x1 = min(image.width, math.ceil((float(original['x']) + float(original['width']) - image_x) * px_per_point_x + pad_x)) | |
| y1 = min(image.height, math.ceil((float(original['y']) + float(original['height']) - image_y) * px_per_point_y + pad_y)) | |
| if x1 <= x0 or y1 <= y0: | |
| continue | |
| source_rectangles.append((x0, y0, x1, y1)) | |
| if source_rectangles: | |
| try: | |
| import cv2 | |
| import numpy as np | |
| rgb = np.asarray(image.convert('RGB')) | |
| mask = np.zeros((image.height, image.width), dtype=np.uint8) | |
| for x0, y0, x1, y1 in source_rectangles: | |
| cv2.rectangle(mask, (x0, y0), (x1 - 1, y1 - 1), 255, thickness=-1) | |
| radius = max(2, min(18, round(max((y1 - y0 for _, y0, _, y1 in source_rectangles)) * 0.3))) | |
| repaired = cv2.inpaint(rgb, mask, radius, cv2.INPAINT_TELEA) | |
| alpha = image.getchannel('A') | |
| image = Image.fromarray(repaired).convert('RGBA') | |
| image.putalpha(alpha) | |
| except Exception: | |
| draw = ImageDraw.Draw(image) | |
| for x0, y0, x1, y1 in source_rectangles: | |
| strip = image.crop((x0, max(0, y0 - 2), x1, y0)).convert('RGB') | |
| pixels = list(strip.getdata()) | |
| if not pixels: | |
| strip = image.crop((x0, min(image.height - 1, y1), x1, min(image.height, y1 + 2))).convert('RGB') | |
| pixels = list(strip.getdata()) | |
| fill = tuple((sorted(channel)[len(channel) // 2] for channel in zip(*pixels))) if pixels else (255, 255, 255) | |
| draw.rectangle((x0, y0, x1, y1), fill=(*fill, 255)) | |
| painter = ImageDraw.Draw(image) | |
| for edit, original in edits: | |
| if edit.deleted or not edit.text.strip(): | |
| continue | |
| if edit.x < image_x - 0.01 or edit.y < image_y - 0.01 or edit.x + edit.width > image_x + float(image_object['width']) + 0.01 or (edit.y + edit.height > image_y + float(image_object['height']) + 0.01): | |
| raise HTTPException(status_code=409, detail='Background text must stay inside its background image') | |
| font_name, font_path = insertion_font(edit, original, folder) | |
| if font_path is None: | |
| font_path = font_file_for_family('DejaVu Sans', edit.bold, edit.italic, edit.text) | |
| try: | |
| font = ImageFont.truetype(str(font_path), max(1, round(edit.fontSize * px_per_point_y))) if font_path else ImageFont.load_default() | |
| except Exception: | |
| font = ImageFont.load_default() | |
| x = round((edit.x - image_x) * px_per_point_x) | |
| y = round((edit.y - image_y) * px_per_point_y) | |
| color = tuple((int(edit.color[index:index + 2], 16) for index in (1, 3, 5))) | |
| angle = normalized_rotation(edit.rotation) | |
| if abs(angle) < 0.0005: | |
| painter.text((x, y), edit.text, font=font, fill=(*color, 255), anchor='lt', spacing=max(1, round(edit.fontSize * px_per_point_y * 0.15))) | |
| else: | |
| bbox = painter.textbbox((0, 0), edit.text, font=font, anchor='lt') | |
| text_width = max(1, bbox[2] - bbox[0]) | |
| text_height = max(1, bbox[3] - bbox[1]) | |
| pad = max(8, round(text_height * 0.35)) | |
| layer = Image.new('RGBA', (text_width + pad * 2, text_height + pad * 2), (0, 0, 0, 0)) | |
| layer_draw = ImageDraw.Draw(layer) | |
| layer_draw.text((pad, pad), edit.text, font=font, fill=(*color, 255), anchor='lt') | |
| layer = layer.rotate(-angle, resample=Image.Resampling.BICUBIC, expand=True) | |
| center_x = round((edit.x + edit.width / 2 - image_x) * px_per_point_x) | |
| center_y = round((edit.y + edit.height / 2 - image_y) * px_per_point_y) | |
| image.alpha_composite(layer, (center_x - layer.width // 2, center_y - layer.height // 2)) | |
| output = BytesIO() | |
| image.save(output, format='PNG', optimize=True) | |
| return output.getvalue() | |
| def apply_editor_changes(pdf: fitz.Document, folder: Path, source_model: dict[str, Any], request: ExportRequest, page_filter: int | None=None) -> None: | |
| source_by_id = {item['id']: item for item in source_model['objects']} | |
| requested_by_id = {item.id: item for item in request.objects} | |
| requested_source_ids = {item.id for item in request.objects if item.id in source_by_id and (not item.isNew)} | |
| inserted_objects = [item for item in request.objects if item.id not in source_by_id] | |
| if len(requested_by_id) != len(request.objects) or requested_source_ids != set(source_by_id) or any((not item.isNew for item in inserted_objects)) or any((item.isNew for item in request.objects if item.id in source_by_id)): | |
| raise HTTPException(status_code=400, detail='Editor objects do not match this document') | |
| table_groups: list[dict[str, Any]] = [] | |
| seen_groups: set[tuple[int, str]] = set() | |
| for edit in request.tableGroups: | |
| if edit.page >= len(source_model['pages']): | |
| raise HTTPException(status_code=400, detail='Moved table is outside the document') | |
| key = (edit.page, edit.id) | |
| if key in seen_groups: | |
| raise HTTPException(status_code=400, detail='Duplicate table movement') | |
| seen_groups.add(key) | |
| region = next((entry for entry in source_model['pages'][edit.page].get('regions', []) if entry.get('id') == edit.id and entry.get('tableStructure')), None) | |
| if region is None: | |
| raise HTTPException(status_code=400, detail='The selected table is no longer available; analyze it again') | |
| original_rect = fitz.Rect(region['x'], region['y'], region['x'] + region['width'], region['y'] + region['height']) | |
| if abs(edit.width - original_rect.width) > 0.05 or abs(edit.height - original_rect.height) > 0.05 or edit.x + edit.width > source_model['pages'][edit.page]['width'] + 1 or (edit.y + edit.height > source_model['pages'][edit.page]['height'] + 1): | |
| raise HTTPException(status_code=400, detail='Tables can be moved within the page without resizing') | |
| table_groups.append({'edit': edit, 'region': region, 'sourceRect': original_rect, 'dx': edit.x - original_rect.x0, 'dy': edit.y - original_rect.y0}) | |
| def table_group_for(source_object: dict[str, Any]) -> dict[str, Any] | None: | |
| page_index = int(source_object['page']) | |
| center_x = float(source_object['x']) + float(source_object['width']) / 2 | |
| center_y = float(source_object['y']) + float(source_object['height']) / 2 | |
| candidates = [] | |
| for group in table_groups: | |
| edit = group['edit'] | |
| if edit.page != page_index: | |
| continue | |
| region_id = group['region']['id'] | |
| if (source_object.get('tableCell') or {}).get('regionId') == region_id or source_object.get('regionId') == region_id or group['sourceRect'].contains(fitz.Point(center_x, center_y)): | |
| candidates.append(group) | |
| return min(candidates, key=lambda item: item['sourceRect'].width * item['sourceRect'].height) if candidates else None | |
| replacements: list[EditObject] = [] | |
| image_edits: dict[str, EditObject] = {} | |
| image_fallbacks: list[EditObject] = [] | |
| grouped_text_edits: list[tuple[EditObject, dict[str, Any], fitz.Rect, bool]] = [] | |
| grouped_image_edits: list[tuple[EditObject, dict[str, Any]]] = [] | |
| background_raster_text_edits: dict[str, list[tuple[EditObject, dict[str, Any]]]] = {} | |
| new_texts: list[EditObject] = [] | |
| new_images: list[EditObject] = [] | |
| exact_move_clips: list[tuple[EditObject, fitz.Rect, fitz.Rect]] = [] | |
| annotated_pages: set[int] = set() | |
| for obj_id, edit in requested_by_id.items(): | |
| if obj_id not in source_by_id: | |
| if edit.page >= len(source_model['pages']): | |
| raise HTTPException(status_code=400, detail='Inserted object is outside the document') | |
| page_size = source_model['pages'][edit.page] | |
| if edit.x + edit.width > page_size['width'] + 1 or edit.y + edit.height > page_size['height'] + 1: | |
| raise HTTPException(status_code=400, detail='Inserted object is outside the document') | |
| if not edit.deleted: | |
| (new_images if edit.type == 'image' else new_texts).append(edit) | |
| continue | |
| original = source_by_id[obj_id] | |
| if edit.page != original['page'] or edit.page >= len(source_model['pages']): | |
| raise HTTPException(status_code=400, detail='Edited object is outside the document') | |
| page_size = source_model['pages'][edit.page] | |
| if edit.x > page_size['width'] or edit.y > page_size['height'] or edit.x + edit.width > page_size['width'] + 1 or (edit.y + edit.height > page_size['height'] + 1): | |
| raise HTTPException(status_code=400, detail='Edited object is outside the document') | |
| if edit.type != original.get('type', 'text'): | |
| raise HTTPException(status_code=400, detail='Object type cannot be changed') | |
| if page_filter is not None and edit.page != page_filter: | |
| continue | |
| group = table_group_for(original) | |
| expected_x = float(original['x']) + (group['dx'] if group else 0.0) | |
| expected_y = float(original['y']) + (group['dy'] if group else 0.0) | |
| geometry_changed = abs(edit.x - expected_x) > 0.01 or abs(edit.y - expected_y) > 0.01 or abs(edit.width - original['width']) > 0.01 or (abs(edit.height - original['height']) > 0.01) | |
| rotation_changed = edit.type == 'text' and abs(normalized_rotation(edit.rotation) - normalized_rotation(original.get('rotation', 0))) > 0.01 | |
| text_style_changed = edit.type == 'text' and (edit.text != original['text'] or edit.color != original['color'] or edit.bold != original['bold'] or (edit.italic != original['italic']) or (abs(edit.fontSize - original['fontSize']) > 0.01) or (edit.fontFamily != original.get('fontFamily', 'original')) or rotation_changed) | |
| changed = edit.deleted or geometry_changed or text_style_changed | |
| if changed: | |
| if edit.type == 'image': | |
| if group: | |
| grouped_image_edits.append((edit, group)) | |
| continue | |
| image_edits[obj_id] = edit | |
| continue | |
| if original.get('isBackgroundText') and original.get('source') == 'ocr': | |
| image_id = str(original.get('backgroundImageId', '')) | |
| background_image = source_by_id.get(image_id) | |
| if not background_image or background_image.get('type') != 'image' or (not background_image.get('isBackground')) or (background_image.get('page') != edit.page) or (int(background_image.get('imageXref', 0) or 0) <= 0): | |
| raise HTTPException(status_code=409, detail='This scanned background text has no independently editable image layer') | |
| background_raster_text_edits.setdefault(image_id, []).append((edit, original)) | |
| continue | |
| if original.get('isBackgroundText') and original.get('source') == 'native': | |
| raise HTTPException(status_code=409, detail='Covered native PDF text cannot be safely rewritten as a separate layer. Scanned background text and background images can be edited independently.') | |
| if group: | |
| rect = fitz.Rect(original['x'] + group['dx'] - 0.6, original['y'] + group['dy'] - 0.6, original['x'] + group['dx'] + original['width'] + 0.6, original['y'] + group['dy'] + original['height'] + 0.6) | |
| preserve_shape = geometry_changed and (not text_style_changed) and (not edit.deleted) and (abs(edit.width - original['width']) <= 0.01) and (abs(edit.height - original['height']) <= 0.01) | |
| grouped_text_edits.append((edit, group, rect, preserve_shape)) | |
| continue | |
| rect = fitz.Rect(original['x'] - 0.6, original['y'] - 0.6, original['x'] + original['width'] + 0.6, original['y'] + original['height'] + 0.6) | |
| background = sample_background_color(pdf[edit.page], rect) | |
| pdf[edit.page].add_redact_annot(rect, fill=background, cross_out=False) | |
| annotated_pages.add(edit.page) | |
| preserve_shape = geometry_changed and (not text_style_changed) and (not edit.deleted) and (abs(edit.width - original['width']) <= 0.01) and (abs(edit.height - original['height']) <= 0.01) | |
| if preserve_shape: | |
| source_rect = fitz.Rect(original['x'] - 0.1, original['y'] - 0.1, original['x'] + original['width'] + 0.1, original['y'] + original['height'] + 0.1) | |
| target_rect = fitz.Rect(edit.x - 0.1, edit.y - 0.1, edit.x + edit.width + 0.1, edit.y + edit.height + 0.1) | |
| exact_move_clips.append((edit, source_rect, target_rect)) | |
| elif not edit.deleted and edit.text: | |
| replacements.append(edit) | |
| for group in table_groups: | |
| edit, source_rect = (group['edit'], group['sourceRect']) | |
| if page_filter is not None and edit.page != page_filter: | |
| continue | |
| if abs(group['dx']) < 0.01 and abs(group['dy']) < 0.01: | |
| continue | |
| background = sample_background_color(pdf[edit.page], source_rect) | |
| pdf[edit.page].add_redact_annot(source_rect, fill=background, cross_out=False) | |
| annotated_pages.add(edit.page) | |
| direct_image_xrefs: set[int] = set() | |
| background_image_overrides: dict[str, bytes] = {} | |
| for obj_id, edit in list(image_edits.items()): | |
| original = source_by_id[obj_id] | |
| xref = int(original.get('imageXref', 0) or 0) | |
| if xref <= 0: | |
| rect = fitz.Rect(original['x'] - 0.6, original['y'] - 0.6, original['x'] + original['width'] + 0.6, original['y'] + original['height'] + 0.6) | |
| background = sample_background_color(pdf[edit.page], rect) | |
| pdf[edit.page].add_redact_annot(rect, fill=background, cross_out=False) | |
| annotated_pages.add(edit.page) | |
| if not edit.deleted: | |
| image_fallbacks.append(edit) | |
| continue | |
| if xref in direct_image_xrefs: | |
| continue | |
| try: | |
| pdf[edit.page].delete_image(xref) | |
| direct_image_xrefs.add(xref) | |
| except Exception: | |
| rect = fitz.Rect(original['x'] - 0.6, original['y'] - 0.6, original['x'] + original['width'] + 0.6, original['y'] + original['height'] + 0.6) | |
| background = sample_background_color(pdf[edit.page], rect) | |
| pdf[edit.page].add_redact_annot(rect, fill=background, cross_out=False) | |
| annotated_pages.add(edit.page) | |
| if not edit.deleted: | |
| image_fallbacks.append(edit) | |
| for image_id, edits in background_raster_text_edits.items(): | |
| image_object = source_by_id[image_id] | |
| background_image_overrides[image_id] = patch_background_raster_text(folder, image_object, edits, source_by_id) | |
| xref = int(image_object.get('imageXref', 0) or 0) | |
| if xref not in direct_image_xrefs: | |
| try: | |
| pdf[image_object['page']].delete_image(xref) | |
| direct_image_xrefs.add(xref) | |
| except Exception as exc: | |
| raise HTTPException(status_code=409, detail='The background image could not be isolated from foreground content') from exc | |
| for page_index in annotated_pages: | |
| pdf[page_index].apply_redactions(images=fitz.PDF_REDACT_IMAGE_PIXELS) | |
| moved_table_groups = [group for group in table_groups if (page_filter is None or group['edit'].page == page_filter) and (abs(group['dx']) >= 0.01 or abs(group['dy']) >= 0.01)] | |
| grouped_pages: set[int] = set() | |
| if moved_table_groups: | |
| source_pdf = fitz.open(folder / 'original.pdf') | |
| try: | |
| for group in moved_table_groups: | |
| edit = group['edit'] | |
| target_rect = fitz.Rect(edit.x, edit.y, edit.x + edit.width, edit.y + edit.height) | |
| pdf[edit.page].show_pdf_page(target_rect, source_pdf, edit.page, clip=group['sourceRect'], keep_proportion=False, overlay=True) | |
| grouped_pages.add(edit.page) | |
| finally: | |
| source_pdf.close() | |
| for edit, group, rect, preserve_shape in grouped_text_edits: | |
| background = sample_background_color(pdf[edit.page], rect) | |
| pdf[edit.page].add_redact_annot(rect, fill=background, cross_out=False) | |
| grouped_pages.add(edit.page) | |
| if preserve_shape: | |
| original = source_by_id[edit.id] | |
| source_rect = fitz.Rect(original['x'] - 0.1, original['y'] - 0.1, original['x'] + original['width'] + 0.1, original['y'] + original['height'] + 0.1) | |
| target_rect = fitz.Rect(edit.x - 0.1, edit.y - 0.1, edit.x + edit.width + 0.1, edit.y + edit.height + 0.1) | |
| exact_move_clips.append((edit, source_rect, target_rect)) | |
| elif not edit.deleted and edit.text: | |
| replacements.append(edit) | |
| for edit, group in grouped_image_edits: | |
| original = source_by_id[edit.id] | |
| rect = fitz.Rect(original['x'] + group['dx'] - 0.6, original['y'] + group['dy'] - 0.6, original['x'] + group['dx'] + original['width'] + 0.6, original['y'] + group['dy'] + original['height'] + 0.6) | |
| background = sample_background_color(pdf[edit.page], rect) | |
| pdf[edit.page].add_redact_annot(rect, fill=background, cross_out=False) | |
| grouped_pages.add(edit.page) | |
| if not edit.deleted: | |
| image_fallbacks.append(edit) | |
| for page_index in grouped_pages: | |
| pdf[page_index].apply_redactions(images=fitz.PDF_REDACT_IMAGE_PIXELS) | |
| if exact_move_clips: | |
| source_pdf = fitz.open(folder / 'original.pdf') | |
| try: | |
| for edit, source_rect, target_rect in exact_move_clips: | |
| pdf[edit.page].show_pdf_page(target_rect, source_pdf, edit.page, clip=source_rect, keep_proportion=False, overlay=True) | |
| finally: | |
| source_pdf.close() | |
| if direct_image_xrefs: | |
| for sibling in source_model['objects']: | |
| xref = int(sibling.get('imageXref', 0) or 0) | |
| if sibling.get('type') != 'image' or xref not in direct_image_xrefs: | |
| continue | |
| sibling_edit = requested_by_id[sibling['id']] | |
| edit_is_in_scope = page_filter is None or sibling['page'] == page_filter | |
| changed_image = sibling['id'] in image_edits and edit_is_in_scope | |
| if changed_image and sibling_edit.deleted: | |
| continue | |
| image_id = sibling.get('assetId', '') | |
| extension = sibling.get('assetExt', '') | |
| if not re.fullmatch('[0-9a-f]{32}', image_id) or not re.fullmatch('[a-z0-9]{1,8}', extension): | |
| raise HTTPException(status_code=400, detail='The source image asset is invalid') | |
| image_path = folder / 'images' / f'{image_id}.{extension}' | |
| if not image_path.is_file(): | |
| raise HTTPException(status_code=400, detail='The source image asset is missing') | |
| if changed_image: | |
| rect = fitz.Rect(sibling_edit.x, sibling_edit.y, sibling_edit.x + sibling_edit.width, sibling_edit.y + sibling_edit.height) | |
| else: | |
| rect = fitz.Rect(sibling['x'], sibling['y'], sibling['x'] + sibling['width'], sibling['y'] + sibling['height']) | |
| if sibling['id'] in background_image_overrides and edit_is_in_scope: | |
| pdf[sibling['page']].insert_image(rect, stream=background_image_overrides[sibling['id']], keep_proportion=False, overlay=False) | |
| else: | |
| pdf[sibling['page']].insert_image(rect, filename=str(image_path), keep_proportion=False, overlay=not bool(sibling.get('isBackground'))) | |
| for edit in replacements: | |
| page = pdf[edit.page] | |
| font_name, font_file = insertion_font(edit, source_by_id[edit.id], folder) | |
| color = tuple((int(edit.color[i:i + 2], 16) / 255 for i in (1, 3, 5))) | |
| insert_editor_text(page, edit, font_name, font_file, color, overlay=True) | |
| for edit in image_fallbacks: | |
| original = source_by_id[edit.id] | |
| asset_id = original.get('assetId', '') | |
| extension = original.get('assetExt', '') | |
| if not re.fullmatch('[0-9a-f]{32}', asset_id) or not re.fullmatch('[a-z0-9]{1,8}', extension): | |
| raise HTTPException(status_code=400, detail='The source image asset is invalid') | |
| image_path = folder / 'images' / f'{asset_id}.{extension}' | |
| if not image_path.is_file(): | |
| raise HTTPException(status_code=400, detail='The source image asset is missing') | |
| image_rect = fitz.Rect(edit.x, edit.y, edit.x + edit.width, edit.y + edit.height) | |
| pdf[edit.page].insert_image(image_rect, filename=str(image_path), keep_proportion=False, overlay=not bool(original.get('isBackground'))) | |
| for edit in new_texts: | |
| font_name, font_file = insertion_font(edit, {}, folder) | |
| color = tuple((int(edit.color[index:index + 2], 16) / 255 for index in (1, 3, 5))) | |
| insert_editor_text(pdf[edit.page], edit, font_name, font_file, color, overlay=not edit.isBackground) | |
| for edit in new_images: | |
| asset_id, extension = (str(edit.assetId or ''), str(edit.assetExt or '')) | |
| if not re.fullmatch('[0-9a-f]{32}', asset_id) or not re.fullmatch('png|jpe?g|webp', extension): | |
| raise HTTPException(status_code=400, detail='The inserted image asset is invalid') | |
| image_path = folder / 'images' / f'{asset_id}.{extension}' | |
| if not image_path.is_file(): | |
| raise HTTPException(status_code=400, detail='The inserted image asset is missing') | |
| pdf[edit.page].insert_image(fitz.Rect(edit.x, edit.y, edit.x + edit.width, edit.y + edit.height), filename=str(image_path), keep_proportion=False, overlay=not edit.isBackground) | |
| def export_pdf(document_id: str, request: ExportRequest) -> Path: | |
| folder = document_dir(document_id) | |
| model_path = folder / 'document.json' | |
| source_model = json.loads(model_path.read_text(encoding='utf-8')) | |
| pdf = fitz.open(folder / 'original.pdf') | |
| try: | |
| apply_editor_changes(pdf, folder, source_model, request) | |
| output_path = folder / 'edited.pdf' | |
| pdf.save(output_path, garbage=4, deflate=True) | |
| finally: | |
| pdf.close() | |
| return FileResponse(output_path, media_type='application/pdf', filename=f'{safe_stem}-edited.pdf') | |