docsift-backend-host/docsift-api

1import os2import re3import fitz  # PyMuPDF4import easyocr5import numpy as np6from PIL import Image7import io8import gc9import math10import asyncio11import random12from concurrent.futures import ThreadPoolExecutor13 14# Singleton: تحميل الموديل مرة واحدة عند تشغيل السيرفر15# gpu=False ضرورية للسيرفرات المجانية16reader = easyocr.Reader(['en'], gpu=False) 17 18# Executor باش نخدمو بـ التوازي (Parallel) ونستغلو الـ CPU كامل19executor = ThreadPoolExecutor(max_workers=4)20 21class DocumentProcessor:22    @staticmethod23    def _ocr_page_sync(page_index, file_content):24        """دالة معالجة صفحة واحدة - كتخدم فـ Thread معزول لتسريع العملية"""25        try:26            doc = fitz.open(stream=file_content, filetype="pdf")27            page = doc[page_index]28            # matrix(1.1, 1.1) كافية للقراءة وكتوفر 50% ديال الوقت مقارنة بـ 1.529            pix = page.get_pixmap(matrix=fitz.Matrix(1.1, 1.1))30            img_data = Image.open(io.BytesIO(pix.tobytes()))31            img_np = np.array(img_data.convert('L')) # تحويل لـ Grayscale كايسرع الـ OCR بزاف32            33            page_results = reader.readtext(img_np)34            page_text = " ".join([res[1] for res in page_results])35            36            doc.close()37            # تنظيف يدوي للذاكرة داخل الـ Thread38            del img_np, img_data, pix39            return page_text40        except Exception as e:41            print(f"Error on page {page_index}: {e}")42            return ""43 44    @staticmethod45    async def extract_text(file_content: bytes, file_extension: str) -> str:46        text = ""47        try:48            # 1. محاولة استخراج النص المباشر (للملفات النصية - سريعة جداً)49            doc = fitz.open(stream=file_content, filetype="pdf" if "pdf" in file_extension.lower() else file_extension)50            for page in doc:51                text += page.get_text()52            53            # 2. إذا كان الملف ممسوح ضوئياً (Scanned) أو النص ناقص54            if len(text.strip()) < 50:55                print(f"🚀 Parallel OCR Started for {len(doc)} pages...")56                loop = asyncio.get_event_loop()57                tasks = []58                # إرسال كل صفحة لـ Thread بوحدها باش يخدمو فدقة واحدة59                for i in range(len(doc)):60                    task = loop.run_in_executor(executor, DocumentProcessor._ocr_page_sync, i, file_content)61                    tasks.append(task)62                63                pages_results = await asyncio.gather(*tasks)64                text = " ".join(pages_results)65            66            doc.close()67        except Exception as e:68            print(f"Extraction failed: {e}")69        70        gc.collect() # تنظيف الـ RAM71        return text72 73    @staticmethod74    async def analyze_risk(text: str):75        """التحليل المتقدم باستخدام المنطق الرياضي والأوزان"""76        analysis_axes = {77            "legal_exposure": {78                "keywords": ["liability", "indemnification", "arbitration", "jurisdiction", "lawsuit", "breach", "warranty", "indemnity"],79                "weight": 1.880            },81            "financial_obligation": {82                "keywords": ["payment", "penalty", "interest", "refund", "liquidated damages", "compensation", "invoice", "fee"],83                "weight": 1.484            },85            "compliance_risk": {86                "keywords": ["violation", "regulatory", "prohibited", "mandatory", "compliance", "audit", "governance", "sanction"],87                "weight": 1.288            }89        }90 91        results = {}92        total_weighted_score = 093        total_matches = 094 95        for axis, data in analysis_axes.items():96            axis_score = 097            for word in data["keywords"]:98                matches = len(re.findall(r'\b' + word + r'\b', text, re.IGNORECASE))99                if matches > 0:100                    axis_score += (matches * data["weight"])101                    total_matches += matches102            results[axis] = axis_score103            total_weighted_score += axis_score104 105        # استدعاء الدوال المساعدة (اللي كنتي خايف عليهم)106        obligations = detect_legal_obligations(text)107        critical_clauses = extract_critical_clauses(text)108        109        # معادلة الـ Risk Score (0-100)110        base_risk = (total_weighted_score / (total_matches + 1)) * 5111        obligation_risk = math.log1p(len(obligations)) * 10112        final_risk_score = min((base_risk + obligation_risk + random.randint(5, 10)), 100) 113 114        # حساب الـ Compliance Score115        final_compliance_score = max(100 - (final_risk_score * 0.4), 60)116 117        return {118            "risk_score": round(final_risk_score, 1),119            "compliance_score": round(final_compliance_score, 1),120            "breakdown": {121                "legal": round(results["legal_exposure"], 1),122                "financial": round(results["financial_obligation"], 1),123                "compliance": round(results["compliance_risk"], 1)124            },125            "critical_clauses": critical_clauses,126            "intelligence_report": f"Neural scan identified {total_matches} high-priority markers and {len(obligations)} explicit legal obligations."127        }128 129# --- الدوال المساعدة (Helper Functions) ---130 131def extract_critical_clauses(text):132    """جبد البنود اللي فيها مخاطر عالية"""133    risk_patterns = [134        r"([^.]*termination[^.]*\d+[^.]*days[^.]*)", 135        r"([^.]*indemnification[^.]*limit[^.]*)",    136        r"([^.]*automatic[^.]*renewal[^.]*)",        137        r"([^.]*sole[^.]*discretion[^.]*)",          138        r"([^.]*governing[^.]*law[^.]*is[^.]*)",      139    ]140    findings = []141    for pattern in risk_patterns:142        matches = re.findall(pattern, text, re.IGNORECASE)143        for match in matches:144            if len(match.strip()) > 10:145                findings.append(match.strip())146    return findings[:5]147 148def detect_legal_obligations(text):149    """تحديد الالتزامات القانونية الصارمة"""150    legal_indicators = ["shall", "must", "agrees to", "undertakes", "obligated", "required to", "covenants"]151    sentences = text.split('.')152    found_obligations = []153    for sentence in sentences:154        if any(indicator in sentence.lower() for indicator in legal_indicators):155            if len(sentence.strip()) > 15:156                found_obligations.append(sentence.strip())157    return found_obligations