Spaces:
Running
Running
| import os | |
| import re | |
| import tempfile | |
| import traceback | |
| import cv2 | |
| import numpy as np | |
| import uvicorn | |
| from io import BytesIO | |
| from typing import List, Dict, Any | |
| from PIL import Image | |
| from fastapi import FastAPI, File, UploadFile, HTTPException, Query | |
| from fastapi.middleware.cors import CORSMiddleware | |
| # -------------------- Writable env for PaddleX -------------------- | |
| # بنحاول نوجّه أي cache/resource paths لأماكن writable | |
| os.environ["OMP_NUM_THREADS"] = "1" | |
| os.environ["OPENBLAS_NUM_THREADS"] = "1" | |
| os.environ["MKL_NUM_THREADS"] = "1" | |
| os.environ["NUMEXPR_NUM_THREADS"] = "1" | |
| os.environ["VECLIB_MAXIMUM_THREADS"] = "1" | |
| os.environ["HOME"] = os.environ.get("HOME", "/home/user") or "/home/user" | |
| os.environ["XDG_CACHE_HOME"] = os.environ.get("XDG_CACHE_HOME", "/home/user/.cache") or "/home/user/.cache" | |
| os.environ["PADDLEX_HOME"] = os.environ.get("PADDLEX_HOME", "/home/user/.paddlex") or "/home/user/.paddlex" | |
| os.environ["TMPDIR"] = os.environ.get("TMPDIR", "/tmp") or "/tmp" | |
| # دعم ملفات PDF | |
| try: | |
| from pdf2image import convert_from_bytes | |
| PDF_AVAILABLE = True | |
| except ImportError: | |
| PDF_AVAILABLE = False | |
| # دعم النص العربي - إعادة الترتيب والـ BiDi | |
| try: | |
| import arabic_reshaper | |
| from bidi.algorithm import get_display | |
| ARABIC_RESHAPER_AVAILABLE = True | |
| except ImportError: | |
| ARABIC_RESHAPER_AVAILABLE = False | |
| print("Warning: arabic_reshaper or python-bidi not installed. Arabic text may appear reversed.") | |
| # -------------------- Global OCR Models -------------------- | |
| # PaddleOCR instances | |
| ocr_ar = None # Arabic OCR engine | |
| ocr_en = None # English OCR engine | |
| model_load_error = None | |
| # للتوافق مع /load-models response | |
| _loaded_detector_name = "PP-OCRv4 (built-in)" | |
| _loaded_recognizer_name = "arabic (lang=ar)" | |
| _loaded_en_recognizer_name = "en (lang=en)" | |
| # aliases للتوافق مع الكود القديم | |
| paddle_detector = None | |
| paddle_recognizer = None | |
| paddle_en_recognizer = None | |
| app = FastAPI(title="OCR Scan Vision API", version="1.2.0") | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=["*"], | |
| allow_credentials=True, | |
| allow_methods=["*"], | |
| allow_headers=["*"], | |
| ) | |
| def fix_arabic_text(text: str) -> str: | |
| """ | |
| إصلاح النص العربي القادم من OCR: | |
| - arabic_reshaper: يحوّل presentation forms لـ Unicode عربي معياري | |
| - بدون get_display: لأنها للـ rendering فقط وتعكس الكلمات في الـ storage | |
| """ | |
| if not text or not ARABIC_RESHAPER_AVAILABLE: | |
| return text | |
| try: | |
| config = arabic_reshaper.ArabicReshaper(configuration={ | |
| 'delete_tatweel': False, | |
| 'support_ligatures': True, # مهم لتركيب الحروف العربية صح | |
| 'RIAL SIGN': True, | |
| 'support_zwj': True, | |
| }) | |
| # reshape كلمة كلمة للحفاظ على المسافات بين الكلمات | |
| words = text.split(' ') | |
| reshaped_words = [config.reshape(w) for w in words] | |
| return ' '.join(reshaped_words) | |
| except Exception as e: | |
| print(f"Arabic reshaping error: {e}") | |
| return text | |
| # -------------------- Script Detection -------------------- | |
| def detect_script(text: str) -> str: | |
| """تحديد لغة النص: arabic أو latin أو mixed.""" | |
| if not text: | |
| return "unknown" | |
| arabic_chars = len(re.findall(r'[\u0600-\u06FF]', text)) | |
| latin_chars = len(re.findall(r'[a-zA-Z]', text)) | |
| if arabic_chars == 0 and latin_chars == 0: | |
| return "unknown" | |
| if arabic_chars >= latin_chars: | |
| return "arabic" | |
| return "latin" | |
| # -------------------- Arabic Text Cleaning -------------------- | |
| def smart_clean_arabic_text(text: str) -> str: | |
| if not text: | |
| return "" | |
| # إزالة التشكيل فقط | |
| text = re.sub(r"[\u064B-\u065F]", "", text) | |
| # الحفاظ على العربي والأرقام وعلامات الترقيم - حذف رموز غير مفيدة فقط | |
| text = re.sub(r"[^\u0600-\u06FF\u0750-\u077F\uFB50-\uFDFFa-zA-Z0-9\s\.,;:!?()\-/\u0660-\u0669]", "", text) | |
| # تنظيف المسافات الزائدة | |
| text = re.sub(r"\s+", " ", text) | |
| text = fix_arabic_text(text) | |
| return text.strip() | |
| # -------------------- Helpers -------------------- | |
| def get_versions_info() -> Dict[str, str]: | |
| info = {} | |
| try: | |
| import paddle | |
| info["paddle"] = getattr(paddle, "__version__", "unknown") | |
| except Exception as e: | |
| info["paddle"] = f"not available: {e}" | |
| # مهم: ما نعملش import paddlex هنا | |
| # لأن مجرد import كان بيفتح path fonts وبيكسر مع permission issues | |
| info["paddlex"] = "lazy import only" | |
| try: | |
| info["numpy"] = np.__version__ | |
| except Exception as e: | |
| info["numpy"] = f"not available: {e}" | |
| try: | |
| import cv2 as _cv2 | |
| info["opencv"] = getattr(_cv2, "__version__", "unknown") | |
| except Exception as e: | |
| info["opencv"] = f"not available: {e}" | |
| return info | |
| def validate_image_array(img: np.ndarray) -> None: | |
| if img is None: | |
| raise ValueError("Image is None") | |
| if not isinstance(img, np.ndarray): | |
| raise ValueError("Image is not a numpy array") | |
| if img.size == 0: | |
| raise ValueError("Image array is empty") | |
| if len(img.shape) != 3 or img.shape[2] != 3: | |
| raise ValueError(f"Expected BGR image with 3 channels, got shape={img.shape}") | |
| def ensure_runtime_dirs(): | |
| paths = [ | |
| os.environ.get("HOME", "/home/user"), | |
| os.environ.get("XDG_CACHE_HOME", "/home/user/.cache"), | |
| os.environ.get("PADDLEX_HOME", "/home/user/.paddlex"), | |
| "/tmp", | |
| ] | |
| for p in paths: | |
| if p: | |
| os.makedirs(p, exist_ok=True) | |
| def load_first_available_model(create_model_fn, role: str, candidates: List[str]): | |
| """Kept for compatibility but not used with PaddleOCR backend.""" | |
| raise RuntimeError(f"load_first_available_model not used with PaddleOCR backend") | |
| def save_temp_image(img: np.ndarray, suffix: str = ".png") -> str: | |
| validate_image_array(img) | |
| tmp = tempfile.NamedTemporaryFile(delete=False, suffix=suffix) | |
| tmp_path = tmp.name | |
| tmp.close() | |
| if not cv2.imwrite(tmp_path, img): | |
| raise RuntimeError(f"Failed to write temporary image: {tmp_path}") | |
| return tmp_path | |
| def get_models(): | |
| global ocr_ar, ocr_en, paddle_detector, paddle_recognizer, paddle_en_recognizer, model_load_error | |
| if ocr_ar is not None: | |
| return ocr_ar, ocr_ar, ocr_en # compat tuple | |
| if model_load_error is not None: | |
| raise HTTPException(status_code=500, detail=model_load_error) | |
| try: | |
| ensure_runtime_dirs() | |
| print("Loading PaddleOCR models...") | |
| from paddleocr import PaddleOCR | |
| import paddleocr as _paddleocr_module | |
| _paddleocr_version = tuple( | |
| int(x) for x in | |
| getattr(_paddleocr_module, "__version__", "2.0.0").split(".")[:2] | |
| ) | |
| _new_api = _paddleocr_version >= (2, 9) # API تغيّر من 2.9+ | |
| print(f"PaddleOCR version: {_paddleocr_version}, new_api={_new_api}") | |
| # Arabic OCR engine | |
| if _new_api: | |
| ocr_ar = PaddleOCR(lang='ar') | |
| else: | |
| paddlex_home = os.environ.get("PADDLEX_HOME", "/home/user/.paddlex") | |
| ocr_ar = PaddleOCR( | |
| lang='ar', | |
| use_angle_cls=True, | |
| use_gpu=False, | |
| show_log=False, | |
| det_model_dir=os.path.join(paddlex_home, "ar_det"), | |
| rec_model_dir=os.path.join(paddlex_home, "ar_rec"), | |
| cls_model_dir=os.path.join(paddlex_home, "cls"), | |
| ) | |
| # English OCR engine | |
| try: | |
| if _new_api: | |
| ocr_en = PaddleOCR(lang='en') | |
| else: | |
| paddlex_home = os.environ.get("PADDLEX_HOME", "/home/user/.paddlex") | |
| ocr_en = PaddleOCR( | |
| lang='en', | |
| use_angle_cls=False, | |
| use_gpu=False, | |
| show_log=False, | |
| det_model_dir=os.path.join(paddlex_home, "en_det"), | |
| rec_model_dir=os.path.join(paddlex_home, "en_rec"), | |
| ) | |
| except Exception as en_err: | |
| print(f"Warning: English OCR engine failed: {en_err}") | |
| ocr_en = None | |
| # compat aliases | |
| paddle_detector = ocr_ar | |
| paddle_recognizer = ocr_ar | |
| paddle_en_recognizer = ocr_en | |
| print("PaddleOCR models loaded successfully. ar={} en={}".format( | |
| ocr_ar is not None, ocr_en is not None | |
| )) | |
| except Exception as e: | |
| traceback.print_exc() | |
| ocr_ar = None | |
| ocr_en = None | |
| paddle_detector = None | |
| paddle_recognizer = None | |
| paddle_en_recognizer = None | |
| model_load_error = { | |
| "message": "OCR models failed to load", | |
| "error": str(e), | |
| "hint": "PaddleOCR init failed. Check paddleocr installation and model download.", | |
| "versions": get_versions_info(), | |
| } | |
| raise HTTPException(status_code=500, detail=model_load_error) | |
| return ocr_ar, ocr_ar, ocr_en | |
| def safe_result_get(result: Any, key: str, default: Any = None) -> Any: | |
| if result is None: | |
| return default | |
| if isinstance(result, dict): | |
| return result.get(key, default) | |
| if hasattr(result, key): | |
| try: | |
| value = getattr(result, key) | |
| return default if value is None else value | |
| except Exception: | |
| pass | |
| try: | |
| value = result[key] | |
| return default if value is None else value | |
| except Exception: | |
| return default | |
| def normalize_items(value: Any) -> List[Any]: | |
| if value is None: | |
| return [] | |
| if isinstance(value, np.ndarray): | |
| return value.tolist() | |
| if isinstance(value, (list, tuple)): | |
| return list(value) | |
| return [value] | |
| def describe_result_shape(result: Any) -> Dict[str, Any]: | |
| description = { | |
| "type": type(result).__name__, | |
| } | |
| if isinstance(result, dict): | |
| description["keys"] = list(result.keys())[:10] | |
| return description | |
| try: | |
| description["attrs"] = [name for name in dir(result) if not name.startswith("_")][:10] | |
| except Exception: | |
| description["attrs"] = [] | |
| return description | |
| def extract_boxes_from_detection_result(result: Any) -> List[Any]: | |
| if result is None: | |
| return [] | |
| if isinstance(result, (list, tuple)): | |
| collected_boxes = [] | |
| for item in result: | |
| collected_boxes.extend(extract_boxes_from_detection_result(item)) | |
| return collected_boxes | |
| for key in ["dt_polys", "polys", "boxes", "det_polys", "dt_boxes"]: | |
| boxes = safe_result_get(result, key) | |
| normalized = normalize_items(boxes) | |
| if normalized: | |
| return normalized | |
| for key in ["res", "result", "data", "output"]: | |
| nested = safe_result_get(result, key) | |
| if nested is None or nested is result: | |
| continue | |
| nested_boxes = extract_boxes_from_detection_result(nested) | |
| if nested_boxes: | |
| return nested_boxes | |
| return [] | |
| def extract_recognition_payload(result: Any) -> Dict[str, Any]: | |
| if result is None: | |
| return { | |
| "raw_text": "", | |
| "score": 0.0, | |
| } | |
| if isinstance(result, (list, tuple)): | |
| best_payload = { | |
| "raw_text": "", | |
| "score": 0.0, | |
| } | |
| for item in result: | |
| payload = extract_recognition_payload(item) | |
| if payload["score"] > best_payload["score"] or ( | |
| payload["raw_text"] and not best_payload["raw_text"] | |
| ): | |
| best_payload = payload | |
| return best_payload | |
| for key in ["res", "result", "data", "output"]: | |
| nested = safe_result_get(result, key) | |
| if nested is not None and nested is not result: | |
| nested_payload = extract_recognition_payload(nested) | |
| if nested_payload["raw_text"]: | |
| return nested_payload | |
| texts = None | |
| for key in ["rec_text", "text", "label", "transcription", "ocr_text", "rec_texts"]: | |
| texts = safe_result_get(result, key) | |
| if texts not in [None, ""]: | |
| break | |
| scores = None | |
| for key in ["rec_score", "score", "confidence", "probability", "rec_scores"]: | |
| scores = safe_result_get(result, key) | |
| if scores is not None: | |
| break | |
| normalized_texts = normalize_items(texts) | |
| normalized_scores = normalize_items(scores) | |
| raw_text = "" | |
| if normalized_texts: | |
| raw_text = " ".join(str(item).strip() for item in normalized_texts if str(item).strip()) | |
| elif isinstance(texts, str): | |
| raw_text = texts.strip() | |
| score = 0.0 | |
| if normalized_scores: | |
| try: | |
| score = max(float(item) for item in normalized_scores if item is not None) | |
| except Exception: | |
| score = 0.0 | |
| else: | |
| try: | |
| if scores is not None: | |
| score = float(scores) | |
| except Exception: | |
| score = 0.0 | |
| return { | |
| "raw_text": raw_text, | |
| "score": score, | |
| } | |
| def run_paddleocr(ocr_engine, img: np.ndarray) -> list: | |
| """ | |
| تشغيل PaddleOCR على صورة وإرجاع النتايج. | |
| يدعم API القديم (< 2.9) والجديد (>= 2.9). | |
| الناتج دايمًا: list of [bbox, (text, score)] | |
| """ | |
| try: | |
| result = ocr_engine.ocr(img) | |
| if result is None: | |
| return [] | |
| # --- API القديم (< 2.9) --- | |
| # result = [ [ [bbox, (text, score)], ... ] ] | |
| if ( | |
| isinstance(result, list) | |
| and result | |
| and isinstance(result[0], list) | |
| and result[0] | |
| and isinstance(result[0][0], (list, tuple)) | |
| and len(result[0][0]) == 2 | |
| and isinstance(result[0][0][1], (list, tuple)) | |
| ): | |
| page = result[0] or [] | |
| return page | |
| # --- API الجديد (>= 2.9) --- | |
| # result هو list of OCRResult objects أو dicts | |
| lines = [] | |
| items = result if isinstance(result, list) else [result] | |
| for item in items: | |
| # لو OCRResult object | |
| if hasattr(item, "rec_texts") and hasattr(item, "dt_polys"): | |
| texts = item.rec_texts or [] | |
| scores = item.rec_scores or [] | |
| boxes = item.dt_polys or [] | |
| for bbox, text, score in zip(boxes, texts, scores): | |
| if text: | |
| lines.append([bbox, (text, float(score) if score else 0.0)]) | |
| # لو dict | |
| elif isinstance(item, dict): | |
| texts = item.get("rec_texts", item.get("texts", [])) | |
| scores = item.get("rec_scores", item.get("scores", [])) | |
| boxes = item.get("dt_polys", item.get("boxes", [])) | |
| for bbox, text, score in zip(boxes, texts, scores): | |
| if text: | |
| lines.append([bbox, (text, float(score) if score else 0.0)]) | |
| # fallback: old-style single page | |
| elif isinstance(item, list) and item and isinstance(item[0], (list, tuple)): | |
| lines.extend(item) | |
| return lines | |
| except Exception as e: | |
| print(f"PaddleOCR inference error: {e}") | |
| traceback.print_exc() | |
| return [] | |
| def process_image(img: np.ndarray, ar_ocr, recognizer, min_conf: float, en_ocr=None) -> List[Dict]: | |
| """ | |
| معالجة صورة باستخدام PaddleOCR ثنائي اللغة (عربي + إنجليزي). | |
| ar_ocr: PaddleOCR instance للعربي | |
| en_ocr: PaddleOCR instance للإنجليزي (اختياري) | |
| """ | |
| validate_image_array(img) | |
| h_img, w_img = img.shape[:2] | |
| # Arabic OCR | |
| ar_lines = run_paddleocr(ar_ocr, img) | |
| # English OCR — بس لو في موديل إنجليزي | |
| en_lines = [] | |
| if en_ocr is not None: | |
| en_lines = run_paddleocr(en_ocr, img) | |
| print(f"AR lines: {len(ar_lines)}, EN lines: {len(en_lines)}") | |
| # دمج النتايج — كل line عبارة عن [bbox, (text, score)] | |
| # بناء dict بـ bbox key لتفادي التكرار | |
| combined: dict = {} | |
| def add_lines(lines, script_hint): | |
| for line in lines: | |
| if not line or len(line) < 2: | |
| continue | |
| try: | |
| bbox_raw, (text, score) = line[0], line[1] | |
| text = (text or "").strip() | |
| score = float(score) if score else 0.0 | |
| if not text or score < min_conf: | |
| continue | |
| # تحويل bbox لـ [x1,y1,x2,y2] | |
| pts = np.array(bbox_raw, dtype=np.int32) | |
| x, y, w, h = cv2.boundingRect(pts) | |
| x1 = max(int(x), 0) | |
| y1 = max(int(y), 0) | |
| x2 = min(int(x + w), w_img) | |
| y2 = min(int(y + h), h_img) | |
| if x2 <= x1 or y2 <= y1: | |
| continue | |
| key = (x1, y1, x2, y2) | |
| # تحديد اللغة | |
| script = detect_script(text) | |
| cleaned = smart_clean_arabic_text(text) | |
| if not cleaned: | |
| continue | |
| # لو الـ bbox موجود بالفعل، خد الأعلى confidence | |
| if key in combined: | |
| if score > combined[key]["confidence"]: | |
| combined[key] = { | |
| "text": cleaned, | |
| "confidence": round(score, 4), | |
| "bbox": list(key), | |
| "script": script, | |
| } | |
| else: | |
| combined[key] = { | |
| "text": cleaned, | |
| "confidence": round(score, 4), | |
| "bbox": list(key), | |
| "script": script, | |
| } | |
| except Exception as e: | |
| print(f"Line parse error: {e}") | |
| continue | |
| add_lines(ar_lines, "arabic") | |
| add_lines(en_lines, "latin") | |
| # ترتيب من أعلى لأسفل، ومن اليمين لليسار | |
| ocr_results = sorted( | |
| combined.values(), | |
| key=lambda x: (x["bbox"][1], -x["bbox"][0]) | |
| ) | |
| # إضافة box_id | |
| for i, item in enumerate(ocr_results): | |
| item["box_id"] = i + 1 | |
| print(f"OCR produced {len(ocr_results)} item(s) with min_conf={min_conf}") | |
| return ocr_results | |
| def read_uploaded_image(contents: bytes) -> np.ndarray: | |
| try: | |
| pil_img = Image.open(BytesIO(contents)).convert("RGB") | |
| img = cv2.cvtColor(np.array(pil_img), cv2.COLOR_RGB2BGR) | |
| validate_image_array(img) | |
| return img | |
| except Exception as e: | |
| raise HTTPException(status_code=400, detail=f"Invalid image file: {str(e)}") | |
| # -------------------- Events -------------------- | |
| def startup_event(): | |
| global model_load_error | |
| ensure_runtime_dirs() | |
| print("Application startup complete.") | |
| print("Versions:", get_versions_info()) | |
| print( | |
| "Thread env:", | |
| { | |
| "OMP_NUM_THREADS": os.environ.get("OMP_NUM_THREADS"), | |
| "OPENBLAS_NUM_THREADS": os.environ.get("OPENBLAS_NUM_THREADS"), | |
| "MKL_NUM_THREADS": os.environ.get("MKL_NUM_THREADS"), | |
| }, | |
| ) | |
| try: | |
| get_models() | |
| print("Startup warmup completed successfully.") | |
| except Exception as e: | |
| # Reset error so first real request can retry instead of getting permanent 500 | |
| model_load_error = None | |
| print(f"Startup warmup failed (will retry on first request): {e}") | |
| traceback.print_exc() | |
| # -------------------- Endpoints -------------------- | |
| def root(): | |
| return { | |
| "name": "OCR Scan Vision API", | |
| "status": "ok", | |
| "pdf_support": PDF_AVAILABLE, | |
| "versions": get_versions_info(), | |
| } | |
| def health(): | |
| return { | |
| "status": "healthy", | |
| "versions": get_versions_info(), | |
| } | |
| def versions(): | |
| return get_versions_info() | |
| def load_models(): | |
| detector, recognizer, en_recognizer = get_models() | |
| return { | |
| "message": "Models loaded successfully", | |
| "detector_loaded": detector is not None, | |
| "ar_recognizer_loaded": recognizer is not None, | |
| "en_recognizer_loaded": en_recognizer is not None, | |
| "detector_name": _loaded_detector_name, | |
| "ar_recognizer_name": _loaded_recognizer_name, | |
| "en_recognizer_name": _loaded_en_recognizer_name, | |
| "arabic_recognizer": "arabic" in (_loaded_recognizer_name or "").lower(), | |
| "versions": get_versions_info(), | |
| } | |
| def reset_models(): | |
| global paddle_detector, paddle_recognizer, paddle_en_recognizer, model_load_error | |
| global _loaded_detector_name, _loaded_recognizer_name, _loaded_en_recognizer_name | |
| paddle_detector = None | |
| paddle_recognizer = None | |
| paddle_en_recognizer = None | |
| model_load_error = None | |
| _loaded_detector_name = None | |
| _loaded_recognizer_name = None | |
| _loaded_en_recognizer_name = None | |
| return {"message": "Model state reset. Call /load-models to reload."} | |
| async def ocr_image( | |
| file: UploadFile = File(...), | |
| min_conf: float = Query(default=0.0, ge=0.0, le=1.0), | |
| ): | |
| contents = await file.read() | |
| img = read_uploaded_image(contents) | |
| detector, recognizer, en_recognizer = get_models() | |
| try: | |
| ocr_results = process_image(img, detector, recognizer, min_conf, en_recognizer) | |
| except Exception as e: | |
| traceback.print_exc() | |
| raise HTTPException( | |
| status_code=500, | |
| detail={ | |
| "message": "OCR inference failed", | |
| "error": str(e), | |
| "versions": get_versions_info(), | |
| "filename": file.filename, | |
| "min_conf": min_conf, | |
| } | |
| ) | |
| full_text = "\n".join([r["text"] for r in ocr_results]) | |
| return { | |
| "items": ocr_results, | |
| "text": full_text, | |
| "total_boxes": len(ocr_results), | |
| "filename": file.filename, | |
| } | |
| async def ocr_pdf( | |
| file: UploadFile = File(...), | |
| # DPI أقل = أسرع + مناسب لمعظم الـ CVs | |
| dpi: int = Query(default=200, ge=72, le=600), | |
| min_conf: float = Query(default=0.0, ge=0.0, le=1.0), | |
| # Limit للصفحات لتفادي timeouts في n8n/clients | |
| max_pages: int = Query(default=2, ge=1, le=20), | |
| # لو محتاج النص فقط (أسرع وأخف في الشبكة) | |
| text_only: bool = Query(default=True), | |
| ): | |
| if not PDF_AVAILABLE: | |
| raise HTTPException( | |
| status_code=500, | |
| detail="PDF support not available. pdf2image or poppler may be missing." | |
| ) | |
| try: | |
| contents = await file.read() | |
| # pdf2image: تحديد عدد صفحات أقل يقلل الزمن جدًا | |
| pages = convert_from_bytes(contents, dpi=dpi, first_page=1, last_page=max_pages) | |
| except Exception as e: | |
| raise HTTPException(status_code=400, detail=f"Invalid PDF file: {str(e)}") | |
| detector, recognizer, en_recognizer = get_models() | |
| all_results = [] | |
| all_text = [] | |
| for page_num, pil_img in enumerate(pages, start=1): | |
| try: | |
| img = cv2.cvtColor(np.array(pil_img.convert("RGB")), cv2.COLOR_RGB2BGR) | |
| validate_image_array(img) | |
| page_results = process_image(img, detector, recognizer, min_conf, en_recognizer) | |
| for item in page_results: | |
| item["page"] = page_num | |
| all_results.extend(page_results) | |
| page_text = "\n".join([r["text"] for r in page_results]) | |
| if page_text: | |
| all_text.append(f"--- Page {page_num} ---\n{page_text}") | |
| except Exception as e: | |
| traceback.print_exc() | |
| raise HTTPException( | |
| status_code=500, | |
| detail={ | |
| "message": f"OCR failed on PDF page {page_num}", | |
| "error": str(e), | |
| "versions": get_versions_info(), | |
| "filename": file.filename, | |
| "dpi": dpi, | |
| "min_conf": min_conf, | |
| } | |
| ) | |
| joined_text = "\n\n".join(all_text) | |
| if text_only: | |
| return { | |
| "pages": len(pages), | |
| "text": joined_text, | |
| "filename": file.filename, | |
| } | |
| return { | |
| "pages": len(pages), | |
| "items": all_results, | |
| "text": joined_text, | |
| "total_boxes": len(all_results), | |
| "filename": file.filename, | |
| } | |
| if __name__ == "__main__": | |
| port = int(os.environ.get("PORT", 7860)) | |
| uvicorn.run(app, host="0.0.0.0", port=port) | |