modules/services/risk_analyzer.py

"""

خدمة تحليل المخاطر في المستندات

"""

import re
import pandas as pd
import numpy as np
from nltk.tokenize import sent_tokenize
import config

class RiskAnalyzer:
    """تحليل المخاطر في المستندات"""
    
    def __init__(self):
        # قائمة بالمصطلحات التي تشير إلى المخاطر
        self.risk_indicators = {
            'مخاطر مالية': [
                'غرامة', 'عقوبة', 'تعويض', 'دفعة', 'ضمان', 'تأخير', 'سعر',
                'تكلفة', 'زيادة', 'تمويل', 'استرداد', 'مصادرة', 'كفالة',
                'مستحقات', 'فاتورة', 'سداد', 'دفع', 'مطالبة', 'تقلبات'
            ],
            'مخاطر زمنية': [
                'مدة', 'فترة', 'تاريخ', 'موعد', 'تأخير', 'جدول زمني', 'تمديد',
                'تسليم', 'تسريع', 'إنجاز', 'تنفيذ', 'انتهاء', 'بدء', 'تعليق'
            ],
            'مخاطر فنية': [
                'مواصفات', 'معايير', 'اختبار', 'فحص', 'جودة', 'عيب', 'خلل',
                'تقنية', 'فني', 'تصميم', 'أداء', 'مخططات', 'تشغيل', 'صيانة'
            ],
            'مخاطر إدارية': [
                'مراسلات', 'اجتماع', 'تنسيق', 'تواصل', 'إشراف', 'إدارة',
                'تغيير', 'تعديل', 'موافقة', 'رفض', 'تفويض', 'صلاحية'
            ],
            'مخاطر تنظيمية': [
                'لائحة', 'تصريح', 'ترخيص', 'قانون', 'نظام', 'حكومي', 'بلدية',
                'تشريع', 'امتثال', 'تعميم', 'شهادة', 'موافقة'
            ],
            'مخاطر سوقية': [
                'توريد', 'مورد', 'سوق', 'منافسة', 'مواد', 'نقص', 'تقلب', 'أسعار',
                'استيراد', 'تصدير', 'جمارك', 'نقل', 'تخزين'
            ],
        }
        
        # قائمة بالمصطلحات التي تشير إلى تأثير المخاطر
        self.impact_indicators = {
            'عالي': [
                'كبير', 'خطير', 'جسيم', 'كلي', 'مرتفع', 'عالي', 'ضخم', 'هام',
                'جوهري', 'أساسي', 'رئيسي'
            ],
            'متوسط': [
                'متوسط', 'معتدل', 'وسط', 'مقبول', 'عادي', 'معقول'
            ],
            'منخفض': [
                'صغير', 'قليل', 'ضئيل', 'بسيط', 'منخفض', 'هامشي', 'محدود',
                'طفيف', 'غير مؤثر'
            ]
        }
        
        # قائمة بالمصطلحات التي تشير إلى احتمالية المخاطر
        self.probability_indicators = {
            'مؤكد': [
                'مؤكد', 'حتمي', 'قطعي', 'دائماً', 'يجب', 'ملزم', 'إلزامي',
                'مطلوب'
            ],
            'محتمل': [
                'محتمل', 'ممكن', 'قد', 'ربما', 'يمكن', 'متوقع'
            ],
            'غير محتمل': [
                'نادر', 'بعيد', 'استثنائي', 'غير متوقع', 'غير محتمل', 'ضئيل'
            ]
        }
        
        # استراتيجيات معالجة المخاطر
        self.mitigation_strategies = {
            'مخاطر مالية': [
                "تخصيص مبلغ احتياطي",
                "التفاوض مع العميل لتخفيف الشروط المالية",
                "تحديد سقف للغرامات",
                "التخطيط للتدفق النقدي",
                "تأمين خط ائتمان احتياطي"
            ],
            'مخاطر زمنية': [
                "زيادة فريق العمل",
                "استخدام موارد إضافية",
                "وضع خطة عمل بديلة",
                "استباق التأخيرات المحتملة",
                "تقديم طلب تمديد مسبق"
            ],
            'مخاطر فنية': [
                "طلب توضيح من العميل",
                "استشارة خبراء متخصصين",
                "إجراء اختبارات إضافية",
                "توثيق المراسلات الفنية",
                "تعيين مسؤول ضبط جودة"
            ],
            'مخاطر إدارية': [
                "تحسين آليات التواصل",
                "توثيق جميع المراسلات",
                "وضع خطة اتصال واضحة",
                "عقد اجتماعات دورية",
                "تعيين مدير مشروع متفرغ"
            ],
            'مخاطر تنظيمية': [
                "التخطيط المسبق للمتطلبات التنظيمية",
                "التواصل مع الجهات المعنية",
                "الاستعانة بمستشار قانوني",
                "متابعة التغييرات التنظيمية",
                "تجهيز الوثائق المطلوبة مبكراً"
            ],
            'مخاطر سوقية': [
                "تثبيت أسعار المواد مع الموردين",
                "البحث عن موردين بدلاء",
                "شراء المواد الرئيسية مبكراً",
                "إبرام عقود توريد طويلة الأجل",
                "مراقبة تقلبات السوق"
            ]
        }
    
    def analyze_risks(self, text):
        """تحليل المخاطر في النص المعطى"""
        if not text:
            return pd.DataFrame()
        
        # تقسيم النص إلى جمل
        sentences = sent_tokenize(text)
        
        # تحليل المخاطر في كل جملة
        risks = []
        risk_id = 1
        
        for sentence in sentences:
            # تحديد نوع المخاطرة إذا وجدت
            risk_category = self._determine_risk_category(sentence)
            
            if risk_category:
                # تحديد التأثير والاحتمالية
                impact = self._determine_impact(sentence)
                probability = self._determine_probability(sentence)
                
                # اختيار استراتيجية المعالجة
                mitigation = np.random.choice(self.mitigation_strategies.get(risk_category, ["مراجعة فريق المخاطر"]))
                
                # إضافة المخاطرة إلى القائمة
                risks.append({
                    'رقم المخاطرة': f"R{risk_id:02d}",
                    'وصف المخاطرة': sentence.strip(),
                    'الفئة': risk_category,
                    'التأثير': impact,
                    'الاحتمالية': probability,
                    'استراتيجية المعالجة': mitigation
                })
                
                risk_id += 1
        
        # تحويل القائمة إلى DataFrame
        risks_df = pd.DataFrame(risks)
        
        # التأكد من وجود بيانات
        if risks_df.empty:
            # إنشاء DataFrame فارغ بالأعمدة المطلوبة
            risks_df = pd.DataFrame(columns=[
                'رقم المخاطرة', 'وصف المخاطرة', 'الفئة', 
                'التأثير', 'الاحتمالية', 'استراتيجية المعالجة'
            ])
        
        return risks_df
    
    def _determine_risk_category(self, text):
        """تحديد فئة المخاطرة بناءً على محتوى النص"""
        # البحث عن الكلمات المفتاحية في النص
        scores = {}
        
        for category, indicators in self.risk_indicators.items():
            score = sum(1 for indicator in indicators if indicator in text.lower())
            scores[category] = score
        
        # اختيار الفئة ذات الدرجة الأعلى إذا وجدت
        if max(scores.values(), default=0) > 0:
            return max(scores.items(), key=lambda x: x[1])[0]
        else:
            return None
    
    def _determine_impact(self, text):
        """تحديد تأثير المخاطرة بناءً على محتوى النص"""
        # البحث عن الكلمات المفتاحية في النص
        scores = {}
        
        for impact, indicators in self.impact_indicators.items():
            score = sum(1 for indicator in indicators if indicator in text.lower())
            scores[impact] = score
        
        # اختيار التأثير ذو الدرجة الأعلى
        if max(scores.values(), default=0) > 0:
            return max(scores.items(), key=lambda x: x[1])[0]
        else:
            # اختيار عشوائي مع ترجيح أكبر للتأثير المتوسط
            return np.random.choice(
                ["عالي", "متوسط", "منخفض"],
                p=[0.3, 0.5, 0.2]
            )
    
    def _determine_probability(self, text):
        """تحديد احتمالية المخاطرة بناءً على محتوى النص"""
        # البحث عن الكلمات المفتاحية في النص
        scores = {}
        
        for probability, indicators in self.probability_indicators.items():
            score = sum(1 for indicator in indicators if indicator in text.lower())
            scores[probability] = score
        
        # اختيار الاحتمالية ذات الدرجة الأعلى
        if max(scores.values(), default=0) > 0:
            return max(scores.items(), key=lambda x: x[1])[0]
        else:
            # اختيار عشوائي مع ترجيح أكبر للاحتمالية المتوسطة
            return np.random.choice(
                ["مؤكد", "محتمل", "غير محتمل"],
                p=[0.2, 0.6, 0.2]
            )