#!/usr/bin/env python3 """Local-only PDF text extraction with OCR fallback and review separation.""" from __future__ import annotations import argparse import hashlib import json import os import re import sqlite3 import subprocess import tempfile from datetime import datetime, timezone from pathlib import Path from dashboard_v5.document_originals import configured_original_roots from dashboard_v5.sprint6f_b_schema import apply_pipeline_schema MIN_USEFUL_CHARS = 80 MAX_DOCUMENT_BYTES = 20 * 1024 * 1024 MAX_SECTION_CHARS = 4000 def file_hash(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(1024 * 1024), b""): digest.update(block) return digest.hexdigest() def command_version(command: list[str]) -> str: try: result = subprocess.run(command, capture_output=True, text=True, timeout=15, check=False) line = (result.stdout or result.stderr).splitlines()[0] return line[:120] except Exception: return "version unavailable" def safe_original(value: object, roots: tuple[Path, ...]) -> Path | None: if not isinstance(value, str) or not value: return None candidate = Path(value).absolute() if candidate.is_symlink(): return None matched_root = next( (root for root in roots if candidate == root or root in candidate.parents), None ) if ( matched_root is None or matched_root.is_symlink() or matched_root.resolve() != matched_root ): return None current = matched_root for part in candidate.relative_to(matched_root).parts: current = current / part if current.is_symlink(): return None try: resolved = candidate.resolve(strict=True) except (OSError, RuntimeError): return None if not resolved.is_file() or resolved.is_symlink() or resolved.stat().st_size > MAX_DOCUMENT_BYTES: return None if not any(resolved == root or root in resolved.parents for root in roots): return None return resolved def useful(text: str) -> bool: compact = " ".join(text.split()) if len(compact) < MIN_USEFUL_CHARS: return False alnum = sum(char.isalnum() for char in compact) return alnum / max(1, len(compact)) >= 0.45 def pdftotext(path: Path) -> str: result = subprocess.run(["pdftotext", "-layout", str(path), "-"], capture_output=True, text=True, timeout=90, check=False) return result.stdout if result.returncode == 0 else "" def extract_pdf(path: Path) -> tuple[str, str, str, float | None]: text = pdftotext(path) if useful(text): return text, "text_layer", "pdftotext", 1.0 with tempfile.TemporaryDirectory(prefix="health-ocr-") as directory: temp = Path(directory) os.chmod(temp, 0o700) output = temp / "ocr.pdf" result = subprocess.run( ["ocrmypdf", "--skip-text", "--deskew", "--clean", "--language", "deu+eng", str(path), str(output)], capture_output=True, text=True, timeout=900, check=False, ) if result.returncode not in {0, 6} or not output.exists(): return "", "failed", "ocrmypdf+tesseract", None text = pdftotext(output) if not useful(text): return text, "no_text", "ocrmypdf+tesseract", 0.0 compact = " ".join(text.split()) confidence = round(min(0.95, max(0.4, sum(char.isalnum() for char in compact) / max(1, len(compact)))), 3) return text, "ocr", "ocrmypdf+tesseract", confidence def normalize_section(text: str) -> str: return " ".join(re.sub(r"[^\w]+", " ", text.casefold(), flags=re.UNICODE).split()) def sections(text: str) -> list[tuple[int, int, str, str]]: result: list[tuple[int, int, str, str]] = [] for page_number, page in enumerate(text.split("\f"), 1): paragraphs = [" ".join(part.split()) for part in re.split(r"\n\s*\n", page) if useful(part)] section_number = 0 for paragraph in paragraphs: for start in range(0, len(paragraph), MAX_SECTION_CHARS): chunk = paragraph[start : start + MAX_SECTION_CHARS] normalized = normalize_section(chunk) if len(normalized) < 40: continue section_number += 1 digest = hashlib.sha256(normalized.encode()).hexdigest() result.append((page_number, section_number, chunk, digest)) return result def run_pipeline(database: Path, max_documents: int, report_path: Path) -> dict[str, int]: roots = tuple(root.resolve() for root in configured_original_roots()) connection = sqlite3.connect(database) connection.row_factory = sqlite3.Row apply_pipeline_schema(connection) rows = connection.execute( "SELECT id,local_original_path FROM dokumente WHERE local_original_path IS NOT NULL ORDER BY COALESCE(document_date,upload_datum),id LIMIT ?", (max_documents,), ).fetchall() counters = {"registered": len(rows), "text_layer": 0, "ocr": 0, "no_text": 0, "failed": 0, "skipped_existing": 0, "unsafe_or_missing": 0, "sections": 0, "repeated_sections": 0} versions = { "pdftotext": command_version(["pdftotext", "-v"]), "ocrmypdf": command_version(["ocrmypdf", "--version"]), "tesseract": command_version(["tesseract", "--version"]), } for row in rows: path = safe_original(row["local_original_path"], roots) if path is None or path.suffix.casefold() != ".pdf": counters["unsafe_or_missing"] += 1 continue digest = file_hash(path) if connection.execute("SELECT 1 FROM document_extractions WHERE document_id=? AND source_hash=?", (row["id"], digest)).fetchone(): counters["skipped_existing"] += 1 continue text, status, engine, confidence = extract_pdf(path) counters[status] += 1 now = datetime.now(timezone.utc).isoformat(timespec="seconds") cursor = connection.execute( """INSERT INTO document_extractions (document_id,source_hash,extraction_status,extraction_engine,engine_version,extraction_confidence,review_status,canonical_data_status,extracted_at) VALUES(?,?,?,?,?,?,?,?,?)""", (row["id"], digest, status, engine, versions.get(engine.split("+")[0], versions.get("ocrmypdf", "unknown")), confidence, "unreviewed", "not_promoted", now), ) extraction_id = int(cursor.lastrowid) previous_for_document = bool(connection.execute( "SELECT 1 FROM document_extractions WHERE document_id=? AND id<>? LIMIT 1", (row["id"], extraction_id) ).fetchone()) connection.execute("DELETE FROM health_document_machine_fts WHERE document_id=?", (row["id"],)) for page_number, section_number, chunk, section_hash in sections(text): repeated = bool(connection.execute( "SELECT 1 FROM document_extraction_sections WHERE normalized_section_hash=? LIMIT 1", (section_hash,) ).fetchone()) repetition_status = "repeated_from_earlier" if repeated else ("new_or_changed" if previous_for_document else "first_documented") connection.execute( """INSERT INTO document_extraction_sections (extraction_id,page_number,section_number,text_content,bounding_box_json,engine,confidence,source_hash,normalized_section_hash,repetition_status,created_at) VALUES(?,?,?,?,?,?,?,?,?,?,?)""", (extraction_id, page_number, section_number, chunk, None, engine, confidence, digest, section_hash, repetition_status, now), ) connection.execute( "INSERT INTO health_document_machine_fts(document_id,page_number,section_number,content,review_label) VALUES(?,?,?,?,?)", (row["id"], page_number, section_number, chunk, "Maschinell extrahiert · ungeprüft"), ) counters["sections"] += 1 counters["repeated_sections"] += int(repeated) connection.commit() connection.close() report = {"status": "completed", "engines": versions, "counters": counters, "review_rule": "extraction never changes review_status or canonical laboratory data"} report_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n") report_path.chmod(0o600) return counters def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--db", required=True, type=Path) parser.add_argument("--max-documents", type=int, default=500) parser.add_argument("--report", required=True, type=Path) args = parser.parse_args() if not 1 <= args.max_documents <= 2000: parser.error("max documents out of range") print(json.dumps(run_pipeline(args.db.resolve(), args.max_documents, args.report.resolve()), sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())