"""PDF 유효성 검증 모듈 — GPT 비전으로 생성된 PDF의 품질을 검증.""" import base64 import json import socket import time import urllib.error import urllib.request from pathlib import Path from typing import Any, Dict from app.core.config import settings PROMPTS_DIR = ( Path(__file__).resolve().parent.parent.parent.parent / "prompts" / "_base" / "pdf_validation" / "v1" ) from app.core.config import settings as _settings VALIDATION_SCHEMA = { "type": "object", "additionalProperties": False, "properties": { "text_readable": {"type": "boolean"}, "images_present": {"type": "boolean"}, "layout_correct": {"type": "boolean"}, "caption_visible": {"type": "boolean"}, "overall_quality": { "type": "integer", "minimum": 1, "maximum": 10, }, "issues": { "type": "array", "items": {"type": "string"}, }, }, "required": [ "text_readable", "images_present", "layout_correct", "caption_visible", "overall_quality", "issues", ], } def _load_prompt(filename: str) -> str: return (PROMPTS_DIR / filename).read_text(encoding="utf-8").strip() def _pdf_first_page_to_png(pdf_path: Path) -> bytes: """Convert the first page of a PDF to a PNG image. Uses fpdf2 to render a summary page since pypdf cannot rasterize. Falls back to reading the raw PDF bytes for vision API. """ # Try using subprocess with sips (macOS) or pdftoppm import subprocess import tempfile with tempfile.TemporaryDirectory() as tmpdir: out_path = Path(tmpdir) / "page" # Try pdftoppm first (poppler) try: subprocess.run( [ "pdftoppm", "-png", "-f", "1", "-l", "1", "-r", "150", str(pdf_path), str(out_path), ], check=True, capture_output=True, timeout=30, ) # pdftoppm outputs as page-1.png or page-01.png for candidate in out_path.parent.iterdir(): if candidate.suffix == ".png": return candidate.read_bytes() except (subprocess.CalledProcessError, FileNotFoundError): pass # Fallback: use sips on macOS to convert PDF to PNG try: png_out = Path(tmpdir) / "page.png" subprocess.run( [ "sips", "-s", "format", "png", "--resampleHeight", "2000", str(pdf_path), "--out", str(png_out), ], check=True, capture_output=True, timeout=30, ) if png_out.exists(): return png_out.read_bytes() except (subprocess.CalledProcessError, FileNotFoundError): pass raise RuntimeError( "Could not convert PDF to image. Install poppler (pdftoppm) " "or run on macOS with sips available." ) def _call_vision_api( image_bytes: bytes, system_prompt: str, ) -> Dict[str, Any]: """Call OpenAI vision API with an image for PDF validation.""" api_key = settings.openai_api_key if not api_key: raise RuntimeError("OpenAI API key is not configured.") image_b64 = base64.b64encode(image_bytes).decode("ascii") body = { "model": settings.openai_model, "instructions": system_prompt, "input": [ { "role": "user", "content": [ { "type": "input_image", "image_url": f"data:image/png;base64,{image_b64}", }, { "type": "input_text", "text": "Validate this webbook PDF page.", }, ], }, ], "temperature": 0.1, "store": False, "text": { "format": { "type": "json_schema", "name": "pdf_validation_result", "strict": True, "schema": VALIDATION_SCHEMA, } }, } request = urllib.request.Request( "https://api.openai.com/v1/responses", data=json.dumps(body).encode("utf-8"), headers={ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", }, method="POST", ) timeout = _settings.llm_timeout_validation max_retries = _settings.llm_max_retries last_error: Exception | None = None for attempt in range(1, max_retries + 2): # 첫 시도 + max_retries 재시도 try: with urllib.request.urlopen(request, timeout=timeout) as resp: payload = json.loads(resp.read().decode("utf-8")) break except urllib.error.HTTPError as exc: error_text = exc.read().decode("utf-8", errors="replace") last_error = RuntimeError(f"OpenAI API error {exc.code}: {error_text}") if exc.code in {429, 500, 502, 503, 504} and attempt <= max_retries: time.sleep(2 * attempt) continue raise last_error from exc except (urllib.error.URLError, socket.timeout) as exc: last_error = exc if attempt <= max_retries: time.sleep(2 * attempt) continue raise RuntimeError(f"OpenAI API failed after {max_retries} retries: {exc}") from exc else: raise RuntimeError(f"OpenAI API failed: {last_error}") # Extract text from response output_text = payload.get("output_text") if not isinstance(output_text, str) or not output_text.strip(): output = payload.get("output") if isinstance(output, list): for item in output: if not isinstance(item, dict): continue content = item.get("content") if not isinstance(content, list): continue for part in content: if isinstance(part, dict) and part.get("type") == "output_text": text = part.get("text") if isinstance(text, str) and text.strip(): output_text = text break if not output_text: raise RuntimeError("OpenAI vision response did not include output text.") return json.loads(output_text) class PDFValidator: """PDF 품질 검증기 — GPT 비전 기반.""" def validate(self, pdf_path: Path) -> Dict[str, Any]: """Validate a generated PDF using GPT vision. Converts the first page to an image, sends it to GPT for layout and quality analysis. Args: pdf_path: Path to the PDF file to validate. Returns: Validation result dict with text_readable, images_present, layout_correct, caption_visible, overall_quality, issues. """ if not pdf_path.exists(): raise FileNotFoundError(f"PDF not found: {pdf_path}") system_prompt = _load_prompt("validate.md") image_bytes = _pdf_first_page_to_png(pdf_path) result = _call_vision_api(image_bytes, system_prompt) # Add metadata result["pdf_path"] = str(pdf_path) result["pdf_filename"] = pdf_path.name return result def validate_batch( self, pdf_paths: list[Path] ) -> list[Dict[str, Any]]: """Validate multiple PDFs. Args: pdf_paths: List of PDF file paths. Returns: List of validation results. """ results = [] for pdf_path in pdf_paths: try: result = self.validate(pdf_path) results.append(result) except Exception as exc: results.append( { "pdf_path": str(pdf_path), "pdf_filename": pdf_path.name, "error": str(exc), "text_readable": False, "images_present": False, "layout_correct": False, "caption_visible": False, "overall_quality": 0, "issues": [f"Validation failed: {exc}"], } ) return results