#!/usr/bin/env python # -*- coding: utf-8 -*- """ تطبيق المساعد الذكي لنظام واهبي لتحليل العقود والمناقصات - الوحدة الرئيسية AI Assistant Main Module for WAHBI Tender Analysis System """ import os import sys import json import time import logging from datetime import datetime from pathlib import Path # تهيئة مكتبات الواجهة import streamlit as st import pandas as pd import numpy as np # مكتبات ذكاء اصطناعي try: import openai from openai import OpenAI openai_available = True except ImportError: openai_available = False logging.warning("لم يتم العثور على مكتبة OpenAI. بعض الوظائف قد لا تعمل.") try: import plotly.express as px import plotly.graph_objects as go plotly_available = True except ImportError: plotly_available = False logging.warning("لم يتم العثور على مكتبة Plotly. بعض الوظائف قد لا تعمل.") # إضافة مسار النظام للوصول للملفات المشتركة sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../.."))) class AIAssistant: """فئة المساعد الذكي""" def __init__(self): """تهيئة المساعد الذكي""" self.initialize_session_state() self.load_conversation_history() self.openai_client = self.initialize_openai_client() def initialize_session_state(self): """تهيئة متغيرات حالة الجلسة""" if 'ai_messages' not in st.session_state: st.session_state.ai_messages = [] if 'waiting_for_answer' not in st.session_state: st.session_state.waiting_for_answer = False if 'api_key_error' not in st.session_state: st.session_state.api_key_error = False if 'assistant_context' not in st.session_state: st.session_state.assistant_context = "عام" # سياق المحادثة الافتراضي def initialize_openai_client(self): """تهيئة اتصال OpenAI""" try: # التحقق من وجود مفتاح API api_key = os.environ.get("OPENAI_API_KEY") if not api_key: st.session_state.api_key_error = True return None # إنشاء عميل OpenAI if openai_available: return OpenAI() return None except Exception as e: logging.error(f"خطأ في تهيئة اتصال OpenAI: {str(e)}") st.session_state.api_key_error = True return None def load_conversation_history(self): """تحميل سجل المحادثات من الملف""" try: history_path = Path("data/assistant_history.json") if history_path.exists(): with open(history_path, "r", encoding="utf-8") as f: history = json.load(f) # تحميل آخر 10 رسائل فقط if 'history' in history and len(history['history']) > 0: st.session_state.ai_messages = history['history'][-10:] except Exception as e: logging.error(f"خطأ في تحميل سجل المحادثات: {str(e)}") def save_conversation_history(self): """حفظ سجل المحادثات إلى ملف""" try: history_path = Path("data/assistant_history.json") history_path.parent.mkdir(exist_ok=True, parents=True) # تحديث أو إنشاء ملف جديد if history_path.exists(): with open(history_path, "r", encoding="utf-8") as f: history = json.load(f) if 'history' in history: # إضافة المحادثة الحالية وحفظ آخر 50 محادثة فقط history['history'] = history['history'] + st.session_state.ai_messages history['history'] = history['history'][-50:] else: history['history'] = st.session_state.ai_messages else: history = {'history': st.session_state.ai_messages} with open(history_path, "w", encoding="utf-8") as f: json.dump(history, f, ensure_ascii=False, indent=2) except Exception as e: logging.error(f"خطأ في حفظ سجل المحادثات: {str(e)}") def render(self): """عرض واجهة المساعد الذكي""" st.markdown("

المساعد الذكي

", unsafe_allow_html=True) st.markdown("""

المساعد الذكي هو واجهة تفاعلية مدعومة بتقنيات الذكاء الاصطناعي لمساعدتك في جميع أنشطة إدارة المشاريع والعقود. يمكنك طرح أسئلة بلغتك الطبيعية والحصول على إجابات فورية، أو طلب مساعدة في مهام محددة مثل تحليل بنود العقد أو تقدير التكاليف.

