""" وحدة تكامل البيانات مع الذكاء الاصطناعي هذا الملف يحتوي على الفئات والدوال اللازمة لتكامل وحدة تحليل البيانات مع وحدة الذكاء الاصطناعي. """ import pandas as pd import numpy as np import matplotlib.pyplot as plt import plotly.express as px import plotly.graph_objects as go from datetime import datetime import json import os import sys from pathlib import Path # إضافة المسار للوصول إلى وحدة تحليل البيانات current_dir = os.path.dirname(os.path.abspath(__file__)) parent_dir = os.path.dirname(os.path.dirname(current_dir)) if parent_dir not in sys.path: sys.path.append(parent_dir) # محاولة استيراد وحدة تحليل البيانات try: from modules.data_analysis.data_analysis_app import DataAnalysisApp except ImportError: # تعريف فئة بديلة في حالة فشل الاستيراد class DataAnalysisApp: def __init__(self): pass def run(self): pass class DataAIIntegration: """فئة تكامل البيانات مع الذكاء الاصطناعي""" def __init__(self): """تهيئة فئة تكامل البيانات مع الذكاء الاصطناعي""" self.data_analysis_app = DataAnalysisApp() def analyze_tender_data(self, tender_data): """ تحليل بيانات المناقصة باستخدام الذكاء الاصطناعي المعلمات: tender_data (dict): بيانات المناقصة العوائد: dict: نتائج التحليل """ # تحويل البيانات إلى DataFrame if isinstance(tender_data, dict): df = pd.DataFrame([tender_data]) elif isinstance(tender_data, list): df = pd.DataFrame(tender_data) else: df = tender_data # تحليل البيانات results = { 'summary': self._generate_summary(df), 'recommendations': self._generate_recommendations(df), 'risk_analysis': self._analyze_risks(df), 'cost_analysis': self._analyze_costs(df), 'competitive_analysis': self._analyze_competition(df) } return results def analyze_historical_data(self, project_type=None, location=None, time_period=None): """ تحليل البيانات التاريخية للمناقصات المعلمات: project_type (str): نوع المشروع (اختياري) location (str): الموقع (اختياري) time_period (str): الفترة الزمنية (اختياري) العوائد: dict: نتائج التحليل """ # الحصول على البيانات التاريخية (محاكاة) historical_data = self._get_historical_data() # تطبيق التصفية إذا تم تحديدها filtered_data = historical_data.copy() if project_type: filtered_data = filtered_data[filtered_data['نوع المشروع'] == project_type] if location: filtered_data = filtered_data[filtered_data['الموقع'] == location] if time_period: # تنفيذ تصفية الفترة الزمنية (محاكاة) pass # تحليل البيانات results = { 'win_rate': self._calculate_win_rate(filtered_data), 'avg_profit_margin': self._calculate_avg_profit_margin(filtered_data), 'price_trends': self._analyze_price_trends(filtered_data), 'success_factors': self._identify_success_factors(filtered_data), 'visualizations': self._generate_visualizations(filtered_data) } return results def predict_tender_success(self, tender_data): """ التنبؤ بفرص نجاح المناقصة المعلمات: tender_data (dict): بيانات المناقصة العوائد: dict: نتائج التنبؤ """ # تحويل البيانات إلى DataFrame if isinstance(tender_data, dict): df = pd.DataFrame([tender_data]) elif isinstance(tender_data, list): df = pd.DataFrame(tender_data) else: df = tender_data # تنفيذ التنبؤ (محاكاة) success_probability = np.random.uniform(0, 100) # تحديد العوامل المؤثرة (محاكاة) factors = [ {'name': 'السعر التنافسي', 'impact': np.random.uniform(0, 1), 'direction': 'إيجابي' if np.random.random() > 0.5 else 'سلبي'}, {'name': 'الخبرة السابقة', 'impact': np.random.uniform(0, 1), 'direction': 'إيجابي' if np.random.random() > 0.5 else 'سلبي'}, {'name': 'الجودة الفنية', 'impact': np.random.uniform(0, 1), 'direction': 'إيجابي' if np.random.random() > 0.5 else 'سلبي'}, {'name': 'المدة الزمنية', 'impact': np.random.uniform(0, 1), 'direction': 'إيجابي' if np.random.random() > 0.5 else 'سلبي'}, {'name': 'المنافسة', 'impact': np.random.uniform(0, 1), 'direction': 'إيجابي' if np.random.random() > 0.5 else 'سلبي'} ] # ترتيب العوامل حسب التأثير factors = sorted(factors, key=lambda x: x['impact'], reverse=True) # إعداد النتائج results = { 'success_probability': success_probability, 'confidence': np.random.uniform(70, 95), 'factors': factors, 'recommendations': self._generate_success_recommendations(factors) } return results def optimize_pricing(self, tender_data, competitors_data=None): """ تحسين التسعير للمناقصة المعلمات: tender_data (dict): بيانات المناقصة competitors_data (list): بيانات المنافسين (اختياري) العوائد: dict: نتائج التحسين """ # تحويل البيانات إلى DataFrame if isinstance(tender_data, dict): df = pd.DataFrame([tender_data]) elif isinstance(tender_data, list): df = pd.DataFrame(tender_data) else: df = tender_data # تحليل بيانات المنافسين إذا كانت متوفرة if competitors_data: competitors_df = pd.DataFrame(competitors_data) else: # استخدام بيانات افتراضية للمنافسين competitors_df = self._get_competitors_data() # تنفيذ تحسين التسعير (محاكاة) base_price = float(df['الميزانية التقديرية'].iloc[0]) if 'الميزانية التقديرية' in df.columns else 10000000 # حساب نطاق السعر المقترح min_price = base_price * 0.85 optimal_price = base_price * 0.92 max_price = base_price * 0.98 # تحليل حساسية السعر price_sensitivity = [] for price_factor in np.linspace(0.8, 1.1, 7): price = base_price * price_factor win_probability = max(0, min(100, 100 - (price_factor - 0.9) * 200)) profit = price - (base_price * 0.75) expected_value = win_probability / 100 * profit price_sensitivity.append({ 'price_factor': price_factor, 'price': price, 'win_probability': win_probability, 'profit': profit, 'expected_value': expected_value }) # إعداد النتائج results = { 'min_price': min_price, 'optimal_price': optimal_price, 'max_price': max_price, 'price_sensitivity': price_sensitivity, 'market_position': self._analyze_market_position(optimal_price, competitors_df), 'recommendations': self._generate_pricing_recommendations(optimal_price, price_sensitivity) } return results def analyze_dwg_files(self, file_path): """ تحليل ملفات DWG باستخدام الذكاء الاصطناعي المعلمات: file_path (str): مسار ملف DWG العوائد: dict: نتائج التحليل """ # محاكاة تحليل ملف DWG results = { 'file_name': os.path.basename(file_path), 'file_size': f"{np.random.randint(1, 10)} MB", 'elements_count': np.random.randint(100, 1000), 'layers_count': np.random.randint(5, 20), 'dimensions': { 'width': f"{np.random.randint(10, 100)} م", 'height': f"{np.random.randint(10, 100)} م", 'area': f"{np.random.randint(100, 10000)} م²" }, 'elements': { 'walls': np.random.randint(10, 100), 'doors': np.random.randint(5, 50), 'windows': np.random.randint(5, 50), 'columns': np.random.randint(5, 50), 'stairs': np.random.randint(1, 10) }, 'materials': [ {'name': 'خرسانة', 'volume': f"{np.random.randint(10, 1000)} م³"}, {'name': 'حديد', 'weight': f"{np.random.randint(1, 100)} طن"}, {'name': 'طابوق', 'count': f"{np.random.randint(1000, 10000)} قطعة"}, {'name': 'زجاج', 'area': f"{np.random.randint(10, 1000)} م²"}, {'name': 'خشب', 'volume': f"{np.random.randint(1, 50)} م³"} ], 'cost_estimate': { 'materials': np.random.randint(100000, 1000000), 'labor': np.random.randint(50000, 500000), 'equipment': np.random.randint(10000, 100000), 'total': np.random.randint(200000, 2000000) }, 'recommendations': [ 'يمكن تقليل تكلفة المواد باستخدام بدائل أقل تكلفة', 'يمكن تحسين كفاءة استخدام المساحة', 'يمكن تقليل عدد الأعمدة لتوفير التكلفة', 'يمكن تحسين تصميم السلالم لزيادة السلامة', 'يمكن تحسين توزيع النوافذ لزيادة الإضاءة الطبيعية' ] } return results def integrate_with_ai_assistant(self, ai_assistant): """ تكامل وحدة تحليل البيانات مع وحدة الذكاء الاصطناعي المعلمات: ai_assistant: كائن وحدة الذكاء الاصطناعي العوائد: bool: نجاح التكامل """ try: # إضافة وظائف تحليل البيانات إلى وحدة الذكاء الاصطناعي ai_assistant.data_integration = self # إضافة دوال التحليل إلى وحدة الذكاء الاصطناعي ai_assistant.analyze_tender_data = self.analyze_tender_data ai_assistant.analyze_historical_data = self.analyze_historical_data ai_assistant.predict_tender_success = self.predict_tender_success ai_assistant.optimize_pricing = self.optimize_pricing ai_assistant.analyze_dwg_files = self.analyze_dwg_files return True except Exception as e: print(f"خطأ في تكامل وحدة تحليل البيانات مع وحدة الذكاء الاصطناعي: {str(e)}") return False # دوال مساعدة داخلية def _get_historical_data(self): """الحصول على البيانات التاريخية""" # محاكاة البيانات التاريخية np.random.seed(42) n_tenders = 50 tender_ids = [f"T-{2021 + i//20}-{i%20 + 1:03d}" for i in range(n_tenders)] tender_types = np.random.choice(["مبنى إداري", "مبنى سكني", "مدرسة", "مستشفى", "طرق", "جسور", "بنية تحتية"], n_tenders) tender_locations = np.random.choice(["الرياض", "جدة", "الدمام", "مكة", "المدينة", "أبها", "تبوك"], n_tenders) tender_areas = np.random.randint(1000, 10000, n_tenders) tender_durations = np.random.randint(6, 36, n_tenders) tender_budgets = np.random.randint(1000000, 50000000, n_tenders) tender_costs = np.array([budget * np.random.uniform(0.8, 1.1) for budget in tender_budgets]) tender_profits = tender_budgets - tender_costs tender_profit_margins = tender_profits / tender_budgets * 100 tender_statuses = np.random.choice(["فائز", "خاسر", "قيد التنفيذ", "منجز"], n_tenders) tender_dates = [f"202{1 + i//20}-{np.random.randint(1, 13):02d}-{np.random.randint(1, 29):02d}" for i in range(n_tenders)] # إنشاء DataFrame للمناقصات السابقة tenders_data = { "رقم المناقصة": tender_ids, "نوع المشروع": tender_types, "الموقع": tender_locations, "المساحة (م2)": tender_areas, "المدة (شهر)": tender_durations, "الميزانية (ريال)": tender_budgets, "التكلفة (ريال)": tender_costs, "الربح (ريال)": tender_profits, "هامش الربح (%)": tender_profit_margins, "الحالة": tender_statuses, "تاريخ التقديم": tender_dates } return pd.DataFrame(tenders_data) def _get_competitors_data(self): """الحصول على بيانات المنافسين""" # محاكاة بيانات المنافسين n_competitors = 10 competitor_ids = [f"C-{i+1:02d}" for i in range(n_competitors)] competitor_names = [ "شركة الإنشاءات المتطورة", "شركة البناء الحديث", "شركة التطوير العمراني", "شركة الإعمار الدولية", "شركة البنية التحتية المتكاملة", "شركة المقاولات العامة", "شركة التشييد والبناء", "شركة الهندسة والإنشاءات", "شركة المشاريع الكبرى", "شركة التطوير العقاري" ] competitor_specialties = np.random.choice(["مباني", "طرق", "جسور", "بنية تحتية", "متعددة"], n_competitors) competitor_sizes = np.random.choice(["صغيرة", "متوسطة", "كبيرة"], n_competitors) competitor_market_shares = np.random.uniform(1, 15, n_competitors) competitor_win_rates = np.random.uniform(10, 60, n_competitors) competitor_avg_margins = np.random.uniform(5, 20, n_competitors) # إنشاء DataFrame للمنافسين competitors_data = { "رمز المنافس": competitor_ids, "اسم المنافس": competitor_names, "التخصص": competitor_specialties, "الحجم": competitor_sizes, "حصة السوق (%)": competitor_market_shares, "معدل الفوز (%)": competitor_win_rates, "متوسط هامش الربح (%)": competitor_avg_margins } return pd.DataFrame(competitors_data) def _generate_summary(self, df): """توليد ملخص للبيانات""" # محاكاة توليد ملخص return "تحليل البيانات يشير إلى أن هذه المناقصة تتعلق بمشروع إنشائي متوسط الحجم. تتضمن المناقصة متطلبات فنية متوسطة المستوى وشروط تعاقدية معيارية. بناءً على البيانات التاريخية، هناك فرصة جيدة للفوز بهذه المناقصة إذا تم تقديم عرض تنافسي مع التركيز على الجوانب الفنية والجودة." def _generate_recommendations(self, df): """توليد توصيات بناءً على البيانات""" # محاكاة توليد توصيات return [ "تقديم عرض سعر تنافسي يقل بنسبة 5-10% عن الميزانية التقديرية", "التركيز على الخبرات السابقة في مشاريع مماثلة", "تقديم حلول مبتكرة لتقليل مدة التنفيذ", "تعزيز الجوانب الفنية في العرض", "تقديم خطة تنفيذ مفصلة مع جدول زمني واضح" ] def _analyze_risks(self, df): """تحليل المخاطر""" # محاكاة تحليل المخاطر return [ {"risk": "ارتفاع أسعار المواد", "probability": "متوسطة", "impact": "عالي", "mitigation": "تثبيت أسعار المواد الرئيسية مع الموردين"}, {"risk": "تأخر التنفيذ", "probability": "متوسطة", "impact": "عالي", "mitigation": "وضع خطة تنفيذ مفصلة مع هوامش زمنية"}, {"risk": "نقص العمالة الماهرة", "probability": "منخفضة", "impact": "متوسط", "mitigation": "التعاقد المسبق مع مقاولي الباطن"}, {"risk": "تغيير نطاق العمل", "probability": "متوسطة", "impact": "عالي", "mitigation": "توثيق نطاق العمل بدقة وتحديد إجراءات التغيير"}, {"risk": "مشاكل في التربة", "probability": "منخفضة", "impact": "عالي", "mitigation": "إجراء فحوصات شاملة للتربة قبل البدء"} ] def _analyze_costs(self, df): """تحليل التكاليف""" # محاكاة تحليل التكاليف total_budget = float(df['الميزانية التقديرية'].iloc[0]) if 'الميزانية التقديرية' in df.columns else 10000000 # توزيع التكاليف materials_cost = total_budget * 0.6 labor_cost = total_budget * 0.25 equipment_cost = total_b (Content truncated due to size limit. Use line ranges to read in chunks)