Spaces:
Running
Running
import json | |
import os # أعدنا استدعاء هذه المكتبة المهمة | |
import requests | |
from dotenv import load_dotenv | |
import gradio as gr | |
from openai import OpenAI | |
# 1. تحميل الإعدادات الأولية | |
load_dotenv(override=True) | |
# 2. دوال الأدوات لإرسال الإشعارات | |
def push(text): | |
token = os.getenv("PUSHOVER_TOKEN") | |
user = os.getenv("PUSHOVER_USER") | |
if not token or not user: | |
print("!!! تحذير: مفاتيح Pushover غير موجودة.") | |
return | |
try: | |
response = requests.post( | |
"https://api.pushover.net/1/messages.json", | |
data={"token": token, "user": user, "message": text} | |
) | |
response.raise_for_status() | |
print(f"✅ تم إرسال طلب الإشعار بنجاح.") | |
except Exception as e: | |
print(f"❌ فشل إرسال الإشعار: {e}") | |
def record_user_details(email, name="Name not provided", notes="not provided"): | |
push(f"عميل محتمل جديد!\nالاسم: {name}\nالإيميل: {email}\nملاحظات: {notes}") | |
return {"status": "ok", "recorded_email": email} | |
def record_unknown_question(question): | |
push(f"سؤال لم تتم معرفة إجابته:\n{question}") | |
return {"status": "ok", "recorded_question": question} | |
# 3. تعريف الأدوات للـ AI | |
record_user_details_json = { | |
"name": "record_user_details", "description": "استخدم هذه الأداة عندما يعبر المستخدم عن اهتمامه بالتواصل ويوفر عنوان بريده الإلكتروني.", | |
"parameters": {"type": "object", "properties": {"email": {"type": "string"}, "name": {"type": "string"}, "notes": {"type": "string"}}, "required": ["email"]} | |
} | |
record_unknown_question_json = { | |
"name": "record_unknown_question", "description": "استخدم هذه الأداة دائمًا لتسجيل أي سؤال لم تتمكن من الإجابة عليه.", | |
"parameters": {"type": "object", "properties": {"question": {"type": "string"}}, "required": ["question"]} | |
} | |
tools = [{"type": "function", "function": record_user_details_json}, {"type": "function", "function": record_unknown_question_json}] | |
# 4. الفئة الرئيسية التي تمثل شخصيتك | |
class Me: | |
def __init__(self): | |
self.openai = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) | |
self.name = "يوسف" | |
# **-- التصحيح النهائي هنا: البحث عن الملفات داخل مجلد "me" --** | |
try: | |
# هذا السطر يبني المسار الصحيح: "me/summary.txt" | |
with open(os.path.join("me", "summary.txt"), "r", encoding="utf-8") as f: | |
self.summary = f.read() | |
print("✅ تم تحميل 'summary.txt' بنجاح.") | |
except FileNotFoundError: | |
print("❌ فشل: لم يتم العثور على ملف 'summary.txt' داخل مجلد 'me'.") | |
self.summary = "لا يتوفر ملخص حاليًا." | |
try: | |
# هذا السطر يبني المسار الصحيح: "me/professional_info.txt" | |
with open(os.path.join("me", "professional_info.txt"), "r", encoding="utf-8") as f: | |
self.linkedin = f.read() | |
print("✅ تم تحميل 'professional_info.txt' بنجاح.") | |
except FileNotFoundError: | |
print("❌ فشل: لم يتم العثور على ملف 'professional_info.txt' داخل مجلد 'me'.") | |
self.linkedin = "لا تتوفر معلومات مهنية حاليًا." | |
def handle_tool_call(self, tool_calls): | |
results = [] | |
for tool_call in tool_calls: | |
tool_name = tool_call.function.name | |
arguments = json.loads(tool_call.function.arguments) | |
tool_to_call = globals().get(tool_name) | |
if tool_to_call: result = tool_to_call(**arguments) | |
else: result = {"error": f"Tool {tool_name} not found."} | |
results.append({"role": "tool", "content": json.dumps(result), "tool_call_id": tool_call.id}) | |
return results | |
def system_prompt(self): | |
return f""" | |
You are an AI assistant acting as {self.name}. You are professional, helpful, and fluent in both Arabic and English. | |
Your primary goal is to answer questions about {self.name}'s career, skills, and projects based on the provided info. | |
You MUST reply in the same language the user is using. | |
If you don't know the answer, use the `record_unknown_question` tool. | |
If a user provides contact details, use the `record_user_details` tool. | |
## Summary: | |
{self.summary} | |
## Detailed Professional Information: | |
{self.linkedin} | |
""" | |
def chat(self, message, history): | |
messages = [{"role": "system", "content": self.system_prompt()}] + history + [{"role": "user", "content": message}] | |
response = self.openai.chat.completions.create(model="gpt-4o-mini", messages=messages, tools=tools) | |
response_message = response.choices[0].message | |
if response_message.tool_calls: | |
tool_results = self.handle_tool_call(response_message.tool_calls) | |
messages.append(response_message) | |
messages.extend(tool_results) | |
final_response = self.openai.chat.completions.create(model="gpt-4o-mini", messages=messages) | |
final_reply_text = final_response.choices[0].message.content | |
else: | |
final_reply_text = response_message.content | |
if not history: | |
proactive_prompt_ar = "\n\nبالمناسبة، إذا كنت مهتمًا بالتعاون أو التواصل، لا تتردد في ترك اسمك وبريدك الإلكتروني." | |
proactive_prompt_en = "\n\nBy the way, if you are interested in collaborating, feel free to leave your name and email address." | |
try: | |
message.encode('ascii') | |
is_english = True | |
except UnicodeEncodeError: | |
is_english = False | |
if is_english: | |
final_reply_text += proactive_prompt_en | |
else: | |
final_reply_text += proactive_prompt_ar | |
return final_reply_text | |
# 5. جزء التشغيل النهائي | |
if __name__ == "__main__": | |
me = Me() | |
iface = gr.ChatInterface( | |
me.chat, | |
type="messages", | |
title=f"المساعد الذكي لـ {me.name}", | |
description=f"أهلاً بك! أنا مساعد ذكي يمثل {me.name}. يمكنك سؤالي عن مهاراته ومشاريعه.", | |
examples=[["ما هي أهم مهاراتك؟"], ["ما هي مشاريعك؟"], ["أرغب في التواصل معك، هذا بريدي: test@example.com"]], | |
theme="soft" | |
) | |
iface.launch() |