Update app.py

app.py

@@ -1,59 +1,82 @@
 import os
 import logging
+import numpy as np
+import torch
+import librosa
+import soundfile as sf
+from pydub import AudioSegment
 from telegram import Update
 from telegram.ext import Application, MessageHandler, filters
 from transformers import pipeline, AutoTokenizer, VitsModel
-import librosa
-import soundfile as sf  # <-- المكتبة البديلة
-from pydub import AudioSegment
-import numpy as np
-import torch
+from huggingface_hub import login  # المكتبة الجديدة المطلوبة
+
+# ================ الخطوة 1: تسجيل الدخول باستخدام توكن Hugging Face ================
+login(token=os.getenv("HF_TOKEN"))  # تأكد من إضافة متغير HF_TOKEN في الإعدادات
 
-# تهيئة النظام
+# ================ تهيئة نظام التسجيل ================
 logging.basicConfig(
     format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
     level=logging.INFO
 )
 logger = logging.getLogger(__name__)
 
-# تهيئة النماذج
-asr_pipeline = pipeline(
-    "automatic-speech-recognition",
-    model="facebook/wav2vec2-large-xlsr-53-arabic"
-)
+# ================ تهيئة النماذج مع التوكن ================
+try:
+    # 1. نموذج التعرف على الكلام (ASR)
+    asr_pipeline = pipeline(
+        task="automatic-speech-recognition",
+        model="jonatasgrosman/wav2vec2-large-xlsr-53-arabic",  # نموذج بديل متوفر
+        token=os.getenv("HF_TOKEN")  # استخدام التوكن هنا
+    )
+
+    # 2. نموذج توليف الكلام (TTS)
+    tts_tokenizer = AutoTokenizer.from_pretrained(
+        "facebook/mms-tts-ara",
+        token=os.getenv("HF_TOKEN")  # استخدام التوكن هنا
+    )
+    tts_model = VitsModel.from_pretrained(
+        "facebook/mms-tts-ara",
+        token=os.getenv("HF_TOKEN")  # استخدام التوكن هنا
+    )
 
-tts_tokenizer = AutoTokenizer.from_pretrained("facebook/mms-tts-ara")
-tts_model = VitsModel.from_pretrained("facebook/mms-tts-ara")
+except Exception as e:
+    logger.error(f"خطأ في تحميل النماذج: {str(e)}")
+    raise
 
-def enhance_audio(input_path, output_path):
-    """تحسين جودة الصوت باستخدام تأثيرات متقدمة"""
+# ================ دالة تحسين جودة الصوت ================
+def enhance_audio(input_path: str, output_path: str) -> bool:
+    """تطبيق تأثيرات تحسين على الملف الصوتي"""
     try:
         audio = AudioSegment.from_wav(input_path)
-        audio = audio.low_pass_filter(3000)
-        audio = audio.high_pass_filter(100)
-        audio = audio.normalize()
-        audio = audio.fade_in(150).fade_out(150)
+        audio = audio.low_pass_filter(3000)  # تقليل الضوضاء
+        audio = audio.high_pass_filter(100)   # إزالة الهوم
+        audio = audio.normalize()             # توحيد مستوى الصوت
+        audio = audio.fade_in(150).fade_out(150)  # إضافة fade
         audio.export(output_path, format="wav")
         return True
     except Exception as e:
         logger.error(f"فشل تحسين الصوت: {str(e)}")
         return False
 
-async def speech_to_text(audio_path):
+# ================ الدوال الرئيسية ================
+async def speech_to_text(audio_path: str) -> str:
+    """تحويل الصوت إلى نص"""
     try:
         audio, sr = librosa.load(audio_path, sr=16000)
-        sf.write("temp.wav", audio, sr)  # <-- استبدال torchaudio بـ soundfile
+        sf.write("temp.wav", audio, sr)  # تحويل إلى wav
         result = asr_pipeline("temp.wav")
         return result["text"]
     except Exception as e:
         logger.error(f"فشل التعرف على الصوت: {str(e)}")
         return ""
 
-async def generate_response(text):
+async def generate_response(text: str) -> str:
+    """توليد رد الذكاء الاصطناعي"""
     try:
         chatbot = pipeline(
             "text-generation",
-            model="aubmindlab/aragpt2-base"
+            model="aubmindlab/aragpt2-base",
+            token=os.getenv("HF_TOKEN")  # استخدام التوكن هنا
         )
         response = chatbot(
             text,
@@ -64,26 +87,22 @@ async def generate_response(text):
         return response[0]['generated_text']
     except Exception as e:
         logger.error(f"فشل توليد الرد: {str(e)}")
-        return "عذرًا، لم أفهم ما تقصد."
+        return "حدث خطأ في توليد الرد."
 
-async def text_to_speech(text):
+async def text_to_speech(text: str) -> None:
+    """تحويل النص إلى صوت"""
     try:
         inputs = tts_tokenizer(text, return_tensors="pt")
         with torch.no_grad():
             output = tts_model(**inputs)
         waveform = output.waveform[0].numpy()
-        
-        # ===== التغيير الرئيسي هنا =====
-        sf.write("bot_response.wav", waveform, tts_model.config.sampling_rate)  # <-- استبدال torchaudio
-        
+        sf.write("bot_response.wav", waveform, tts_model.config.sampling_rate)  # حفظ الملف
     except Exception as e:
         logger.error(f"فشل تحويل النص إلى صوت: {str(e)}")
 
-async def process_voice(update: Update, context):
+async def process_voice(update: Update, context) -> None:
+    """معالجة الرسائل الصوتية الواردة"""
     try:
-        user = update.message.from_user
-        logger.info(f"رسالة صوتية من {user.first_name}")
-
         # تحميل الملف الصوتي
         voice_file = await update.message.voice.get_file()
         await voice_file.download_to_drive("user_voice.ogg")
@@ -93,7 +112,7 @@ async def process_voice(update: Update, context):
         bot_response = await generate_response(user_text)
         await text_to_speech(bot_response)
         
-        # تحسين الجودة وإرسال الرد
+        # إرسال الرد
         if enhance_audio("bot_response.wav", "bot_response_enhanced.wav"):
             await update.message.reply_voice("bot_response_enhanced.wav")
         else:
@@ -101,10 +120,11 @@ async def process_voice(update: Update, context):
             
     except Exception as e:
         logger.error(f"خطأ رئيسي: {str(e)}")
-        await update.message.reply_text("⚠️ حدث خطأ غير متوقع، الرجاء المحاولة لاحقًا.")
+        await update.message.reply_text("⚠️ حدث خطأ غير متوقع.")
 
+# ================ تشغيل البوت ================
 if __name__ == "__main__":
-    TOKEN = os.getenv("TELEGRAM_TOKEN")
+    TOKEN = os.getenv("TELEGRAM_TOKEN")  # تأكد من إضافة TELEGRAM_TOKEN في الإعدادات
     application = Application.builder().token(TOKEN).build()
     application.add_handler(MessageHandler(filters.VOICE, process_voice))
     application.run_polling()