#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
وحدة تحليل وتقييم مخاطر العقود بشكل آلي
"""
import os
import sys
import json
import datetime
import re
import numpy as np
import pandas as pd
import streamlit as st
import plotly.express as px
import plotly.graph_objects as go
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.cluster import KMeans
from collections import Counter
# إضافة مسار النظام للوصول للملفات المشتركة
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../..")))
# استيراد المكونات المساعدة
from utils.helpers import create_directory_if_not_exists, format_time, get_user_info
# تعليق استيراد Anthropic (سيتم تنفيذه لاحقًا بعد ضبط ملف anthropic)
# نستخدم نمط fallback للتعامل مع الخطأ
try:
from anthropic import Anthropic
def analyzeBillOfQuantities(text):
return {"analysis": "تحليل فرضي لجدول الكميات", "items": [], "summary": "لا يوجد تحليل حقيقي متاح حاليًا"}
def analyzeTermsAndConditions(text):
return {"analysis": "تحليل فرضي للشروط والأحكام", "risks": [], "summary": "لا يوجد تحليل حقيقي متاح حاليًا"}
anthropic = None # سيتم تعيينه لاحقًا عند الحاجة
except ImportError:
# في حالة عدم وجود مكتبة أنثروبيك، نستخدم دوال فرضية
def analyzeBillOfQuantities(text):
return {"analysis": "تحليل فرضي لجدول الكميات", "items": [], "summary": "لا يوجد تحليل حقيقي متاح حاليًا"}
def analyzeTermsAndConditions(text):
return {"analysis": "تحليل فرضي للشروط والأحكام", "risks": [], "summary": "لا يوجد تحليل حقيقي متاح حاليًا"}
anthropic = None
class ContractRiskAnalyzer:
"""فئة تحليل وتقييم المخاطر في العقود بشكل آلي"""
def __init__(self):
"""تهيئة محلل مخاطر العقود"""
self.risk_data_dir = os.path.join(os.path.dirname(__file__), '..', '..', 'data', 'risk_assessment')
create_directory_if_not_exists(self.risk_data_dir)
# تعريف أنواع المخاطر
self.risk_categories = {
"legal": "قانونية",
"financial": "مالية",
"operational": "تشغيلية",
"technical": "فنية",
"compliance": "امتثال",
"environmental": "بيئية",
"safety": "سلامة",
"schedule": "جدولة",
"resource": "موارد",
"quality": "جودة",
"scope": "نطاق العمل",
"stakeholder": "أصحاب المصلحة",
"commercial": "تجارية",
"contractual": "تعاقدية",
"regulatory": "تنظيمية"
}
# تعريف قائمة المصطلحات الخطرة في العقود
self.risky_terms = {
"legal": [
"تعديل العقد", "فسخ العقد", "إنهاء الاتفاقية", "فض المنازعات", "شرط جزائي",
"تحكيم", "قاهرة", "ظروف قاهرة", "التقاضي", "الإخلال بالعقد", "المنازعات",
"الدفع", "الضمان", "الولاية القضائية", "التعويض"
],
"financial": [
"غرامة تأخير", "غرامات", "دفعة مقدمة", "دفعة نهائية", "ضمان", "تأمين",
"تسعير", "سعر", "خصم", "تكاليف إضافية", "تعديل سعر", "زيادة سعر",
"خسارة", "ربح", "هامش", "تحمل التكاليف", "تمويل", "مخاطر مالية", "ضريبة"
],
"operational": [
"تأخير", "عدم التسليم", "توقف", "انقطاع", "عطل", "خلل", "تعطل",
"عمالة", "أيدي عاملة", "موارد بشرية", "مناولة", "تصاريح", "لوجستيك",
"مخزون", "سلسلة توريد", "عمليات"
],
"technical": [
"مواصفات", "معايير", "شروط فنية", "كفاءة", "جودة", "أداء",
"اختبار", "فحص", "تقنية", "تكنولوجيا", "تشغيل", "تركيب", "صيانة",
"تصميم", "هندسة", "قدرة"
],
"compliance": [
"اللوائح", "القوانين", "التشريعات", "الامتثال", "المعايير", "الترخيص",
"التصريح", "الموافقة", "الالتزام", "التنظيم", "الشهادة"
],
"environmental": [
"بيئي", "بيئة", "تلوث", "تصريف", "نفايات", "انبعاثات", "موارد طبيعية",
"تأثير بيئي", "استدامة", "تعويض بيئي", "ضرر بيئي", "مخلفات"
],
"safety": [
"سلامة", "أمان", "حوادث", "إصابات", "مخاطر صحية", "صحة مهنية",
"وقاية", "حماية", "إجراءات أمان", "تأمين سلامة", "مخاطر السلامة"
],
"schedule": [
"تأخير", "تمديد", "مدة", "جدول زمني", "موعد نهائي", "تسليم",
"مراحل", "مواعيد", "وقت", "فترة", "عاجل", "سريع", "فوري"
],
"resource": [
"مواد", "معدات", "أدوات", "آلات", "عمالة", "كوادر", "فريق",
"موارد بشرية", "توفير", "تأمين", "استقدام", "نقص", "عجز", "كفاية"
],
"quality": [
"جودة", "ضمان الجودة", "معايير", "مواصفات", "أداء", "رداءة",
"ضعف", "خلل", "عيب", "إصلاح", "صيانة", "استبدال", "رفض"
],
"scope": [
"نطاق العمل", "تغيير النطاق", "توسيع", "تقليص", "تعديل", "إضافة",
"أعمال إضافية", "تغييرات", "أوامر تغيير", "متطلبات جديدة"
],
"stakeholder": [
"طرف ثالث", "مالك", "عميل", "المقاول", "المورد", "الاستشاري",
"المشرف", "مدير المشروع", "المقاول من الباطن", "الشريك", "مصلحة"
],
"commercial": [
"منافسة", "سوق", "سعر", "عرض", "طلب", "تجاري", "أعمال",
"استثمار", "عائد", "ربح", "خسارة", "سمعة", "علامة تجارية"
],
"contractual": [
"بند", "شرط", "مادة", "اتفاقية", "عقد", "ملحق", "تعديل",
"تنازل", "تعهد", "التزام", "مسؤولية", "واجب", "حق"
],
"regulatory": [
"تنظيمي", "حكومي", "رسمي", "لائحة", "قانون", "تشريع",
"ترخيص", "تصريح", "موافقة", "امتثال", "اشتراطات", "متطلبات"
]
}
# نموذج تصنيف المخاطر
self.vectorizer = TfidfVectorizer(max_features=1000, stop_words='english')
self.kmeans = KMeans(n_clusters=len(self.risk_categories), random_state=42)
# الصيغ والمصطلحات المتعلقة بالمخاطر التعاقدية
self.contract_risk_patterns = {
"unlimited_liability": [
r"مسؤولية غير محدودة",
r"مسؤولية كاملة",
r"المسؤولية الكاملة",
r"دون تحديد للمسؤولية",
r"دون سقف للمسؤولية",
r"المسؤولية المطلقة",
r"التعويض عن كافة الأضرار"
],
"payment_delay": [
r"(\d+)\s*يوم\s*من تاريخ\s*الفاتورة",
r"(\d+)\s*يوم\s*عمل من تاريخ\s*الفاتورة",
r"(\d+)\s*يوم\s*للدفع",
r"خلال\s*(\d+)\s*يوم",
r"الدفع خلال\s*(\d+)\s*"
],
"excessive_penalties": [
r"غرامة تأخير بنسبة\s*(\d+)%",
r"غرامة تأخير قدرها\s*(\d+)%",
r"غرامة يومية\s*(\d+)%",
r"غرامة اسبوعية\s*(\d+)%",
r"غرامة شهرية\s*(\d+)%"
],
"unilateral_termination": [
r"يحق للطرف الأول إنهاء العقد",
r"يحق للعميل إنهاء العقد",
r"للعميل الحق في إنهاء",
r"للطرف الأول الحق في إنهاء",
r"إنهاء العقد من طرف واحد",
r"إنهاء دون إبداء أسباب"
],
"unrealistic_deadlines": [
r"التسليم خلال\s*(\d+)\s*يوم",
r"مدة التنفيذ\s*(\d+)\s*يوم",
r"الانتهاء خلال\s*(\d+)\s*يوم",
r"إنجاز المشروع خلال\s*(\d+)\s*أسبوع"
],
"scope_creep": [
r"أعمال إضافية",
r"تعديلات على النطاق",
r"توسيع نطاق العمل",
r"إضافة متطلبات",
r"تغيير المواصفات",
r"أعمال غير متوقعة"
],
"indemnification": [
r"تعويض الطرف الأول",
r"تعويض العميل",
r"تعويض كامل",
r"تعويض شامل",
r"التعويض عن كافة الأضرار",
r"التعويض عن أي خسائر"
],
"change_control": [
r"التغييرات بدون تكلفة إضافية",
r"تعديلات دون زيادة السعر",
r"تغييرات دون مقابل",
r"تعديلات لا تؤثر على السعر"
],
"warranty_period": [
r"ضمان لمدة\s*(\d+)\s*شهر",
r"ضمان لمدة\s*(\d+)\s*سنة",
r"فترة ضمان\s*(\d+)\s*شهر",
r"فترة الضمان\s*(\d+)\s*شهر",
r"فترة الصيانة\s*(\d+)\s*شهر"
],
"dispute_resolution": [
r"المحاكم المختصة",
r"محاكم[^.]*للنظر في المنازعات",
r"تسوية النزاعات",
r"فض المنازعات",
r"التحكيم",
r"لجنة تحكيم"
],
"force_majeure": [
r"القوة القاهرة",
r"الظروف القاهرة",
r"ظروف خارجة عن الإرادة",
r"أحداث غير متوقعة",
r"أسباب خارجة عن السيطرة"
],
"regulatory_compliance": [
r"الالتزام بالقوانين",
r"الالتزام بالأنظمة",
r"الالتزام بالتشريعات",
r"الالتزام باللوائح",
r"مراعاة القوانين",
r"وفقاً للقوانين",
r"طبقاً للأنظمة",
],
"intellectual_property": [
r"الملكية الفكرية",
r"حقوق الملكية",
r"حقوق الطبع",
r"حقوق النشر",
r"براءات الاختراع",
r"التصاميم",
r"العلامات التجارية"
],
"confidentiality": [
r"سرية المعلومات",
r"المعلومات السرية",
r"عدم الإفصاح",
r"الحفاظ على السرية",
r"عدم الكشف",
r"معلومات سرية"
],
"insurance_requirements": [
r"متطلبات التأمين",
r"بوليصة تأمين",
r"تأمين ضد المسؤولية",
r"تأمين ضد المخاطر",
r"تأمين شامل",
r"تأمين المشروع"
]
}
# تعريف مستويات خطورة المخاطر
self.severity_levels = {
"low": {
"name": "منخفضة",
"color": "#00b894", # أخضر
"score_range": (0, 33)
},
"medium": {
"name": "متوسطة",
"color": "#fdcb6e", # أصفر
"score_range": (34, 66)
},
"high": {
"name": "عالية",
"color": "#d63031", # أحمر
"score_range": (67, 100)
}
}
# أوزان أنواع المخاطر (الأهمية النسبية)
self.risk_weights = {
"legal": 0.9,
"financial": 0.8,
"operational": 0.7,
"technical": 0.6,
"compliance": 0.8,
"environmental": 0.6,
"safety": 0.7,
"schedule": 0.6,
"resource": 0.5,
"quality": 0.7,
"scope": 0.7,
"stakeholder": 0.5,
"commercial": 0.6,
"contractual": 0.9,
"regulatory": 0.8
}
def scan_contract_text(self, contract_text, title=""):
"""فحص نص العقد لاستخراج المخاطر المحتملة"""
if not contract_text:
return {
"title": title or "عقد غير معروف",
"timestamp": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"risks": [],
"overall_score": 0,
"overall_severity": "منخفضة",
"summary": "لم يتم توفير نص للتحليل."
}
# تحويل النص إلى أحرف صغيرة للفحص
text_lower = contract_text.lower()
risks = []
risk_id = 1
# فحص كل فئة مخاطر والبحث عن المصطلحات المرتبطة بها
for category, category_terms in self.risky_terms.items():
category_risks = []
for term in category_terms:
# البحث عن المصطلح في النص
occurrences = self._find_term_occurrences(contract_text, term)
if occurrences:
for occurrence in occurrences:
context = self._extract_context(contract_text, occurrence, window=100)
# تحديد مستوى الخطورة بناءً على السياق
severity = self._determine_severity_from_context(context, category)
category_risks.append({
"id": risk_id,
"term": term,
"category": category,
"category_ar": self.risk_categories[category],
"context": context,
"severity": severity,
"impact": self._determine_impact(category, severity),
"recommendation": self._generate_recommendation(category, term, severity)
})
risk_id += 1
# إضافة مخاطر الفئة إلى القائمة الرئيسية
risks.extend(category_risks)
# فحص صيغ المخاطر الإضافية في العقد
pattern_risks = self._scan_for_risk_patterns(contract_text, risk_id)
risks.extend(pattern_risks)
# حساب درجة المخاطر الإجمالية
overall_score = self._calculate_overall_risk_score(risks)
# تحديد مستوى الخطورة الإجمالية
overall_severity = self._determine_overall_severity(overall_score)
# توليد ملخص للمخاطر
summary = self._generate_risk_summary(risks, overall_score, overall_severity)
# إنشاء تقرير المخاطر
risk_report = {
"title": title or "عقد غير معروف",
"timestamp": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"risks": risks,
"overall_score": overall_score,
"overall_severity": overall_severity["name"],
"severity_color": overall_severity["color"],
"summary": summary
}
# حفظ تقرير المخاطر
if title:
self._save_risk_report(risk_report, title)
return risk_report
def _find_term_occurrences(self, text, term):
"""البحث عن مواضع ظهور المصطلح في النص"""
occurrences = []
start = 0
while True:
start = text.find(term, start)
if start == -1:
break
occurrences.append(start)
start += len(term)
return occurrences
def _extract_context(self, text, position, window=100):
"""استخراج سياق النص حول موضع معين"""
start = max(0, position - window // 2)
end = min(len(text), position + window // 2)
# البحث عن بداية الجملة
while start > 0 and text[start] not in ['.', '!', '؟', '?', '\n']:
start -= 1
if start > 0:
start += 1 # تجاوز علامة الترقيم
# البحث عن نهاية الجملة
while end < len(text) - 1 and text[end] not in ['.', '!', '؟', '?', '\n']:
end += 1
if end < len(text) - 1:
end += 1 # تضمين علامة الترقيم
return text[start:end].strip()
def _determine_severity_from_context(self, context, category):
"""تحديد مستوى خطورة المخاطر بناءً على السياق"""
# كلمات تزيد من مستوى الخطورة
high_severity_indicators = [
"حرج", "خطير", "ضروري", "إلزامي", "يجب", "مطلوب", "ضمان",
"تعويض", "غرامة", "يلتزم", "مسؤولية", "خسارة", "ضرر",
"تأخير", "مخالفة", "إخلال", "فسخ", "إنهاء", "تعديل"
]
# كلمات تقلل من مستوى الخطورة
low_severity_indicators = [
"قد", "يمكن", "يجوز", "يحتمل", "محتمل", "ممكن", "اختياري",
"تقديري", "بالتوافق", "بالاتفاق", "مناسب", "معقول", "بحسب"
]
# حساب عدد المؤشرات
high_count = sum(1 for indicator in high_severity_indicators if indicator in context)
low_count = sum(1 for indicator in low_severity_indicators if indicator in context)
# تحديد درجة الخطورة
if high_count > low_count * 2:
return "high"
elif high_count > low_count:
return "medium"
else:
return "low"
def _determine_impact(self, category, severity):
"""تحديد تأثير المخاطر بناءً على الفئة ومستوى الخطورة"""
impact_descriptions = {
"legal": {
"high": "قد يؤدي إلى دعاوى قضائية ومسؤولية قانونية كبيرة",
"medium": "قد يتطلب تعديلات قانونية أو مفاوضات إضافية",
"low": "مخاطر قانونية محدودة يمكن معالجتها بسهولة"
},
"financial": {
"high": "مخاطر مالية كبيرة قد تؤثر على ربحية المشروع بشكل كبير",
"medium": "قد يؤدي إلى زيادة التكاليف أو تقليل الهوامش",
"low": "تأثير مالي محدود يمكن استيعابه"
},
"operational": {
"high": "قد يعيق تنفيذ المشروع بشكل كامل",
"medium": "قد يؤثر على كفاءة العمليات ويتطلب خطط بديلة",
"low": "تأثير محدود على العمليات اليومية"
},
"technical": {
"high": "قد يمنع تحقيق المتطلبات الفنية الأساسية",
"medium": "يتطلب حلول فنية إضافية أو تعديلات",
"low": "يمكن معالجته من خلال التعديلات الفنية البسيطة"
},
"compliance": {
"high": "قد يؤدي إلى عدم الامتثال للوائح الهامة",
"medium": "يتطلب تعديلات للامتثال للمتطلبات التنظيمية",
"low": "يمكن حله من خلال تدابير امتثال بسيطة"
},
"environmental": {
"high": "مخاطر بيئية كبيرة قد تؤدي إلى عقوبات أو تأخيرات",
"medium": "يتطلب إجراءات وقائية إضافية للحماية البيئية",
"low": "تأثير بيئي محدود يمكن إدارته"
},
"safety": {
"high": "مخاطر سلامة حرجة قد تهدد سلامة العاملين",
"medium": "يتطلب إجراءات سلامة إضافية وتدريب",
"low": "مخاطر سلامة يمكن معالجتها من خلال الإجراءات القياسية"
},
"schedule": {
"high": "قد يؤدي إلى تأخيرات كبيرة في المشروع",
"medium": "قد يؤثر على بعض مراحل الجدول الزمني",
"low": "تأثير محدود على الجدول الزمني يمكن استيعابه"
},
"resource": {
"high": "نقص حاد في الموارد الأساسية للمشروع",
"medium": "قد يتطلب موارد إضافية أو بديلة",
"low": "يمكن إدارته من خلال تخطيط الموارد المتاحة"
},
"quality": {
"high": "قد يؤدي إلى مشاكل جودة خطيرة تؤثر على قبول المشروع",
"medium": "يتطلب إجراءات ضمان جودة إضافية",
"low": "تأثير محدود على الجودة يمكن معالجته"
},
"scope": {
"high": "تغييرات جوهرية في نطاق العمل قد تؤثر على المشروع بأكمله",
"medium": "يتطلب تعديلات في بعض جوانب نطاق العمل",
"low": "تغييرات بسيطة في النطاق يمكن استيعابها"
},
"stakeholder": {
"high": "قد يؤثر سلباً على العلاقات مع أصحاب المصلحة الرئيسيين",
"medium": "يتطلب إدارة توقعات أصحاب المصلحة",
"low": "تأثير محدود على رضا أصحاب المصلحة"
},
"commercial": {
"high": "مخاطر تجارية كبيرة قد تؤثر على العلاقات التجارية الرئيسية",
"medium": "قد يتطلب إعادة التفاوض على بعض الشروط التجارية",
"low": "تأثير تجاري محدود يمكن إدارته"
},
"contractual": {
"high": "بنود تعاقدية مجحفة قد تؤثر على التزامات وحقوق الأطراف",
"medium": "يتطلب مراجعة قانونية وتعديل بعض البنود",
"low": "قضايا تعاقدية بسيطة يمكن توضيحها"
},
"regulatory": {
"high": "قد يؤدي إلى مخالفة لوائح تنظيمية هامة",
"medium": "يتطلب تغييرات للامتثال للمتطلبات التنظيمية",
"low": "متطلبات تنظيمية يمكن تلبيتها بسهولة"
}
}
return impact_descriptions.get(category, {}).get(severity, "تأثير غير محدد")
def _generate_recommendation(self, category, term, severity):
"""توليد توصيات لمعالجة المخاطر"""
recommendations = {
"legal": {
"high": "مراجعة قانونية شاملة من محامي متخصص وإعادة التفاوض على البنود المتعلقة بـ'{term}'",
"medium": "مراجعة قانونية والتأكد من الصياغة الدقيقة للبنود المتعلقة بـ'{term}'",
"low": "مراقبة البنود المتعلقة بـ'{term}' أثناء تنفيذ العقد"
},
"financial": {
"high": "إعادة التفاوض على الشروط المالية والتأكد من وجود مخصصات كافية لتغطية المخاطر المتعلقة بـ'{term}'",
"medium": "وضع خطة احتياطية لإدارة التكاليف المرتبطة بـ'{term}'",
"low": "متابعة الجوانب المالية المتعلقة بـ'{term}' بشكل دوري"
},
"operational": {
"high": "وضع خطة تفصيلية لإدارة المخاطر التشغيلية المتعلقة بـ'{term}' وتوفير بدائل",
"medium": "تطوير إجراءات للتعامل مع المشكلات التشغيلية المتعلقة بـ'{term}'",
"low": "متابعة العمليات المتعلقة بـ'{term}' بشكل منتظم"
},
"technical": {
"high": "الاستعانة بخبراء فنيين متخصصين لمراجعة المتطلبات المتعلقة بـ'{term}'",
"medium": "إجراء مراجعة فنية للتأكد من قابلية تنفيذ المتطلبات المتعلقة بـ'{term}'",
"low": "التأكد من وضوح المواصفات الفنية المتعلقة بـ'{term}'"
},
"compliance": {
"high": "مراجعة متخصصة للتأكد من الامتثال للوائح المتعلقة بـ'{term}' وإجراء التعديلات اللازمة",
"medium": "وضع إجراءات للتأكد من الامتثال المستمر للمتطلبات المتعلقة بـ'{term}'",
"low": "متابعة متطلبات الامتثال المتعلقة بـ'{term}' بشكل دوري"
},
"environmental": {
"high": "إجراء تقييم بيئي شامل والتأكد من وجود خطط للتعامل مع المخاطر البيئية المتعلقة بـ'{term}'",
"medium": "مراجعة الإجراءات البيئية المتعلقة بـ'{term}' والتأكد من كفايتها",
"low": "متابعة الجوانب البيئية المتعلقة بـ'{term}' أثناء تنفيذ المشروع"
},
"safety": {
"high": "وضع خطة سلامة شاملة ومراجعتها من قبل متخصصين للتعامل مع المخاطر المتعلقة بـ'{term}'",
"medium": "مراجعة إجراءات السلامة الحالية وتعزيزها للتعامل مع المخاطر المتعلقة بـ'{term}'",
"low": "التأكد من تطبيق إجراءات السلامة القياسية المتعلقة بـ'{term}'"
},
"schedule": {
"high": "إعادة تقييم الجدول الزمني بشكل شامل ووضع خطط بديلة للتعامل مع البنود المتعلقة بـ'{term}'",
"medium": "وضع هوامش زمنية كافية للتعامل مع التأخيرات المحتملة المتعلقة بـ'{term}'",
"low": "مراقبة الجدول الزمني بشكل منتظم فيما يتعلق بـ'{term}'"
},
"resource": {
"high": "وضع خطة شاملة لتأمين الموارد اللازمة والبدائل المتعلقة بـ'{term}'",
"medium": "تحديد مصادر بديلة للموارد المتعلقة بـ'{term}'",
"low": "مراقبة توافر الموارد المتعلقة بـ'{term}' بشكل منتظم"
},
"quality": {
"high": "وضع خطة ضمان جودة شاملة والتأكد من وجود معايير واضحة للجوانب المتعلقة بـ'{term}'",
"medium": "تعزيز إجراءات ضمان الجودة للجوانب المتعلقة بـ'{term}'",
"low": "متابعة معايير الجودة المتعلقة بـ'{term}' بشكل منتظم"
},
"scope": {
"high": "توثيق نطاق العمل بشكل تفصيلي ووضع إجراءات واضحة للتعامل مع التغييرات المتعلقة بـ'{term}'",
"medium": "وضع آلية للتحكم في التغييرات المتعلقة بـ'{term}'",
"low": "مراقبة نطاق العمل بشكل منتظم فيما يتعلق بـ'{term}'"
},
"stakeholder": {
"high": "وضع خطة تواصل شاملة مع أصحاب المصلحة للتعامل مع القضايا المتعلقة بـ'{term}'",
"medium": "تعزيز التواصل مع أصحاب المصلحة المعنيين بـ'{term}'",
"low": "متابعة توقعات وملاحظات أصحاب المصلحة فيما يتعلق بـ'{term}'"
},
"commercial": {
"high": "إعادة التفاوض على الشروط التجارية المتعلقة بـ'{term}' والتأكد من تحقيق توازن المصالح",
"medium": "مراجعة الشروط التجارية المتعلقة بـ'{term}' والتأكد من وضوحها",
"low": "مراقبة تنفيذ الشروط التجارية المتعلقة بـ'{term}'"
},
"contractual": {
"high": "مراجعة قانونية شاملة للبنود التعاقدية المتعلقة بـ'{term}' وإعادة التفاوض عند الضرورة",
"medium": "توضيح وتحسين صياغة البنود المتعلقة بـ'{term}'",
"low": "التأكد من فهم الالتزامات التعاقدية المتعلقة بـ'{term}'"
},
"regulatory": {
"high": "الاستعانة بمستشار متخصص للتأكد من الامتثال للمتطلبات التنظيمية المتعلقة بـ'{term}'",
"medium": "مراجعة المتطلبات التنظيمية الحالية والمستقبلية المتعلقة بـ'{term}'",
"low": "متابعة التغييرات في المتطلبات التنظيمية المتعلقة بـ'{term}'"
}
}
recommendation_template = recommendations.get(category, {}).get(severity, "مراجعة البنود المتعلقة بـ'{term}'")
return recommendation_template.replace('{term}', term)
def _scan_for_risk_patterns(self, contract_text, risk_id_start):
"""فحص النص بحثاً عن صيغ المخاطر المحددة مسبقاً"""
risk_id = risk_id_start
pattern_risks = []
for pattern_type, patterns in self.contract_risk_patterns.items():
for pattern in patterns:
matches = re.finditer(pattern, contract_text)
for match in matches:
match_text = match.group(0)
context = self._extract_context(contract_text, match.start(), window=150)
severity = self._determine_pattern_severity(pattern_type, match_text)
category = self._map_pattern_to_category(pattern_type)
pattern_risks.append({
"id": risk_id,
"term": match_text,
"category": category,
"category_ar": self.risk_categories[category],
"pattern_type": pattern_type,
"context": context,
"severity": severity,
"impact": self._determine_pattern_impact(pattern_type, severity),
"recommendation": self._generate_pattern_recommendation(pattern_type, match_text, severity)
})
risk_id += 1
return pattern_risks
def _determine_pattern_severity(self, pattern_type, match_text):
"""تحديد مستوى خطورة المخاطر بناءً على نوع الصيغة ومحتواها"""
severity_rules = {
"unlimited_liability": "high",
"payment_delay": lambda text: "high" if any(int(n) > 60 for n in re.findall(r'(\d+)', text)) else "medium" if any(int(n) > 30 for n in re.findall(r'(\d+)', text)) else "low",
"excessive_penalties": lambda text: "high" if any(int(n) > 1 for n in re.findall(r'(\d+)', text)) else "medium" if any(int(n) > 0.5 for n in re.findall(r'(\d+)', text)) else "low",
"unilateral_termination": "high",
"unrealistic_deadlines": lambda text: "high" if any(int(n) < 30 for n in re.findall(r'(\d+)', text)) else "medium" if any(int(n) < 60 for n in re.findall(r'(\d+)', text)) else "low",
"scope_creep": "medium",
"indemnification": "high",
"change_control": "high",
"warranty_period": lambda text: "high" if any(int(n) > 24 for n in re.findall(r'(\d+)', text)) else "medium" if any(int(n) > 12 for n in re.findall(r'(\d+)', text)) else "low",
"dispute_resolution": "medium",
"force_majeure": "medium",
"regulatory_compliance": "medium",
"intellectual_property": "high",
"confidentiality": "medium",
"insurance_requirements": "medium"
}
rule = severity_rules.get(pattern_type, "medium")
if callable(rule):
return rule(match_text)
else:
return rule
def _map_pattern_to_category(self, pattern_type):
"""تعيين نوع الصيغة إلى فئة المخاطر المناسبة"""
pattern_category_map = {
"unlimited_liability": "legal",
"payment_delay": "financial",
"excessive_penalties": "financial",
"unilateral_termination": "contractual",
"unrealistic_deadlines": "schedule",
"scope_creep": "scope",
"indemnification": "legal",
"change_control": "scope",
"warranty_period": "quality",
"dispute_resolution": "legal",
"force_majeure": "contractual",
"regulatory_compliance": "compliance",
"intellectual_property": "legal",
"confidentiality": "contractual",
"insurance_requirements": "financial"
}
return pattern_category_map.get(pattern_type, "contractual")
def _determine_pattern_impact(self, pattern_type, severity):
"""تحديد تأثير المخاطر بناءً على نوع الصيغة ومستوى الخطورة"""
pattern_impact = {
"unlimited_liability": {
"high": "يمكن أن يعرض الشركة لمسؤولية مالية وقانونية غير محدودة",
"medium": "قد يؤدي إلى مسؤولية مالية كبيرة غير متوقعة",
"low": "زيادة محتملة في المسؤولية القانونية"
},
"payment_delay": {
"high": "تأخر كبير في الدفعات قد يؤثر على التدفق النقدي والسيولة",
"medium": "قد يتسبب في ضغط على التدفق النقدي",
"low": "تأثير محدود على التدفق النقدي"
},
"excessive_penalties": {
"high": "غرامات تأخير مرتفعة قد تؤثر بشكل كبير على ربحية المشروع",
"medium": "غرامات معتدلة قد تقلل من هامش الربح",
"low": "غرامات محدودة يمكن إدارتها من خلال الجدولة الدقيقة"
},
"unilateral_termination": {
"high": "إمكانية إنهاء العقد من طرف واحد دون تعويض مناسب",
"medium": "شروط إنهاء غير متوازنة قد تتطلب إعادة التفاوض",
"low": "بنود إنهاء تحتاج إلى مراقبة وتوثيق"
},
"unrealistic_deadlines": {
"high": "مواعيد نهائية غير واقعية قد تؤدي إلى فشل المشروع أو غرامات كبيرة",
"medium": "جدول زمني ضيق يتطلب موارد إضافية وإدارة مكثفة",
"low": "مواعيد نهائية تحتاج إلى تخطيط دقيق"
},
"scope_creep": {
"high": "توسع غير محدود في نطاق العمل دون تعديل السعر أو الجدول الزمني",
"medium": "تغييرات محتملة في النطاق تتطلب إدارة دقيقة",
"low": "بعض التعديلات المحتملة على النطاق يمكن إدارتها"
},
"indemnification": {
"high": "التزامات تعويض واسعة النطاق قد تؤدي إلى مسؤولية غير محدودة",
"medium": "شروط تعويض تحتاج إلى مراجعة قانونية",
"low": "التزامات تعويض معقولة تحتاج إلى متابعة"
},
"change_control": {
"high": "عدم وجود آلية واضحة للتحكم في التغييرات وتأثيرها على التكلفة",
"medium": "آلية تغيير غير كافية قد تؤدي إلى نزاعات",
"low": "إجراءات تغيير تحتاج إلى تحسين"
},
"warranty_period": {
"high": "فترة ضمان طويلة غير متناسبة مع طبيعة المشروع",
"medium": "فترة ضمان تتطلب موارد إضافية للدعم",
"low": "فترة ضمان معقولة تحتاج إلى تخطيط"
},
"dispute_resolution": {
"high": "آليات غير مناسبة لحل النزاعات قد تؤدي إلى إجراءات مكلفة",
"medium": "آليات حل النزاعات تحتاج إلى توضيح",
"low": "شروط حل النزاعات تحتاج إلى مراجعة"
},
"force_majeure": {
"high": "تعريف ضيق للقوة القاهرة قد يؤدي إلى مسؤولية غير متوقعة",
"medium": "بنود القوة القاهرة تحتاج إلى توضيح",
"low": "شروط القوة القاهرة معقولة ولكن تحتاج إلى مراقبة"
},
"regulatory_compliance": {
"high": "متطلبات امتثال صارمة قد تزيد التكاليف أو المسؤولية",
"medium": "التزامات الامتثال تحتاج إلى موارد إضافية",
"low": "متطلبات امتثال معقولة تحتاج إلى مراقبة"
},
"intellectual_property": {
"high": "نقل واسع لحقوق الملكية الفكرية دون تعويض مناسب",
"medium": "شروط الملكية الفكرية تحتاج إلى توضيح وتعديل",
"low": "بنود الملكية الفكرية تحتاج إلى مراجعة"
},
"confidentiality": {
"high": "التزامات سرية واسعة وطويلة الأمد قد تقيد النشاط المستقبلي",
"medium": "التزامات السرية تحتاج إلى تحديد نطاق ومدة",
"low": "شروط السرية معقولة ولكن تحتاج إلى مراقبة"
},
"insurance_requirements": {
"high": "متطلبات تأمين مرتفعة قد تزيد التكاليف بشكل كبير",
"medium": "متطلبات التأمين تحتاج إلى مراجعة للتأكد من التناسب",
"low": "متطلبات تأمين معقولة تحتاج إلى التحقق من التوافر"
}
}
return pattern_impact.get(pattern_type, {}).get(severity, "تأثير غير محدد")
def _generate_pattern_recommendation(self, pattern_type, match_text, severity):
"""توليد توصيات لمعالجة المخاطر بناءً على نوع الصيغة"""
pattern_recommendations = {
"unlimited_liability": {
"high": "إعادة التفاوض على بنود المسؤولية وتحديد سقف للتعويضات يتناسب مع قيمة العقد",
"medium": "وضع حدود واضحة للمسؤولية وطلب تعديل البنود المتعلقة بها",
"low": "مراجعة بنود المسؤولية والتأكد من وجود تغطية تأمينية مناسبة"
},
"payment_delay": {
"high": "إعادة التفاوض على شروط الدفع وتقليل فترة السداد، مع إضافة فوائد تأخير",
"medium": "وضع آلية واضحة لمتابعة المدفوعات وتحديد إجراءات التصعيد في حالة التأخر",
"low": "مراقبة مواعيد الدفع والتأكد من إصدار الفواتير في الوقت المناسب"
},
"excessive_penalties": {
"high": "إعادة التفاوض على نسب وآليات غرامات التأخير وربطها بالضرر الفعلي",
"medium": "وضع حد أقصى للغرامات ووضع خطة لتجنب التأخير",
"low": "مراقبة تقدم العمل بدقة للالتزام بالجدول الزمني"
},
"unilateral_termination": {
"high": "تعديل بنود الإنهاء لتكون متوازنة وتضمين تعويض مناسب في حالة الإنهاء",
"medium": "وضع شروط واضحة للإنهاء من كلا الطرفين وتحديد آلية التعويض",
"low": "التأكد من وجود خطة للتعامل مع حالات الإنهاء المحتملة"
},
"unrealistic_deadlines": {
"high": "إعادة التفاوض على الجدول الزمني ليكون واقعياً بناءً على تقييم دقيق للموارد والقدرات",
"medium": "وضع خطة تفصيلية للتنفيذ مع تحديد المراحل الحرجة وتوفير موارد إضافية",
"low": "مراقبة الجدول الزمني بشكل مستمر وتحديد المخاطر المحتملة"
},
"scope_creep": {
"high": "تحديد نطاق العمل بدقة ووضع إجراءات صارمة لإدارة التغييرات مع ربطها بالتكلفة والوقت",
"medium": "وضع آلية واضحة لإدارة التغييرات والتأكد من توثيق نطاق العمل بشكل تفصيلي",
"low": "مراقبة نطاق العمل والتأكد من موافقة جميع الأطراف على أي تغييرات"
},
"indemnification": {
"high": "إعادة التفاوض على بنود التعويض لتكون متوازنة ومحددة بمبلغ يتناسب مع قيمة العقد",
"medium": "تحديد نطاق التعويض وربطه بالأضرار المباشرة والفعلية",
"low": "مراجعة بنود التعويض والتأكد من وجود تغطية تأمينية مناسبة"
},
"change_control": {
"high": "وضع آلية واضحة وصارمة لإدارة التغييرات مع تحديد التأثير على التكلفة والوقت",
"medium": "تحسين إجراءات إدارة التغييرات والتأكد من توثيق جميع التغييرات",
"low": "مراقبة التغييرات والتأكد من الحصول على موافقة مكتوبة قبل التنفيذ"
},
"warranty_period": {
"high": "إعادة التفاوض على فترة الضمان لتكون متناسبة مع طبيعة المشروع والمعايير الصناعية",
"medium": "تحديد نطاق الضمان بوضوح وتخصيص موارد كافية للدعم خلال فترة الضمان",
"low": "وضع خطة لإدارة التزامات الضمان والتأكد من توثيق حالة التسليم"
},
"dispute_resolution": {
"high": "تعديل آليات حل النزاعات لتشمل التفاوض والوساطة قبل اللجوء للتحكيم أو القضاء",
"medium": "توضيح إجراءات حل النزاعات وتحديد الاختصاص القضائي والقانون الواجب التطبيق",
"low": "مراجعة آليات حل النزاعات والتأكد من فهم الإجراءات المتبعة"
},
"force_majeure": {
"high": "توسيع تعريف القوة القاهرة ليشمل الحالات المحتملة وتحديد آلية واضحة للإخطار والتعامل",
"medium": "توضيح إجراءات الإخطار والإجراءات المتبعة في حالات القوة القاهرة",
"low": "مراجعة بنود القوة القاهرة والتأكد من شمولها للحالات المحتملة"
},
"regulatory_compliance": {
"high": "تحديد مسؤوليات كل طرف بوضوح فيما يتعلق بالالتزامات التنظيمية والحصول على المشورة القانونية",
"medium": "مراجعة متطلبات الامتثال والتأكد من القدرة على تلبيتها",
"low": "متابعة التغييرات في المتطلبات التنظيمية والتأكد من الالتزام المستمر"
},
"intellectual_property": {
"high": "إعادة التفاوض على بنود الملكية الفكرية لحماية حقوق الشركة والحصول على تعويض مناسب",
"medium": "توضيح حقوق الملكية الفكرية لكل طرف وتحديد نطاق الاستخدام المسموح به",
"low": "مراجعة بنود الملكية الفكرية والتأكد من حماية الأصول الفكرية للشركة"
},
"confidentiality": {
"high": "تحديد نطاق ومدة التزامات السرية بشكل واضح ومتوازن لتجنب القيود غير الضرورية",
"medium": "توضيح نطاق المعلومات السرية وتحديد مدة معقولة للالتزام بالسرية",
"low": "مراجعة التزامات السرية والتأكد من إمكانية الالتزام بها"
},
"insurance_requirements": {
"high": "إعادة التفاوض على متطلبات التأمين لتكون متناسبة مع طبيعة وحجم المشروع والمخاطر الفعلية",
"medium": "التحقق من توافر وتكلفة التغطية التأمينية المطلوبة والتفاوض على تعديلها إذا لزم الأمر",
"low": "التأكد من توافر التغطية التأمينية المطلوبة والحفاظ على سريانها"
}
}
return pattern_recommendations.get(pattern_type, {}).get(severity, "مراجعة وتعديل البنود المتعلقة بهذه المخاطر")
def _calculate_overall_risk_score(self, risks):
"""حساب درجة المخاطر الإجمالية"""
if not risks:
return 0
# تحويل مستويات الخطورة إلى قيم عددية
severity_scores = {"low": 25, "medium": 50, "high": 90}
# حساب مجموع الأوزان ودرجات المخاطر المرجحة
total_weight = 0
weighted_score_sum = 0
# تجميع المخاطر حسب الفئة
risk_categories = {}
for risk in risks:
category = risk["category"]
severity = risk["severity"]
if category not in risk_categories:
risk_categories[category] = []
risk_categories[category].append(severity_scores[severity])
# حساب متوسط درجة المخاطرة لكل فئة وترجيحها بالوزن
for category, scores in risk_categories.items():
category_weight = self.risk_weights.get(category, 0.5)
category_score = sum(scores) / len(scores)
weighted_score_sum += category_score * category_weight
total_weight += category_weight
# حساب الدرجة الإجمالية المرجحة
if total_weight > 0:
overall_score = int(weighted_score_sum / total_weight)
else:
overall_score = 0
return min(100, max(0, overall_score))
def _determine_overall_severity(self, overall_score):
"""تحديد مستوى الخطورة الإجمالية بناءً على الدرجة الإجمالية"""
for severity, info in self.severity_levels.items():
min_score, max_score = info["score_range"]
if min_score <= overall_score <= max_score:
return {
"level": severity,
"name": info["name"],
"color": info["color"]
}
# القيمة الافتراضية
return {
"level": "low",
"name": "منخفضة",
"color": "#00b894"
}
def _generate_risk_summary(self, risks, overall_score, overall_severity):
"""توليد ملخص للمخاطر المكتشفة"""
if not risks:
return "لم يتم اكتشاف مخاطر كبيرة في العقد."
# تجميع المخاطر حسب الفئة ومستوى الخطورة
risk_categories = {}
for risk in risks:
category = risk["category"]
category_ar = risk["category_ar"]
severity = risk["severity"]
if category not in risk_categories:
risk_categories[category] = {
"name_ar": category_ar,
"high": 0,
"medium": 0,
"low": 0,
"total": 0
}
risk_categories[category][severity] += 1
risk_categories[category]["total"] += 1
# حساب إجماليات المخاطر
total_risks = len(risks)
high_risks = sum(risk_categories[category]["high"] for category in risk_categories)
medium_risks = sum(risk_categories[category]["medium"] for category in risk_categories)
low_risks = sum(risk_categories[category]["low"] for category in risk_categories)
# بناء نص الملخص
summary = f"تم تحديد {total_risks} مخاطر محتملة في العقد مع درجة خطورة إجمالية {overall_score}% ({overall_severity['name']})."
summary += f" وتتضمن {high_risks} مخاطر عالية، و{medium_risks} مخاطر متوسطة، و{low_risks} مخاطر منخفضة."
# ذكر أهم فئات المخاطر
summary += " أهم فئات المخاطر المحددة هي:"
# ترتيب فئات المخاطر حسب الأهمية
sorted_categories = sorted(
risk_categories.items(),
key=lambda x: (x[1]["high"], x[1]["medium"], x[1]["total"]),
reverse=True
)
# إضافة أهم 3 فئات مخاطر إلى الملخص
for i, (category, data) in enumerate(sorted_categories[:3]):
summary += f" {data['name_ar']} ({data['total']} مخاطر، منها {data['high']} عالية)"
if i < 2:
summary += "،"
else:
summary += "."
# إضافة توصية عامة
if high_risks > 0:
summary += " يوصى بمراجعة العقد بشكل دقيق ومناقشة المخاطر العالية مع الأطراف المعنية قبل التوقيع."
elif medium_risks > total_risks / 2:
summary += " يوصى بمراجعة المخاطر المتوسطة وتقييم تأثيرها المحتمل قبل التوقيع."
else:
summary += " يمكن قبول العقد مع مراقبة المخاطر المحددة أثناء التنفيذ."
return summary
def _save_risk_report(self, risk_report, report_name):
"""حفظ تقرير المخاطر كملف JSON"""
filename = f"{report_name.replace(' ', '_')}_risk_report.json"
file_path = os.path.join(self.risk_data_dir, filename)
try:
with open(file_path, 'w', encoding='utf-8') as f:
json.dump(risk_report, f, ensure_ascii=False, indent=2)
except Exception as e:
print(f"خطأ في حفظ تقرير المخاطر: {e}")
def load_risk_report(self, report_name):
"""تحميل تقرير مخاطر محفوظ مسبقاً"""
filename = f"{report_name.replace(' ', '_')}_risk_report.json"
file_path = os.path.join(self.risk_data_dir, filename)
if not os.path.exists(file_path):
return None
try:
with open(file_path, 'r', encoding='utf-8') as f:
risk_report = json.load(f)
return risk_report
except Exception as e:
print(f"خطأ في تحميل تقرير المخاطر: {e}")
return None
def generate_risk_comparison(self, contract_text1, contract_text2, title1="العقد الأول", title2="العقد الثاني"):
"""مقارنة المخاطر بين عقدين"""
# تحليل المخاطر في كل عقد
report1 = self.scan_contract_text(contract_text1, title1)
report2 = self.scan_contract_text(contract_text2, title2)
# مقارنة درجات المخاطر الإجمالية
score_diff = report1["overall_score"] - report2["overall_score"]
# تحديد العقد الأقل مخاطرة
less_risky_contract = title2 if score_diff > 0 else title1
# تجميع المخاطر حسب الفئة لكل عقد
categories1 = self._group_risks_by_category(report1["risks"])
categories2 = self._group_risks_by_category(report2["risks"])
# تحديد الفئات الموجودة في كلا العقدين
all_categories = set(categories1.keys()) | set(categories2.keys())
# مقارنة المخاطر في كل فئة
category_comparison = {}
for category in all_categories:
cat_risks1 = categories1.get(category, {"high": 0, "medium": 0, "low": 0, "total": 0, "name_ar": self.risk_categories.get(category, category)})
cat_risks2 = categories2.get(category, {"high": 0, "medium": 0, "low": 0, "total": 0, "name_ar": self.risk_categories.get(category, category)})
# حساب الفرق في المخاطر العالية والمتوسطة
high_diff = cat_risks1["high"] - cat_risks2["high"]
medium_diff = cat_risks1["medium"] - cat_risks2["medium"]
total_diff = cat_risks1["total"] - cat_risks2["total"]
category_comparison[category] = {
"name_ar": cat_risks1["name_ar"],
"contract1": cat_risks1,
"contract2": cat_risks2,
"high_diff": high_diff,
"medium_diff": medium_diff,
"total_diff": total_diff
}
# تجميع المخاطر المشتركة والفريدة
common_risks = []
unique_risks1 = []
unique_risks2 = []
# تحديد المخاطر المشتركة والفريدة (بناءً على المصطلحات)
terms1 = set(risk["term"] for risk in report1["risks"])
terms2 = set(risk["term"] for risk in report2["risks"])
common_terms = terms1 & terms2
unique_terms1 = terms1 - terms2
unique_terms2 = terms2 - terms1
# تجميع المخاطر المشتركة
for risk in report1["risks"]:
if risk["term"] in common_terms:
common_risks.append({
"term": risk["term"],
"category": risk["category"],
"category_ar": risk["category_ar"],
"contract": title1,
"severity": risk["severity"]
})
for risk in report2["risks"]:
if risk["term"] in common_terms:
common_risks.append({
"term": risk["term"],
"category": risk["category"],
"category_ar": risk["category_ar"],
"contract": title2,
"severity": risk["severity"]
})
# تجميع المخاطر الفريدة
for risk in report1["risks"]:
if risk["term"] in unique_terms1:
unique_risks1.append(risk)
for risk in report2["risks"]:
if risk["term"] in unique_terms2:
unique_risks2.append(risk)
# إنشاء تقرير المقارنة
comparison_report = {
"timestamp": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"contract1": {
"title": title1,
"overall_score": report1["overall_score"],
"overall_severity": report1["overall_severity"],
"risk_count": len(report1["risks"])
},
"contract2": {
"title": title2,
"overall_score": report2["overall_score"],
"overall_severity": report2["overall_severity"],
"risk_count": len(report2["risks"])
},
"score_diff": abs(score_diff),
"less_risky_contract": less_risky_contract,
"category_comparison": category_comparison,
"common_risks": common_risks,
"unique_risks1": unique_risks1,
"unique_risks2": unique_risks2,
"summary": self._generate_comparison_summary(report1, report2, title1, title2, score_diff, category_comparison)
}
return comparison_report
def _group_risks_by_category(self, risks):
"""تجميع المخاطر حسب الفئة"""
categories = {}
for risk in risks:
category = risk["category"]
severity = risk["severity"]
if category not in categories:
categories[category] = {
"high": 0,
"medium": 0,
"low": 0,
"total": 0,
"name_ar": risk["category_ar"]
}
categories[category][severity] += 1
categories[category]["total"] += 1
return categories
def _generate_comparison_summary(self, report1, report2, title1, title2, score_diff, category_comparison):
"""توليد ملخص للمقارنة بين العقدين"""
# تحديد العقد الأقل مخاطرة
less_risky = title1 if score_diff <= 0 else title2
summary = f"مقارنة بين {title1} و{title2} أظهرت فرق في درجة المخاطرة الإجمالية بنسبة {abs(score_diff)}%، حيث كان {less_risky} هو الأقل مخاطرة. "
# تحديد الفئات ذات الاختلافات الكبيرة
significant_diff_categories = []
for category, data in category_comparison.items():
if abs(data["high_diff"]) > 1 or abs(data["total_diff"]) > 3:
significant_diff_categories.append((category, data))
# ترتيب الفئات حسب الاختلاف
significant_diff_categories.sort(key=lambda x: (abs(x[1]["high_diff"]), abs(x[1]["total_diff"])), reverse=True)
# إضافة معلومات عن الفئات ذات الاختلافات الكبيرة
if significant_diff_categories:
summary += "أبرز الاختلافات كانت في فئات: "
for i, (category, data) in enumerate(significant_diff_categories[:3]):
name_ar = data["name_ar"]
more_risky = title1 if data["total_diff"] > 0 else title2
diff = abs(data["total_diff"])
summary += f"{name_ar} (الفرق: {diff} مخاطر لصالح {more_risky})"
if i < len(significant_diff_categories[:3]) - 1:
summary += "، "
else:
summary += ". "
# إضافة توصية
if abs(score_diff) > 20:
summary += f"يوصى بالتفاوض على إعادة صياغة العقد على أساس البنود الأقل مخاطرة من {less_risky}."
elif abs(score_diff) > 10:
summary += f"يوصى بمراجعة البنود المتعلقة بالمخاطر العالية ومقارنتها بين العقدين للتفاوض على تحسينها."
else:
summary += "لا توجد اختلافات كبيرة في المخاطر بين العقدين، ويمكن اختيار أيهما بناءً على معايير أخرى."
return summary
def render_risk_dashboard(self, contract_text, title="العقد"):
"""عرض لوحة معلومات تحليل المخاطر"""
st.markdown("تحليل مخاطر العقود الآلي
", unsafe_allow_html=True)
if not contract_text:
st.warning("يرجى إدخال نص العقد أو تحميل ملف العقد للتحليل")
return
# تحليل العقد
risk_report = self.scan_contract_text(contract_text, title)
# عرض ملخص المخاطر
st.markdown("ملخص تحليل المخاطر
", unsafe_allow_html=True)
col1, col2, col3 = st.columns([1, 1, 1])
with col1:
overall_score = risk_report["overall_score"]
severity = risk_report["overall_severity"]
color = risk_report["severity_color"]
st.markdown(f"""
درجة المخاطرة الإجمالية
{overall_score}%
{severity}
""", unsafe_allow_html=True)
with col2:
high_risks = sum(1 for risk in risk_report["risks"] if risk["severity"] == "high")
medium_risks = sum(1 for risk in risk_report["risks"] if risk["severity"] == "medium")
low_risks = sum(1 for risk in risk_report["risks"] if risk["severity"] == "low")
st.markdown(f"""
توزيع المخاطر
{high_risks} عالية
{medium_risks} متوسطة
{low_risks} منخفضة
""", unsafe_allow_html=True)
with col3:
# تجميع المخاطر حسب الفئة
categories = self._group_risks_by_category(risk_report["risks"])
top_categories = sorted(categories.items(), key=lambda x: x[1]["total"], reverse=True)[:3]
st.markdown("""
أبرز فئات المخاطر
""", unsafe_allow_html=True)
for category, data in top_categories:
st.markdown(f"""
{data['name_ar']}
{data['total']} مخاطر
""", unsafe_allow_html=True)
st.markdown("
", unsafe_allow_html=True)
# عرض ملخص نصي
st.markdown(f"""
{risk_report["summary"]}
""", unsafe_allow_html=True)
# عرض مخطط توزيع المخاطر حسب الفئة والخطورة
st.markdown("توزيع المخاطر حسب الفئة
", unsafe_allow_html=True)
# تجميع البيانات للمخطط
chart_data = []
for category, data in categories.items():
chart_data.append({"category": data["name_ar"], "severity": "عالية", "count": data["high"]})
chart_data.append({"category": data["name_ar"], "severity": "متوسطة", "count": data["medium"]})
chart_data.append({"category": data["name_ar"], "severity": "منخفضة", "count": data["low"]})
# التحقق من وجود بيانات قبل إنشاء الرسم البياني
if chart_data:
chart_df = pd.DataFrame(chart_data)
# إنشاء مخطط باستخدام plotly
fig = px.bar(
chart_df,
x="category",
y="count",
color="severity",
title="توزيع المخاطر حسب الفئة والخطورة",
labels={"category": "فئة المخاطر", "count": "عدد المخاطر", "severity": "مستوى الخطورة"},
color_discrete_map={"عالية": "#d63031", "متوسطة": "#fdcb6e", "منخفضة": "#00b894"}
)
# تنسيق المخطط
fig.update_layout(
barmode='stack',
xaxis={'categoryorder': 'total descending'},
direction='rtl',
font=dict(family="Arial, sans-serif", size=14),
height=400,
margin=dict(l=10, r=10, t=50, b=10)
)
st.plotly_chart(fig, use_container_width=True)
else:
# إذا لم تكن هناك بيانات، عرض رسالة بديلة
st.info("لم يتم العثور على مخاطر كافية لعرض الرسم البياني")
# عرض مخطط دائري لتوزيع مستويات الخطورة
col1, col2 = st.columns([1, 1])
with col1:
severity_counts = {
"عالية": high_risks,
"متوسطة": medium_risks,
"منخفضة": low_risks
}
pie_df = pd.DataFrame({
"مستوى الخطورة": list(severity_counts.keys()),
"العدد": list(severity_counts.values())
})
fig = px.pie(
pie_df,
values="العدد",
names="مستوى الخطورة",
title="توزيع مستويات الخطورة",
color="مستوى الخطورة",
color_discrete_map={"عالية": "#d63031", "متوسطة": "#fdcb6e", "منخفضة": "#00b894"}
)
fig.update_layout(
font=dict(family="Arial, sans-serif", size=14),
height=350
)
st.plotly_chart(fig, use_container_width=True)
with col2:
# تجميع المخاطر حسب النوع (pattern_type)
pattern_types = {}
for risk in risk_report["risks"]:
if "pattern_type" in risk:
pattern_type = risk["pattern_type"]
if pattern_type not in pattern_types:
pattern_types[pattern_type] = 0
pattern_types[pattern_type] += 1
if pattern_types:
pattern_names = {
"unlimited_liability": "مسؤولية غير محدودة",
"payment_delay": "تأخير المدفوعات",
"excessive_penalties": "غرامات مبالغ فيها",
"unilateral_termination": "إنهاء من طرف واحد",
"unrealistic_deadlines": "مواعيد نهائية غير واقعية",
"scope_creep": "توسع نطاق العمل",
"indemnification": "شروط التعويض",
"change_control": "التحكم في التغييرات",
"warranty_period": "فترة الضمان",
"dispute_resolution": "حل النزاعات",
"force_majeure": "القوة القاهرة",
"regulatory_compliance": "الامتثال التنظيمي",
"intellectual_property": "الملكية الفكرية",
"confidentiality": "السرية",
"insurance_requirements": "متطلبات التأمين"
}
pattern_df = pd.DataFrame({
"نوع الصيغة": [pattern_names.get(pt, pt) for pt in pattern_types.keys()],
"العدد": list(pattern_types.values())
})
fig = px.bar(
pattern_df,
x="نوع الصيغة",
y="العدد",
title="أنواع الصيغ التعاقدية المكتشفة",
labels={"نوع الصيغة": "نوع الصيغة", "العدد": "عدد المرات"},
color="العدد",
color_continuous_scale="Viridis"
)
fig.update_layout(
xaxis={'categoryorder': 'total descending'},
direction='rtl',
font=dict(family="Arial, sans-serif", size=14),
height=350
)
st.plotly_chart(fig, use_container_width=True)
else:
st.info("لم يتم اكتشاف صيغ تعاقدية محددة في العقد")
# عرض تفاصيل المخاطر
st.markdown("تفاصيل المخاطر المكتشفة
", unsafe_allow_html=True)
# تصفية المخاطر حسب الخطورة
severity_filter = st.multiselect(
"تصفية حسب مستوى الخطورة",
["عالية", "متوسطة", "منخفضة"],
default=["عالية", "متوسطة", "منخفضة"]
)
severity_map = {"high": "عالية", "medium": "متوسطة", "low": "منخفضة"}
filtered_risks = [risk for risk in risk_report["risks"] if severity_map[risk["severity"]] in severity_filter]
for risk in filtered_risks:
severity = risk["severity"]
severity_class = severity
severity_text = severity_map[severity]
st.markdown(f"""
{risk['category_ar']}
{risk['context']}
التأثير المحتمل:
{risk['impact']}
التوصية:
{risk['recommendation']}
""", unsafe_allow_html=True)
# إضافة CSS لتنسيق الواجهة
st.markdown("""
""", unsafe_allow_html=True)
def render_risk_comparison_dashboard(self, contract_text1, contract_text2, title1="العقد الأول", title2="العقد الثاني"):
"""عرض لوحة معلومات مقارنة المخاطر بين عقدين"""
st.markdown("مقارنة مخاطر العقود
", unsafe_allow_html=True)
if not contract_text1 or not contract_text2:
st.warning("يرجى إدخال نص العقدين للمقارنة")
return
# تحليل العقود ومقارنتها
comparison_report = self.generate_risk_comparison(contract_text1, contract_text2, title1, title2)
# عرض ملخص المقارنة
st.markdown("ملخص المقارنة
", unsafe_allow_html=True)
st.markdown(f"""
{comparison_report["summary"]}
""", unsafe_allow_html=True)
# عرض مقارنة الدرجات الإجمالية
st.markdown("مقارنة درجات المخاطرة الإجمالية
", unsafe_allow_html=True)
col1, col2 = st.columns(2)
with col1:
contract1_score = comparison_report["contract1"]["overall_score"]
contract1_severity = comparison_report["contract1"]["overall_severity"]
st.markdown(f"""
{title1}
{contract1_score}%
{contract1_severity}
{comparison_report["contract1"]["risk_count"]} مخاطر محددة
""", unsafe_allow_html=True)
with col2:
contract2_score = comparison_report["contract2"]["overall_score"]
contract2_severity = comparison_report["contract2"]["overall_severity"]
st.markdown(f"""
{title2}
{contract2_score}%
{contract2_severity}
{comparison_report["contract2"]["risk_count"]} مخاطر محددة
""", unsafe_allow_html=True)
# عرض مخطط مقارنة المخاطر حسب الفئة
st.markdown("مقارنة المخاطر حسب الفئة
", unsafe_allow_html=True)
# تجميع البيانات للمخطط
chart_data = []
for category, data in comparison_report["category_comparison"].items():
chart_data.append({
"category": data["name_ar"],
"contract": title1,
"high": data["contract1"]["high"],
"medium": data["contract1"]["medium"],
"low": data["contract1"]["low"],
"total": data["contract1"]["total"]
})
chart_data.append({
"category": data["name_ar"],
"contract": title2,
"high": data["contract2"]["high"],
"medium": data["contract2"]["medium"],
"low": data["contract2"]["low"],
"total": data["contract2"]["total"]
})
# التحقق من وجود بيانات قبل إنشاء الرسم البياني
if chart_data:
chart_df = pd.DataFrame(chart_data)
# إنشاء مخطط شريطي مقارن
fig = go.Figure()
for contract in [title1, title2]:
contract_data = chart_df[chart_df["contract"] == contract]
fig.add_trace(go.Bar(
x=contract_data["category"],
y=contract_data["total"],
name=contract,
text=contract_data["total"],
textposition="auto"
))
fig.update_layout(
title="مقارنة إجمالي المخاطر حسب الفئة",
xaxis_title="فئة المخاطر",
yaxis_title="عدد المخاطر",
barmode='group',
xaxis={'categoryorder': 'total descending'},
direction='rtl',
font=dict(family="Arial, sans-serif", size=14),
height=500,
margin=dict(l=10, r=10, t=50, b=10)
)
st.plotly_chart(fig, use_container_width=True)
# عرض مخطط مقارنة المخاطر العالية
fig = go.Figure()
for contract in [title1, title2]:
contract_data = chart_df[chart_df["contract"] == contract]
fig.add_trace(go.Bar(
x=contract_data["category"],
y=contract_data["high"],
name=contract,
text=contract_data["high"],
textposition="auto",
marker_color="#d63031" if contract == title1 else "#ff7979"
))
fig.update_layout(
title="مقارنة المخاطر العالية حسب الفئة",
xaxis_title="فئة المخاطر",
yaxis_title="عدد المخاطر العالية",
barmode='group',
xaxis={'categoryorder': 'total descending'},
direction='rtl',
font=dict(family="Arial, sans-serif", size=14),
height=400
)
st.plotly_chart(fig, use_container_width=True)
else:
# إذا لم تكن هناك بيانات، عرض رسالة بديلة
st.info("لم يتم العثور على بيانات كافية للمقارنة وعرض الرسم البياني")
# عرض جدول مقارنة المخاطر المشتركة
st.markdown("المخاطر المشتركة بين العقدين
", unsafe_allow_html=True)
# تنظيم المخاطر المشتركة في جدول
common_risks_grouped = {}
for risk in comparison_report["common_risks"]:
term = risk["term"]
if term not in common_risks_grouped:
common_risks_grouped[term] = {
"term": term,
"category": risk["category_ar"],
title1: None,
title2: None
}
if risk["contract"] == title1:
common_risks_grouped[term][title1] = risk["severity"]
else:
common_risks_grouped[term][title2] = risk["severity"]
# تحويل إلى قائمة
common_risks_list = list(common_risks_grouped.values())
# تصنيف المخاطر المشتركة
severity_colors = {
"high": "❌ عالية",
"medium": "⚠️ متوسطة",
"low": "✓ منخفضة"
}
if common_risks_list:
common_risks_df = pd.DataFrame(common_risks_list)
# تطبيق الألوان والرموز على مستويات الخطورة
for contract in [title1, title2]:
common_risks_df[contract] = common_risks_df[contract].map(lambda x: severity_colors.get(x, "غير محدد") if x else "غير موجود")
st.dataframe(
common_risks_df.style.set_properties(**{'text-align': 'right'}),
use_container_width=True,
height=400
)
else:
st.info("لا توجد مخاطر مشتركة بين العقدين")
# عرض المخاطر الفريدة لكل عقد
col1, col2 = st.columns(2)
with col1:
st.markdown(f"المخاطر الفريدة في {title1}
", unsafe_allow_html=True)
if comparison_report["unique_risks1"]:
for risk in comparison_report["unique_risks1"]:
severity = risk["severity"]
severity_class = severity
severity_text = {"high": "عالية", "medium": "متوسطة", "low": "منخفضة"}[severity]
st.markdown(f"""
{risk['term']}
{risk['category_ar']}
{severity_text}
""", unsafe_allow_html=True)
else:
st.info(f"لا توجد مخاطر فريدة في {title1}")
with col2:
st.markdown(f"المخاطر الفريدة في {title2}
", unsafe_allow_html=True)
if comparison_report["unique_risks2"]:
for risk in comparison_report["unique_risks2"]:
severity = risk["severity"]
severity_class = severity
severity_text = {"high": "عالية", "medium": "متوسطة", "low": "منخفضة"}[severity]
st.markdown(f"""
{risk['term']}
{risk['category_ar']}
{severity_text}
""", unsafe_allow_html=True)
else:
st.info(f"لا توجد مخاطر فريدة في {title2}")
# إضافة CSS لتنسيق الواجهة
st.markdown("""
""", unsafe_allow_html=True)
def render(self):
"""عرض واجهة المستخدم لتحليل مخاطر العقود"""
st.markdown("نظام تقييم مخاطر العقود الآلي
", unsafe_allow_html=True)
st.markdown("""
يمكنك استخدام هذا النظام لتحليل مخاطر العقود بشكل آلي وتحديد البنود التي قد تشكل مخاطر محتملة،
مع تقديم توصيات للتعامل مع هذه المخاطر وتحسين العقود.
""", unsafe_allow_html=True)
# إنشاء علامات تبويب لطرق مختلفة للتحليل
tabs = st.tabs([
"تحليل مخاطر العقد",
"مقارنة بين عقدين",
"تحليل عقد من ملف"
])
with tabs[0]:
st.markdown("### تحليل نص العقد")
contract_title = st.text_input("عنوان العقد")
contract_text = st.text_area("أدخل نص العقد هنا", height=300)
if st.button("تحليل المخاطر"):
if contract_text:
self.render_risk_dashboard(contract_text, contract_title or "العقد")
else:
st.warning("يرجى إدخال نص العقد للتحليل")
with tabs[1]:
st.markdown("### مقارنة المخاطر بين عقدين")
col1, col2 = st.columns(2)
with col1:
contract1_title = st.text_input("عنوان العقد الأول")
contract1_text = st.text_area("أدخل نص العقد الأول", height=200)
with col2:
contract2_title = st.text_input("عنوان العقد الثاني")
contract2_text = st.text_area("أدخل نص العقد الثاني", height=200)
if st.button("مقارنة المخاطر"):
if contract1_text and contract2_text:
self.render_risk_comparison_dashboard(
contract1_text,
contract2_text,
contract1_title or "العقد الأول",
contract2_title or "العقد الثاني"
)
else:
st.warning("يرجى إدخال نص كلا العقدين للمقارنة")
with tabs[2]:
st.markdown("### تحليل عقد من ملف")
uploaded_file = st.file_uploader("قم بتحميل ملف العقد", type=["txt", "docx", "pdf", "md"])
if uploaded_file is not None:
file_title = uploaded_file.name
file_content = ""
if uploaded_file.name.endswith(".pdf"):
st.info("جاري معالجة ملف PDF...")
try:
import fitz # PyMuPDF
pdf_bytes = uploaded_file.read()
with open("temp_contract.pdf", "wb") as f:
f.write(pdf_bytes)
doc = fitz.open("temp_contract.pdf")
for page in doc:
file_content += page.get_text()
except ImportError:
file_content = "تعذر قراءة ملف PDF. يرجى التأكد من تثبيت مكتبة PyMuPDF أو قم بنسخ ولصق محتوى العقد في علامة التبويب الأولى."
elif uploaded_file.name.endswith(".docx"):
st.info("جاري معالجة ملف Word...")
try:
from docx import Document
docx_bytes = uploaded_file.read()
with open("temp_contract.docx", "wb") as f:
f.write(docx_bytes)
doc = Document("temp_contract.docx")
file_content = "\n".join([paragraph.text for paragraph in doc.paragraphs])
except ImportError:
file_content = "تعذر قراءة ملف Word. يرجى التأكد من تثبيت مكتبة python-docx أو قم بنسخ ولصق محتوى العقد في علامة التبويب الأولى."
else: # للملفات النصية
file_content = uploaded_file.read().decode("utf-8")
if file_content:
st.markdown("### محتوى الملف")
with st.expander("عرض محتوى الملف"):
st.text(file_content[:5000] + ("..." if len(file_content) > 5000 else ""))
if st.button("تحليل مخاطر الملف"):
self.render_risk_dashboard(file_content, file_title)
else:
st.warning("تعذر قراءة محتوى الملف. يرجى المحاولة مرة أخرى أو استخدام علامة التبويب الأولى.")
# إضافة CSS للتنسيق
st.markdown("""
""", unsafe_allow_html=True)