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 -------------------- @app.on_event("startup") 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 -------------------- @app.get("/") def root(): return { "name": "OCR Scan Vision API", "status": "ok", "pdf_support": PDF_AVAILABLE, "versions": get_versions_info(), } @app.get("/health") def health(): return { "status": "healthy", "versions": get_versions_info(), } @app.get("/versions") def versions(): return get_versions_info() @app.api_route("/load-models", methods=["GET", "POST"]) 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(), } @app.api_route("/reset-models", methods=["GET", "POST"]) 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."} @app.post("/ocr") 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, } @app.post("/ocr-pdf") 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)