Create utils/api_integrations.py

utils/api_integrations.py

@@ -0,0 +1,615 @@
+import os
+import json
+import requests
+from typing import Dict, List, Any, Union, Tuple, Optional
+from datetime import datetime
+
+class MunafasatAPI:
+    """
+    فئة للاتصال بمنصة المنافسات والمشتريات الحكومية
+    """
+    
+    def __init__(self, api_key: str = None):
+        """
+        تهيئة الاتصال بمنصة المنافسات
+        
+        المعاملات:
+        ----------
+        api_key : str, optional
+            مفتاح API (يمكن تعيينه لاحقاً أو استخدام المفتاح من متغيرات البيئة)
+        """
+        self.api_key = api_key or os.getenv("MUNAFASAT_API_KEY")
+        self.base_url = "https://api.etimad.sa/munafasat/v1"  # عنوان API افتراضي
+        
+        # التحقق من توفر مفتاح API
+        if not self.api_key:
+            print("Warning: MUNAFASAT_API_KEY not provided. Some functions might not work.")
+    
+    def get_tenders(self, params: Dict[str, Any] = None) -> Dict[str, Any]:
+        """
+        الحصول على قائمة المناقصات
+        
+        المعاملات:
+        ----------
+        params : Dict[str, Any], optional
+            معاملات البحث والتصفية
+            
+        المخرجات:
+        --------
+        Dict[str, Any]
+            نتائج البحث
+        """
+        # في التطبيق الفعلي، سيتم استدعاء API
+        # هنا نستخدم بيانات افتراضية للتوضيح
+        
+        # إعداد رأس الطلب
+        headers = {
+            "Authorization": f"Bearer {self.api_key}",
+            "Content-Type": "application/json"
+        }
+        
+        try:
+            # إرسال طلب GET للحصول على المناقصات
+            # response = requests.get(f"{self.base_url}/tenders", headers=headers, params=params)
+            # return response.json()
+            
+            # استخدام بيانات افتراضية للتوضيح
+            return {
+                "success": True,
+                "data": [
+                    {
+                        "id": "T001",
+                        "title": "إنشاء مبنى إداري لوزارة الصحة",
+                        "tender_type": "مناقصة عامة",
+                        "sector": "الإنشاءات",
+                        "agency": "وزارة الصحة",
+                        "publication_date": "2023-09-01",
+                        "submission_deadline": "2023-10-01",
+                        "estimated_value": 15000000,
+                        "status": "نشطة"
+                    },
+                    {
+                        "id": "T002",
+                        "title": "تطوير نظام إدارة المستشفيات المركزي",
+                        "tender_type": "مناقصة عامة",
+                        "sector": "تقنية المعلومات",
+                        "agency": "وزارة الصحة",
+                        "publication_date": "2023-08-15",
+                        "submission_deadline": "2023-09-15",
+                        "estimated_value": 8000000,
+                        "status": "مغلقة"
+                    },
+                    {
+                        "id": "T003",
+                        "title": "توريد وتركيب أجهزة طبية لمستشفيات المنطقة الشرقية",
+                        "tender_type": "مناقصة محدودة",
+                        "sector": "الصحة",
+                        "agency": "وزارة الصحة",
+                        "publication_date": "2023-09-10",
+                        "submission_deadline": "2023-10-10",
+                        "estimated_value": 12000000,
+                        "status": "نشطة"
+                    }
+                ],
+                "meta": {
+                    "total": 3,
+                    "page": 1,
+                    "per_page": 10
+                }
+            }
+        except Exception as e:
+            print(f"Error fetching tenders: {str(e)}")
+            return {
+                "success": False,
+                "error": str(e)
+            }
+    
+    def get_tender_details(self, tender_id: str) -> Dict[str, Any]:
+        """
+        الحصول على تفاصيل مناقصة محددة
+        
+        المعاملات:
+        ----------
+        tender_id : str
+            معرف المناقصة
+            
+        المخرجات:
+        --------
+        Dict[str, Any]
+            تفاصيل المناقصة
+        """
+        # في التطبيق الفعلي، سيتم استدعاء API
+        # هنا نستخدم بيانات افتراضية للتوضيح
+        
+        # إعداد رأس الطلب
+        headers = {
+            "Authorization": f"Bearer {self.api_key}",
+            "Content-Type": "application/json"
+        }
+        
+        try:
+            # إرسال طلب GET للحصول على تفاصيل المناقصة
+            # response = requests.get(f"{self.base_url}/tenders/{tender_id}", headers=headers)
+            # return response.json()
+            
+            # استخدام بيانات افتراضية للتوضيح
+            if tender_id == "T001":
+                return {
+                    "success": True,
+                    "data": {
+                        "id": "T001",
+                        "title": "إنشاء مبنى إداري لوزارة الصحة",
+                        "description": "إنشاء مبنى إداري من 5 طوابق بمساحة إجمالية 10000 متر مربع في مدينة الرياض",
+                        "tender_type": "مناقصة عامة",
+                        "sector": "الإنشاءات",
+                        "agency": "وزارة الصحة",
+                        "publication_date": "2023-09-01",
+                        "submission_deadline": "2023-10-01",
+                        "implementation_duration": "24 شهر",
+                        "estimated_value": 15000000,
+                        "required_classification": "مباني - الدرجة الثانية",
+                        "local_content_requirements": "40% كحد أدنى",
+                        "status": "نشطة",
+                        "documents": [
+                            {"name": "كراسة الشروط والمواصفات", "type": "pdf"},
+                            {"name": "المخططات", "type": "dwg"},
+                            {"name": "جدول الكميات", "type": "xlsx"}
+                        ],
+                        "milestones": [
+                            {"name": "الاجتماع التمهيدي", "date": "2023-09-15"},
+                            {"name": "آخر موعد للاستفسارات", "date": "2023-09-20"},
+                            {"name": "فتح المظاريف", "date": "2023-10-05"}
+                        ]
+                    }
+                }
+            else:
+                return {
+                    "success": False,
+                    "error": "Tender not found"
+                }
+        except Exception as e:
+            print(f"Error fetching tender details: {str(e)}")
+            return {
+                "success": False,
+                "error": str(e)
+            }
+    
+    def submit_bid(self, tender_id: str, bid_data: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        تقديم عرض لمناقصة
+        
+        المعاملات:
+        ----------
+        tender_id : str
+            معرف المناقصة
+        bid_data : Dict[str, Any]
+            بيانات العرض
+            
+        المخرجات:
+        --------
+        Dict[str, Any]
+            نتيجة تقديم العرض
+        """
+        # في التطبيق الفعلي، سيتم استدعاء API
+        # هنا نستخدم بيانات افتراضية للتوضيح
+        
+        # إعداد رأس الطلب
+        headers = {
+            "Authorization": f"Bearer {self.api_key}",
+            "Content-Type": "application/json"
+        }
+        
+        try:
+            # إرسال طلب POST لتقديم العرض
+            # response = requests.post(f"{self.base_url}/tenders/{tender_id}/bids", headers=headers, json=bid_data)
+            # return response.json()
+            
+            # استخدام بيانات افتراضية للتوضيح
+            return {
+                "success": True,
+                "data": {
+                    "bid_id": "B001",
+                    "tender_id": tender_id,
+                    "submission_date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+                    "status": "تم التقديم بنجاح"
+                }
+            }
+        except Exception as e:
+            print(f"Error submitting bid: {str(e)}")
+            return {
+                "success": False,
+                "error": str(e)
+            }
+
+
+class EtimadAPI:
+    """
+    فئة للاتصال بمنصة اعتماد
+    """
+    
+    def __init__(self, api_key: str = None):
+        """
+        تهيئة الاتصال بمنصة اعتماد
+        
+        المعاملات:
+        ----------
+        api_key : str, optional
+            مفتاح API (يمكن تعيينه لاحقاً أو استخدام المفتاح من متغيرات البيئة)
+        """
+        self.api_key = api_key or os.getenv("ETIMAD_API_KEY")
+        self.base_url = "https://api.etimad.sa/v1"  # عنوان API افتراضي
+        
+        # التحقق من توفر مفتاح API
+        if not self.api_key:
+            print("Warning: ETIMAD_API_KEY not provided. Some functions might not work.")
+    
+    def verify_supplier(self, cr_number: str) -> Dict[str, Any]:
+        """
+        التحقق من بيانات مورد
+        
+        ال��عاملات:
+        ----------
+        cr_number : str
+            رقم السجل التجاري
+            
+        المخرجات:
+        --------
+        Dict[str, Any]
+            بيانات المورد
+        """
+        # في التطبيق الفعلي، سيتم استدعاء API
+        # هنا نستخدم بيانات افتراضية للتوضيح
+        
+        # إعداد رأس الطلب
+        headers = {
+            "Authorization": f"Bearer {self.api_key}",
+            "Content-Type": "application/json"
+        }
+        
+        try:
+            # إرسال طلب GET للتحقق من بيانات المورد
+            # response = requests.get(f"{self.base_url}/suppliers/{cr_number}", headers=headers)
+            # return response.json()
+            
+            # استخدام بيانات افتراضية للتوضيح
+            if cr_number == "1010000000":
+                return {
+                    "success": True,
+                    "data": {
+                        "cr_number": "1010000000",
+                        "company_name": "شركة الإنشاءات السعودية",
+                        "establishment_date": "2010-01-01",
+                        "company_type": "ذات مسؤولية محدودة",
+                        "owner_name": "محمد عبدالله",
+                        "sector": "الإنشاءات",
+                        "status": "نشط",
+                        "classification": {
+                            "category": "مباني",
+                            "level": "الدرجة الأولى"
+                        },
+                        "local_content_certificate": {
+                            "status": "معتمد",
+                            "percentage": 65,
+                            "expiry_date": "2024-12-31"
+                        },
+                        "financial_status": {
+                            "credit_rating": "A",
+                            "liquid_assets": 5000000
+                        }
+                    }
+                }
+            else:
+                return {
+                    "success": False,
+                    "error": "Supplier not found"
+                }
+        except Exception as e:
+            print(f"Error verifying supplier: {str(e)}")
+            return {
+                "success": False,
+                "error": str(e)
+            }
+    
+    def get_local_content_certificate(self, cr_number: str) -> Dict[str, Any]:
+        """
+        الحصول على شهادة المحتوى المحلي
+        
+        المعاملات:
+        ----------
+        cr_number : str
+            رقم السجل التجاري
+            
+        المخرجات:
+        --------
+        Dict[str, Any]
+            بيانات شهادة المحتوى المحلي
+        """
+        # في التطبيق الفعلي، سيتم استدعاء API
+        # هنا نستخدم بيانات افتراضية للتوضيح
+        
+        # إعداد رأس الطلب
+        headers = {
+            "Authorization": f"Bearer {self.api_key}",
+            "Content-Type": "application/json"
+        }
+        
+        try:
+            # إرسال طلب GET للحصول على شهادة المحتوى المحلي
+            # response = requests.get(f"{self.base_url}/local-content/{cr_number}", headers=headers)
+            # return response.json()
+            
+            # استخدام بيانات افتراضية للتوضيح
+            if cr_number == "1010000000":
+                return {
+                    "success": True,
+                    "data": {
+                        "cr_number": "1010000000",
+                        "company_name": "شركة الإنشاءات السعودية",
+                        "certificate_number": "LC-2023-1234",
+                        "issue_date": "2023-01-01",
+                        "expiry_date": "2024-12-31",
+                        "percentage": 65,
+                        "status": "معتمد",
+                        "details": {
+                            "manpower": {
+                                "saudi_percentage": 55,
+                                "score": 16.5
+                            },
+                            "local_materials": {
+                                "percentage": 70,
+                                "score": 35.0
+                            },
+                            "local_services": {
+                                "percentage": 45,
+                                "score": 13.5
+                            }
+                        }
+                    }
+                }
+            else:
+                return {
+                    "success": False,
+                    "error": "Local content certificate not found"
+                }
+        except Exception as e:
+            print(f"Error fetching local content certificate: {str(e)}")
+            return {
+                "success": False,
+                "error": str(e)
+            }
+
+
+class BaladyAPI:
+    """
+    فئة للاتصال بمنصة بلدي
+    """
+    
+    def __init__(self, api_key: str = None):
+        """
+        تهيئة الاتصال بمنصة بلدي
+        
+        المعاملات:
+        ----------
+        api_key : str, optional
+            مفتاح API (يمكن تعيينه لاحقاً أو استخدام المفتاح من متغيرات البيئة)
+        """
+        self.api_key = api_key or os.getenv("BALADY_API_KEY")
+        self.base_url = "https://api.balady.gov.sa/v1"  # عنوان API افتراضي
+        
+        # التحقق من توفر مفتاح API
+        if not self.api_key:
+            print("Warning: BALADY_API_KEY not provided. Some functions might not work.")
+    
+    def get_location_info(self, location_id: str) -> Dict[str, Any]:
+        """
+        الحصول على معلومات موقع
+        
+        المعاملات:
+        ----------
+        location_id : str
+            معرف الموقع
+            
+        المخرجات:
+        --------
+        Dict[str, Any]
+            معلومات الموقع
+        """
+        # في التطبيق الفعلي، سيتم استدعاء API
+        # هنا نستخدم بيانات افتراضية للتوضيح
+        
+        # إعداد رأس الطلب
+        headers = {
+            "Authorization": f"Bearer {self.api_key}",
+            "Content-Type": "application/json"
+        }
+        
+        try:
+            # إرسال طلب GET للحصول على معلومات الموقع
+            # response = requests.get(f"{self.base_url}/locations/{location_id}", headers=headers)
+            # return response.json()
+            
+            # استخدام بيانات افتراضية للتوضيح
+            if location_id == "L001":
+                return {
+                    "success": True,
+                    "data": {
+                        "location_id": "L001",
+                        "name": "حي الملك فهد",
+                        "municipality": "أمانة الرياض",
+                        "city": "الرياض",
+                        "coordinates": {
+                            "latitude": 24.7136,
+                            "longitude": 46.6753
+                        },
+                        "land_use": "تجاري وسكني",
+                        "zoning": {
+                            "type": "تجاري",
+                            "max_height": "15 طابق",
+                            "max_build_ratio": 60
+                        },
+                        "services": {
+                            "water": True,
+                            "electricity": True,
+                            "sewage": True,
+                            "internet": True
+                        }
+                    }
+                }
+            else:
+                return {
+                    "success": False,
+                    "error": "Location not found"
+                }
+        except Exception as e:
+            print(f"Error fetching location info: {str(e)}")
+            return {
+                "success": False,
+                "error": str(e)
+            }
+    
+    def get_permit_requirements(self, project_type: str, location_id: str) -> Dict[str, Any]:
+        """
+        الحصول على متطلبات التصاريح
+        
+        المعاملات:
+        ----------
+        project_type : str
+            نوع المشروع
+        location_id : str
+            معرف الموقع
+            
+        المخرجات:
+        --------
+        Dict[str, Any]
+            متطلبات التصاريح
+        """
+        # في التطبيق الفعلي، سيتم استدعاء API
+        # هنا نستخدم بيانات افتراضية للتوضيح
+        
+        # إعداد رأس الطلب
+        headers = {
+            "Authorization": f"Bearer {self.api_key}",
+            "Content-Type": "application/json"
+        }
+        
+        # إعداد معاملات الطلب
+        params = {
+            "project_type": project_type,
+            "location_id": location_id
+        }
+        
+        try:
+            # إرسال طلب GET للحصول على متطلبات التصاريح
+            # response = requests.get(f"{self.base_url}/permits/requirements", headers=headers, params=params)
+            # return response.json()
+            
+            # استخدام بيانات افتراضية للتوضيح
+            if project_type == "مبنى تجاري" and location_id == "L001":
+                return {
+                    "success": True,
+                    "data": {
+                        "project_type": "مبنى تجاري",
+                        "location_id": "L001",
+                        "required_documents": [
+                            {"name": "صك الأرض", "mandatory": True},
+                            {"name": "رخصة البناء", "mandatory": True},
+                            {"name": "المخططات المعتمدة", "mandatory": True},
+                            {"name": "تقرير التربة", "mandatory": True},
+                            {"name": "موافقة الدفاع المدني", "mandatory": True},
+                            {"name": "موافقة شركة الكهرباء", "mandatory": True},
+                            {"name": "موافقة شركة المياه", "mandatory": True}
+                        ],
+                        "regulations": [
+                            "الارتداد الأمامي لا يقل عن 5 متر",
+                            "الارتداد الجانبي لا يقل عن 2 متر",
+                            "الارتداد الخلفي لا يقل عن 3 متر",
+                            "نسبة البناء لا تتجاوز 60% من مساحة الأرض",
+                            "ارتفاع المبنى لا يتجاوز 15 طابق"
+                        ],
+                        "fees": [
+                            {"name": "رسوم إصدار رخصة البناء", "amount": 5000},
+                            {"name": "تأمين نظافة الموقع", "amount": 10000, "refundable": True}
+                        ],
+                        "processing_time": "30 يوم عمل"
+                    }
+                }
+            else:
+                return {
+                    "success": False,
+                    "error": "Requirements not found for the specified project type and location"
+                }
+        except Exception as e:
+            print(f"Error fetching permit requirements: {str(e)}")
+            return {
+                "success": False,
+                "error": str(e)
+            }
+    
+    def check_permit_status(self, permit_id: str) -> Dict[str, Any]:
+        """
+        التحقق من حالة تصريح
+        
+        المعاملات:
+        ----------
+        permit_id : str
+            معرف التصريح
+            
+        المخرجات:
+        --------
+        Dict[str, Any]
+            حالة التصريح
+        """
+        # في التطبيق الفعلي، سيتم استدعاء API
+        # هنا نستخدم بيانات افتراضية للتوضيح
+        
+        # إعداد رأس الطلب
+        headers = {
+            "Authorization": f"Bearer {self.api_key}",
+            "Content-Type": "application/json"
+        }
+        
+        try:
+            # إرسال طلب GET للتحقق من حالة التصريح
+            # response = requests.get(f"{self.base_url}/permits/{permit_id}/status", headers=headers)
+            # return response.json()
+            
+            # استخدام بيانات افتراضية للتوضيح
+            if permit_id == "P001":
+                return {
+                    "success": True,
+                    "data": {
+                        "permit_id": "P001",
+                        "type": "رخصة بناء",
+                        "status": "موافقة",
+                        "issue_date": "2023-05-15",
+                        "expiry_date": "2024-05-15",
+                        "notes": "تم إصدار الرخصة بنجاح"
+                    }
+                }
+            elif permit_id == "P002":
+                return {
+                    "success": True,
+                    "data": {
+                        "permit_id": "P002",
+                        "type": "رخصة بناء",
+                        "status": "قيد المراجعة",
+                        "submission_date": "2023-09-10",
+                        "expected_completion_date": "2023-10-10",
+                        "progress": 60,
+                        "pending_approvals": [
+                            "موافقة الدفاع المدني",
+                            "موافقة شركة الكهرباء"
+                        ]
+                    }
+                }
+            else:
+                return {
+                    "success": False,
+                    "error": "Permit not found"
+                }
+        except Exception as e:
+            print(f"Error checking permit status: {str(e)}")
+            return {
+                "success": False,
+                "error": str(e)
+            }