ai_utils/document_processor.py

"""
معالج المستندات للـ RAG
Document Processor for HR Genie RAG System
"""

from typing import List, Set
from ai_utils.rag_chain import Document


class DocumentProcessor:
    """معالج المستندات للـ RAG"""
    
    def __init__(self, chunk_size: int = 500, chunk_overlap: int = 50):
        """
        تهيئة المعالج
        
        Args:
            chunk_size: حجم القطعة (عدد الأحرف)
            chunk_overlap: التداخل بين القطع
        """
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
    
    def split_documents(self, documents: List[Document]) -> List[Document]:
        """
        تقسيم المستندات إلى أجزاء صغيرة (مبسط)
        
        Args:
            documents: قائمة المستندات
        
        Returns:
            قائمة المستندات المقسمة
        """
        # نظراً لأن المستندات صغيرة نسبياً (أسئلة وأجوبة)، لا نحتاج لتقسيمها
        # يمكن إضافة تقسيم مخصص إذا لزم الأمر
        return documents
    
    def process_for_rag(self, documents: List[Document]) -> List[Document]:
        """
        معالجة المستندات للـ RAG
        
        Args:
            documents: قائمة المستندات الأصلية
        
        Returns:
            قائمة المستندات المعالجة
        """
        processed = []
        
        for doc in documents:
            # تنظيف المحتوى
            content = doc.page_content.strip()
            
            # تخطي المستندات الفارغة أو القصيرة جداً
            if len(content) < 20:
                continue
            
            # إضافة معلومات إضافية
            metadata = doc.metadata.copy()
            metadata['content_length'] = len(content)
            metadata['processed'] = True
            
            processed_doc = Document(
                page_content=content,
                metadata=metadata
            )
            processed.append(processed_doc)
        
        print(f"✅ تم معالجة {len(processed)} مستند")
        return processed
    
    def deduplicate(self, documents: List[Document]) -> List[Document]:
        """
        إزالة المستندات المكررة
        
        Args:
            documents: قائمة المستندات
        
        Returns:
            قائمة المستندات بدون تكرار
        """
        seen_content: Set[str] = set()
        unique_docs: List[Document] = []
        
        for doc in documents:
            # استخدام hash للمحتوى للكشف عن التكرار
            content_hash = hash(doc.page_content.strip().lower())
            
            if content_hash not in seen_content:
                seen_content.add(content_hash)
                unique_docs.append(doc)
        
        removed_count = len(documents) - len(unique_docs)
        if removed_count > 0:
            print(f"🗑️ تم إزالة {removed_count} مستند مكرر")
        
        return unique_docs
    
    def enrich_metadata(self, documents: List[Document]) -> List[Document]:
        """
        إثراء البيانات الوصفية للمستندات
        
        Args:
            documents: قائمة المستندات
        
        Returns:
            قائمة المستندات مع بيانات وصفية إضافية
        """
        for doc in documents:
            content = doc.page_content.lower()
            
            # تصنيف تلقائي بناءً على الكلمات المفتاحية
            if any(word in content for word in ['إجازة', 'اجازة', 'عطلة', 'راحة']):
                doc.metadata['auto_category'] = 'الإجازات'
            elif any(word in content for word in ['راتب', 'مرتب', 'بدل', 'علاوة', 'مكافأة']):
                doc.metadata['auto_category'] = 'الرواتب والمزايا'
            elif any(word in content for word in ['تدريب', 'دورة', 'كورس', 'تطوير']):
                doc.metadata['auto_category'] = 'التدريب والتطوير'
            elif any(word in content for word in ['تأمين', 'صحي', 'طبي']):
                doc.metadata['auto_category'] = 'التأمين الصحي'
            elif any(word in content for word in ['استقالة', 'ترك', 'إنهاء']):
                doc.metadata['auto_category'] = 'الاستقالة'
            elif any(word in content for word in ['تقييم', 'أداء', 'ترقية']):
                doc.metadata['auto_category'] = 'تقييم الأداء'
            else:
                doc.metadata['auto_category'] = 'عام'
            
            # إضافة طول المحتوى
            doc.metadata['word_count'] = len(doc.page_content.split())
        
        return documents
    
    def full_process(self, documents: List[Document], 
                     split: bool = False,
                     dedupe: bool = True,
                     enrich: bool = True) -> List[Document]:
        """
        المعالجة الكاملة للمستندات
        
        Args:
            documents: قائمة المستندات
            split: تقسيم المستندات الكبيرة
            dedupe: إزالة التكرار
            enrich: إثراء البيانات الوصفية
        
        Returns:
            قائمة المستندات المعالجة
        """
        print(f"📥 بداية المعالجة: {len(documents)} مستند")
        
        # المعالجة الأساسية
        processed = self.process_for_rag(documents)
        
        # تقسيم المستندات (اختياري)
        if split:
            processed = self.split_documents(processed)
            print(f"✂️ بعد التقسيم: {len(processed)} مستند")
        
        # إزالة التكرار
        if dedupe:
            processed = self.deduplicate(processed)
        
        # إثراء البيانات الوصفية
        if enrich:
            processed = self.enrich_metadata(processed)
        
        print(f"✅ المعالجة اكتملت: {len(processed)} مستند نهائي")
        return processed


if __name__ == "__main__":
    # اختبار المعالج
    processor = DocumentProcessor()
    
    # إنشاء مستندات تجريبية
    test_docs = [
        Document(page_content="كم عدد أيام الإجازة السنوية؟ يحق لكل موظف 21 يوم إجازة.", metadata={"source": "test"}),
        Document(page_content="متى يتم صرف الراتب؟ يتم صرف الرواتب يوم 27 من كل شهر.", metadata={"source": "test"}),
    ]
    
    processed = processor.full_process(test_docs)
    for doc in processed:
        print(f"📄 {doc.metadata.get('auto_category')}: {doc.page_content[:50]}...")