""", unsafe_allow_html=True) # اختيار سياق المحادثة self._render_context_selector() # عرض محتويات المحادثة self._render_chat_messages() # نموذج إدخال الرسالة self._render_chat_input() # إذا كان هناك مشكلة في مفتاح API if st.session_state.api_key_error: st.error(""" لم يتم العثور على مفتاح API صالح لخدمة OpenAI. يرجى التواصل مع مسؤول النظام لتكوين المفتاح بشكل صحيح. """) # تنبيه: الوحدة قيد التطوير st.info(""" المساعد الذكي في الوضع التجريبي. بعض الميزات قد لا تعمل كما هو متوقع. سيتم إضافة المزيد من القدرات والتكامل مع النماذج المتقدمة مثل Claude-3.7 في الإصدارات القادمة. """) def _render_context_selector(self): """عرض محدد سياق المحادثة""" st.markdown("### سياق المحادثة") col1, col2, col3 = st.columns(3) with col1: if st.button("عام", key="general_context", use_container_width=True, type="primary" if st.session_state.assistant_context == "عام" else "secondary"): st.session_state.assistant_context = "عام" st.rerun() with col2: if st.button("تحليل العقود", key="contract_context", use_container_width=True, type="primary" if st.session_state.assistant_context == "تحليل العقود" else "secondary"): st.session_state.assistant_context = "تحليل العقود" st.rerun() with col3: if st.button("حساب التكاليف", key="cost_context", use_container_width=True, type="primary" if st.session_state.assistant_context == "حساب التكاليف" else "secondary"): st.session_state.assistant_context = "حساب التكاليف" st.rerun() st.markdown("
", unsafe_allow_html=True) def _render_chat_messages(self): """عرض رسائل المحادثة""" st.markdown("### المحادثة") # عرض رسائل المحادثة for message in st.session_state.ai_messages: if message["role"] == "user": st.markdown(f"""

{message["content"]}

{message.get("time", "")}
""", unsafe_allow_html=True) else: st.markdown(f"""

{message["content"]}

