Upload main.py

main.py

@@ -1,3 +1,4 @@
+import logging
 from fastapi import FastAPI, File, UploadFile, HTTPException, Query
 from fastapi.middleware.cors import CORSMiddleware
 from typing import List, Dict, Any
@@ -19,7 +20,11 @@ except ImportError:
 # استيراد مكتبة PaddleOCR الرسمية
 from paddleocr import PaddleOCR
 
-app = FastAPI(title="OCR Scan Vision API", version="2.0.0")
+# إعداد الـ Logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger("OCR_Contracts")
+
+app = FastAPI(title="OCR Arabic Contracts API", version="3.0.0")
 
 app.add_middleware(
     CORSMiddleware,
@@ -29,46 +34,54 @@ app.add_middleware(
     allow_headers=["*"],
 )
 
-# متغير عام لتخزين الموديل في الذاكرة
+# متغير عام لتخزين الموديل
 ocr_model = None
 
-# -------------------- تنظيف النص العربي --------------------
-def clean_arabic_text(text: str) -> str:
+# -------------------- تنظيف النص (مخصص للعقود) --------------------
+def clean_contract_text(text: str) -> str:
     """
-    تنظيف النص مع الحفاظ على الأرقام والعربية
+    تنظيف مخصص للعقود العربية:
+    - يحافظ على العربية والأرقام
+    - يحافظ على رموز التواريخ والعقود (/, -, :, ., ())
+    - يزيل الرموز الغريبة فقط
     """
     if not text:
         return ""
     
-    # استبدال الرموز بمسافات
-    text = re.sub(r"[:\-_/]", " ", text)
-    # الاحتفاظ بالعربية والأرقام والمسافات فقط
-    text = re.sub(r"[^\u0600-\u06FF0-9\s]", "", text)
-    # إزالة التشكيل
-    text = re.sub(r"[\u064B-\u065F]", "", text)
-    # ضبط المسافات المتكررة
-    text = re.sub(r"\s+", " ", text)
+    # 1. إزالة أي رمز غير مرغوب فيه، لكن نترك:
+    # \u0600-\u06FF : الحروف العربية
+    # 0-9 : الأرقام
+    # \s : المسافات
+    # . , - / : ( ) [ ] _ : علامات الترقيم المهمة في العقود والتواريخ
+    clean = re.sub(r"[^\u0600-\u06FF0-9\s\.\,\-\/\:\(\)\[\]\_]", " ", text)
+    
+    # 2. إزالة التشكيل (عادة غير مطلوب في العقود القانونية لتسهيل البحث)
+    clean = re.sub(r"[\u064B-\u065F]", "", clean)
     
-    return text.strip()
+    # 3. ضبط المسافات المتكررة الناتجة عن الحذف
+    clean = re.sub(r"\s+", " ", clean)
+    
+    return clean.strip()
 
 def get_model():
-    """
-    تحميل الموديل مرة واحدة عند بدء التشغيل أو أول طلب.
-    """
     global ocr_model
     if ocr_model is None:
-        print("Loading PaddleOCR model (Arabic)...")
-        # تم تفعيل use_angle_cls=True هنا، وهذا يكفي لتشغيل المصحح
-        ocr_model = PaddleOCR(use_angle_cls=True, lang='ar')
-        print("Model loaded successfully.")
+        logger.info("Loading PaddleOCR model for Arabic Contracts...")
+        try:
+            # lang='ar' يدعم العربية والإنجليزية (للأرقام)
+            # use_angle_cls=True ضروري للمستندات المائلة (Scanned Docs)
+            ocr_model = PaddleOCR(use_angle_cls=True, lang='ar')
+            logger.info("Model loaded successfully.")
+        except Exception as e:
+            logger.error(f"Failed to load model: {e}")
+            raise e
     return ocr_model
 
-def organize_boxes(results: List, y_threshold: int = 15) -> List[Dict]:
+def organize_boxes_rtl(results: List, y_threshold: int = 15) -> List[Dict]:
     """
-    خوارزمية لترتيب النتائج:
-    1. تجميع الكلمات التي تقع على نفس السطر تقريباً.
-    2. ترتيب الأسطر من الأعلى للأسفل.
-    3. ترتيب الكلمات داخل السطر الواحد من اليمين لليسار.
+    خوارزمية ترتيب النصوص العربية (RTL):
+    1. تجميع الكلمات في أسطر بناءً على المحور Y.
+    2. ترتيب الكلمات داخل كل سطر من اليمين إلى اليسار (X تنازلي).
     """
     if not results:
         return []
@@ -79,13 +92,11 @@ def organize_boxes(results: List, y_threshold: int = 15) -> List[Dict]:
         coords = item[0]
         text_data = item[1]
         
-        # حساب إحداثيات الصندوق
         y_min = min(c[1] for c in coords)
         y_max = max(c[1] for c in coords)
         x_min = min(c[0] for c in coords)
         x_max = max(c[0] for c in coords)
         
-        # مركز الصندوق رأسياً
         y_center = (y_min + y_max) / 2
         
         boxes.append({
@@ -93,17 +104,17 @@ def organize_boxes(results: List, y_threshold: int = 15) -> List[Dict]:
             "confidence": text_data[1],
             "coords": coords, 
             "y_center": y_center,
-            "x_max": x_max, # نستخدم أقصى اليمين للترتيب العربي
+            "x_max": x_max, # نستخدم أقصى اليمين للترتيب
             "bbox": [int(x_min), int(y_min), int(x_max), int(y_max)]
         })
 
-    # ترتيب أولي حسب الموقع الرأسي (Y)
+    # 1. ترتيب الصناديق رأسياً (من أعلى لأسفل)
     boxes.sort(key=lambda b: b["y_center"])
 
     lines = []
     current_line = []
     
-    # تجميع العناصر في أسطر
+    # 2. تجميع الأسطر
     for box in boxes:
         if not current_line:
             current_line.append(box)
@@ -111,63 +122,66 @@ def organize_boxes(results: List, y_threshold: int = 15) -> List[Dict]:
         
         last_box = current_line[-1]
         
-        # إذا كان الفرق في الارتفاع الرأسي بسيطاً، نعتبرهم في نفس السطر
+        # إذا كان الفرق في الارتفاع بسيط، نعتبرهم نفس السطر
         if abs(box["y_center"] - last_box["y_center"]) < y_threshold:
             current_line.append(box)
         else:
-            # بدء سطر جديد
             lines.append(current_line)
             current_line = [box]
     
     if current_line:
         lines.append(current_line)
 
-    # الترتيب النهائي
+    # 3. ترتيب كل سطر من اليمين لليسار (هام جداً للعربي)
     final_sorted_boxes = []
     for line in lines:
-        # نرتب عناصر السطر الواحد تنازلياً حسب X (من اليمين لليسار)
+        # ترتيب تنازلي حسب x_max (الأكبر هو الموجود في اليمين)
         line.sort(key=lambda b: -b["x_max"]) 
         final_sorted_boxes.extend(line)
 
     return final_sorted_boxes
 
 def process_image_full(img: np.ndarray, model, min_conf: float) -> List[Dict]:
-    # PaddleOCR يستقبل الصورة مباشرة
+    logger.info("Processing image...")
+    
+    # PaddleOCR Predict
     result = model.ocr(img)
     
-    # التحقق من أن النتيجة صالحة وليست فارغة
-    # PaddleOCR returns [ [line1, line2, ...] ] for single image
     if not result or result[0] is None:
+        logger.warning("No text found.")
         return []
     
     raw_results = result[0]
     
-    # تصفية النتائج بطريقة آمنة لتجنب الأخطاء (IndexError)
+    # تصفية النتائج
     filtered_results = []
     for item in raw_results:
-        # التحقق من هيكل العنصر: يجب أن يكون قائمة تحتوي على عنصرين [coords, (text, conf)]
         if not isinstance(item, (list, tuple)) or len(item) < 2:
             continue
-        
         text_part = item[1]
-        
-        # التحقق من جزء النص والثقة: يجب أن يكون (text, conf)
         if isinstance(text_part, (list, tuple)) and len(text_part) >= 2:
             text, conf = text_part
+            # نقبل النص إذا تجاوز نسبة الثقة المطلوبة
             if conf >= min_conf:
                 filtered_results.append(item)
-    
-    # تطبيق خوارزمية الترتيب والفلترة
-    sorted_boxes = organize_boxes(filtered_results)
+
+    # تطبيق الترتيب العربي (RTL)
+    sorted_boxes = organize_boxes_rtl(filtered_results)
     
     final_output = []
     for i, box in enumerate(sorted_boxes):
-        clean_text = clean_arabic_text(box["text"])
-        if clean_text:
+        # تنظيف النص باستخدام الدالة المخصصة للعقود
+        clean_txt = clean_contract_text(box["text"])
+        
+        # إذا كان النص فارغاً بعد التنظيف، نستخدم النص الأصلي كاحتياط
+        # هذا يمنع فقدان البيانات إذا أخطأ التنظيف
+        final_text = clean_txt if clean_txt else box["text"]
+        
+        if final_text.strip():
             final_output.append({
                 "box_id": i + 1,
-                "text": clean_text,
-                "original_text": box["text"],
+                "text": final_text,
+                # "original_text": box["text"], # يمكن تفعيل هذا للدييباج
                 "confidence": round(box["confidence"], 4),
                 "bbox": box["bbox"]
             })
@@ -176,16 +190,12 @@ def process_image_full(img: np.ndarray, model, min_conf: float) -> List[Dict]:
 
 @app.get("/")
 def root():
-    return {"name": "OCR Scan Vision API v2", "status": "ok", "pdf_support": PDF_AVAILABLE}
-
-@app.get("/health")
-def health():
-    return {"status": "healthy"}
+    return {"name": "Arabic Contract OCR API", "status": "Ready", "pdf_support": PDF_AVAILABLE}
 
 @app.post("/ocr")
 async def ocr_image(
     file: UploadFile = File(...),
-    min_conf: float = Query(default=0.6, ge=0.0, le=1.0),
+    min_conf: float = Query(default=0.4, ge=0.0, le=1.0), # مناسب للمستندات الواضحة نوعاً ما
 ):
     try:
         contents = await file.read()
@@ -197,29 +207,29 @@ async def ocr_image(
     model = get_model()
     ocr_results = process_image_full(img, model, 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)
+        "total_lines": len(ocr_results)
     }
 
 @app.post("/ocr-pdf")
 async def ocr_pdf(
     file: UploadFile = File(...),
     dpi: int = Query(default=300, ge=72, le=600),
-    min_conf: float = Query(default=0.6, ge=0.0, le=1.0),
+    min_conf: float = Query(default=0.4, ge=0.0, le=1.0),
 ):
     if not PDF_AVAILABLE:
-        raise HTTPException(status_code=500, detail="PDF support not available")
+        raise HTTPException(status_code=500, detail="PDF support not available (install pdf2image & poppler)")
 
     try:
         contents = await file.read()
         pages = convert_from_bytes(contents, dpi=dpi)
     except Exception as e:
-        raise HTTPException(status_code=400, detail=f"Invalid PDF file: {e}")
+        raise HTTPException(status_code=400, detail=f"Invalid PDF: {e}")
 
     model = get_model()
     all_results = []
@@ -227,23 +237,22 @@ async def ocr_pdf(
 
     for page_num, pil_img in enumerate(pages, start=1):
         img = cv2.cvtColor(np.array(pil_img.convert("RGB")), cv2.COLOR_RGB2BGR)
-        
         page_results = process_image_full(img, model, min_conf)
 
         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}")
+        
+        page_txt = "\n".join([r["text"] for r in page_results])
+        if page_txt:
+            all_text.append(f"--- Page {page_num} ---\n{page_txt}")
 
     return {
         "pages": len(pages),
         "items": all_results,
         "text": "\n\n".join(all_text),
-        "total_boxes": len(all_results)
+        "total_lines": len(all_results)
     }
 
 if __name__ == "__main__":