app.py

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
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__)

# ================ تهيئة النماذج مع التوكن ================
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")  # استخدام التوكن هنا
    )

except Exception as e:
    logger.error(f"خطأ في تحميل النماذج: {str(e)}")
    raise

# ================ دالة تحسين جودة الصوت ================
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)  # إضافة 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: str) -> str:
    """تحويل الصوت إلى نص"""
    try:
        audio, sr = librosa.load(audio_path, sr=16000)
        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: str) -> str:
    """توليد رد الذكاء الاصطناعي"""
    try:
        chatbot = pipeline(
            "text-generation",
            model="aubmindlab/aragpt2-base",
            token=os.getenv("HF_TOKEN")  # استخدام التوكن هنا
        )
        response = chatbot(
            text,
            max_length=100,
            num_return_sequences=1,
            pad_token_id=50256
        )
        return response[0]['generated_text']
    except Exception as e:
        logger.error(f"فشل توليد الرد: {str(e)}")
        return "حدث خطأ في توليد الرد."

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)  # حفظ الملف
    except Exception as e:
        logger.error(f"فشل تحويل النص إلى صوت: {str(e)}")

async def process_voice(update: Update, context) -> None:
    """معالجة الرسائل الصوتية الواردة"""
    try:
        # تحميل الملف الصوتي
        voice_file = await update.message.voice.get_file()
        await voice_file.download_to_drive("user_voice.ogg")
        
        # معالجة الصوت
        user_text = await speech_to_text("user_voice.ogg")
        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:
            await update.message.reply_voice("bot_response.wav")
            
    except Exception as e:
        logger.error(f"خطأ رئيسي: {str(e)}")
        await update.message.reply_text("⚠️ حدث خطأ غير متوقع.")

# ================ تشغيل البوت ================
if __name__ == "__main__":
    TOKEN = os.getenv("TELEGRAM_TOKEN")  # تأكد من إضافة TELEGRAM_TOKEN في الإعدادات
    application = Application.builder().token(TOKEN).build()
    application.add_handler(MessageHandler(filters.VOICE, process_voice))
    application.run_polling()