{message.get("time", "")}
""", unsafe_allow_html=True) # إذا كان في انتظار الإجابة if st.session_state.waiting_for_answer: with st.spinner("جاري التفكير..."): st.markdown("""
""", unsafe_allow_html=True) def _render_chat_input(self): """عرض حقل إدخال المحادثة""" # مربع الإدخال user_input = st.text_area("اكتب سؤالك هنا", key="user_question", height=100) col1, col2 = st.columns([4, 1]) with col1: # تحديد أزرار السياق السريع st.markdown("#### اختصارات سريعة:") quick_buttons = st.columns(4) quick_prompts = [ "كيف يمكنني تحليل مناقصة؟", "كيف أقوم بتقدير تكاليف مشروع؟", "ما هي المخاطر التعاقدية؟", "كيف أستخدم النظام؟" ] for i, prompt in enumerate(quick_prompts): with quick_buttons[i]: if st.button(prompt, key=f"quick_{i}", use_container_width=True): # إرسال السؤال السريع self.process_user_message(prompt) st.rerun() with col2: st.markdown("
", unsafe_allow_html=True) # زر الإرسال if st.button("إرسال", key="send_button", use_container_width=True, type="primary"): if user_input.strip(): self.process_user_message(user_input) st.rerun() def process_user_message(self, message): """معالجة رسالة المستخدم""" if not message.strip(): return # إضافة رسالة المستخدم إلى المحادثة current_time = datetime.now().strftime("%H:%M") st.session_state.ai_messages.append({ "role": "user", "content": message, "time": current_time }) # تعيين حالة الانتظار st.session_state.waiting_for_answer = True # الحصول على سياق المحادثة الحالي context = st.session_state.assistant_context # الحصول على رد من API (معالجة غير متزامنة) try: # فحص توفر OpenAI if not openai_available or not self.openai_client: response = self.get_fallback_response(context, message) else: response = self.get_ai_response(context, message) # إضافة رد المساعد إلى المحادثة st.session_state.ai_messages.append({ "role": "assistant", "content": response, "time": datetime.now().strftime("%H:%M") }) # حفظ المحادثة self.save_conversation_history() except Exception as e: logging.error(f"خطأ في معالجة رسالة المستخدم: {str(e)}") st.session_state.ai_messages.append({ "role": "assistant", "content": "عذراً، حدث خطأ أثناء معالجة طلبك. يرجى المحاولة مرة أخرى لاحقاً.", "time": datetime.now().strftime("%H:%M") }) # إلغاء حالة الانتظار st.session_state.waiting_for_answer = False def get_ai_response(self, context, message): """الحصول على رد من OpenAI API""" try: # إنشاء رسالة النظام بناءً على السياق system_message = self.get_system_message(context) # تحضير سجل المحادثة للإرسال إلى OpenAI messages = [{"role": "system", "content": system_message}] # إضافة آخر 5 رسائل من المحادثة الحالية history = [msg for msg in st.session_state.ai_messages[-10:] if msg["role"] in ["user", "assistant"]] for msg in history: messages.append({"role": msg["role"], "content": msg["content"]}) # إضافة رسالة المستخدم الحالية إذا لم تكن موجودة بالفعل if history and history[-1]["role"] != "user": messages.append({"role": "user", "content": message}) # استدعاء OpenAI API response = self.openai_client.chat.completions.create( model="gpt-4o", messages=messages, temperature=0.7, max_tokens=1000 ) return response.choices[0].message.content except Exception as e: logging.error(f"خطأ في الحصول على رد من OpenAI: {str(e)}") return self.get_fallback_response(context, message) def get_system_message(self, context): """إنشاء رسالة النظام بناءً على سياق المحادثة""" if context == "تحليل العقود": return """ أنت مساعد متخصص في تحليل العقود والمناقصات في مجال المقاولات والإنشاءات. ساعد المستخدم في فهم البنود القانونية وتحديد المخاطر المحتملة وتقديم النصائح حول تحسين الشروط. استخدم لغة واضحة ومهنية باللغة العربية واذكر المصطلحات الفنية والقانونية المناسبة. """ elif context == "حساب التكاليف": return """ أنت مساعد متخصص في تقدير تكاليف مشاريع البناء والمقاولات. ساعد المستخدم في فهم كيفية حساب تكاليف المواد والعمالة والمعدات، وكيفية تسعير بنود المشروع بشكل صحيح. قدم معلومات حول طرق التسعير المختلفة مثل وحدة القياس والمبلغ المقطوع. استخدم لغة واضحة ومهنية باللغة العربية واذكر المصطلحات الفنية المناسبة. """ else: # سياق عام return """ أنت مساعد ذكي متخصص في مجال المقاولات والإنشاءات باسم "واهبي" من نظام تحليل العقود والمناقصات. ساعد المستخدم بالإجابة على أسئلته حول إدارة المشاريع وتحليل العقود وتقدير التكاليف والمناقصات. تجنب التعليق على مواضيع لا علاقة لها بالمقاولات والإنشاءات. استخدم لغة واضحة ومهنية باللغة العربية. """ def get_fallback_response(self, context, message): """الحصول على رد احتياطي في حالة عدم إمكانية الوصول إلى API""" responses = { "تحليل العقود": """ مرحباً! أنا مساعدك في تحليل العقود والمناقصات. للأسف، هناك مشكلة في الاتصال بخدمة الذكاء الاصطناعي حالياً. للمساعدة في تحليل العقود، يمكنك استخدام وحدة تحليل المستندات في النظام، والتي تتيح لك: - تحميل ملفات العقود بصيغة PDF - تحليل البنود والشروط تلقائياً - استخراج الالتزامات والمواعيد النهائية - تحليل المخاطر المحتملة يمكنك الوصول إلى هذه الوحدة من القائمة الجانبية تحت "تحليل المستندات". """, "حساب التكاليف": """ مرحباً! أنا مساعدك في تقدير تكاليف المشاريع. للأسف، هناك مشكلة في الاتصال بخدمة الذكاء الاصطناعي حالياً. للمساعدة في حساب التكاليف، يمكنك استخدام وحدة التسعير المتكاملة في النظام، والتي تتيح لك: - إنشاء نماذج تسعير متكاملة - حساب تكاليف المواد والعمالة والمعدات - تطبيق نسب الأرباح والمصاريف الإدارية - إنشاء جداول كميات تفصيلية يمكنك الوصول إلى هذه الوحدة من القائمة الجانبية تحت "التسعير المتكاملة". """, "عام": """ مرحباً! أنا مساعدك في نظام تحليل العقود والمناقصات. للأسف، هناك مشكلة في الاتصال بخدمة الذكاء الاصطناعي حالياً. يمكنك الاستفادة من الوظائف المتاحة في النظام مثل: - تحليل المستندات لفهم العقود والمناقصات - التسعير المتكاملة لحساب تكاليف المشاريع - إدارة المشاريع لمتابعة سير العمل - تقييم مخاطر العقود لتقليل المخاطر المحتملة يمكنك الوصول إلى هذه الوحدات من القائمة الجانبية. """ } return responses.get(context, responses["عام"]) # مثال للاستخدام المستقل للوحدة if __name__ == "__main__": st.set_page_config( page_title="المساعد الذكي | WAHBi AI", page_icon="🤖", layout="wide", initial_sidebar_state="expanded" ) # استيراد مكونات واجهة المستخدم from utils.components.sidebar import render_sidebar # عرض الشريط الجانبي render_sidebar() # عرض واجهة المساعد الذكي assistant = AIAssistant() assistant.render()