| import os |
| import requests |
| import telebot |
| from fastapi import FastAPI, Request |
|
|
| |
| HF_TOKEN = os.getenv("HF_TOKEN") |
| TELEGRAM_TOKEN = os.getenv("TELEGRAM_TOKEN") |
| SPACE_ID = os.getenv("SPACE_ID") |
|
|
| bot = telebot.TeleBot(TELEGRAM_TOKEN) |
| app = FastAPI() |
|
|
| |
| if SPACE_ID: |
| USER, SPACE = SPACE_ID.split('/') |
| WEBHOOK_URL = f"https://{USER.lower()}-{SPACE.lower()}.hf.space/webhook" |
| else: |
| WEBHOOK_URL = None |
|
|
| def get_ai_response(user_message): |
| headers = {"Authorization": f"Bearer {HF_TOKEN}"} |
| api_url = "https://api-inference.huggingface.co/models/meta-llama/Meta-Llama-3-8B-Instruct/v1/chat/completions" |
| data = {"messages": [{"role": "user", "content": user_message}], "max_tokens": 1000} |
| try: |
| response = requests.post(api_url, headers=headers, json=data) |
| if response.status_code == 200: |
| return response.json()["choices"][0]["message"]["content"] |
| return "النموذج بيحمل دلوقتي، جرب تاني كمان ثواني." |
| except: |
| return "حصل خطأ في الاتصال بالذكاء الاصطناعي." |
|
|
| |
| @app.on_event("startup") |
| def on_startup(): |
| if WEBHOOK_URL: |
| bot.remove_webhook() |
| bot.set_webhook(url=WEBHOOK_URL) |
|
|
| |
| @app.post("/webhook") |
| async def webhook(request: Request): |
| update = await request.json() |
| update_obj = telebot.types.Update.de_json(update) |
| bot.process_new_updates([update_obj]) |
| return {"status": "ok"} |
|
|
| |
| @app.get("/") |
| def read_root(): |
| return "البوت شغال ومربوط بتليجرام بنجاح! 🚀" |
|
|
| @bot.message_handler(commands=['start']) |
| def send_welcome(message): |
| bot.reply_to(message, "أهلاً بيك يا غالي! أنا بوت ذكاء اصطناعي، اسألني في أي حاجة.") |
|
|
| @bot.message_handler(func=lambda message: True) |
| def handle_message(message): |
| msg = bot.reply_to(message, "ثواني بجمع الإجابة... ⏳") |
| ai_response = get_ai_response(message.text) |
| bot.edit_message_text(chat_id=message.chat.id, message_id=msg.message_id, text=ai_response) |
