app.py

import os
import logging
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

# تهيئة النظام
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"
)

tts_tokenizer = AutoTokenizer.from_pretrained("facebook/mms-tts-ara")
tts_model = VitsModel.from_pretrained("facebook/mms-tts-ara")

def enhance_audio(input_path, output_path):
    """تحسين جودة الصوت باستخدام تأثيرات متقدمة"""
    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.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):
    try:
        audio, sr = librosa.load(audio_path, sr=16000)
        sf.write("temp.wav", audio, sr)  # <-- استبدال torchaudio بـ soundfile
        result = asr_pipeline("temp.wav")
        return result["text"]
    except Exception as e:
        logger.error(f"فشل التعرف على الصوت: {str(e)}")
        return ""

async def generate_response(text):
    try:
        chatbot = pipeline(
            "text-generation",
            model="aubmindlab/aragpt2-base"
        )
        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):
    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
        
    except Exception as e:
        logger.error(f"فشل تحويل النص إلى صوت: {str(e)}")

async def process_voice(update: Update, context):
    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")
        
        # معالجة الصوت
        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")
    application = Application.builder().token(TOKEN).build()
    application.add_handler(MessageHandler(filters.VOICE, process_voice))
    application.run_polling()