2506minecraft commited on
Commit
8859b60
·
verified ·
1 Parent(s): c4de05c

Update app.py

Browse files
Files changed (1) hide show
  1. app.py +33 -38
app.py CHANGED
@@ -1,69 +1,67 @@
1
  import os
2
  import logging
 
3
  import numpy as np
4
  import torch
5
  import librosa
6
  import soundfile as sf
7
  from pydub import AudioSegment
8
  from telegram import Update
9
- from telegram.ext import Application, MessageHandler, filters
10
  from transformers import pipeline, AutoTokenizer, VitsModel
11
- from huggingface_hub import login # المكتبة الجديدة المطلوبة
12
 
13
- # ================ الخطوة 1: تسجيل الدخول باستخدام توكن Hugging Face ================
14
- login(token=os.getenv("HF_TOKEN")) # تأكد من إضافة متغير HF_TOKEN في الإعدادات
15
 
16
- # ================ تهيئة نظام التسجيل ================
17
  logging.basicConfig(
18
  format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
19
  level=logging.INFO
20
  )
21
  logger = logging.getLogger(__name__)
22
 
23
- # ================ تهيئة النماذج مع التوكن ================
24
  try:
25
- # 1. نموذج التعرف على الكلام (ASR)
26
  asr_pipeline = pipeline(
27
- task="automatic-speech-recognition",
28
- model="jonatasgrosman/wav2vec2-large-xlsr-53-arabic", # نموذج بديل متوفر
29
- token=os.getenv("HF_TOKEN") # استخدام التوكن هنا
30
  )
31
 
32
- # 2. نموذج توليف الكلام (TTS)
33
  tts_tokenizer = AutoTokenizer.from_pretrained(
34
  "facebook/mms-tts-ara",
35
- token=os.getenv("HF_TOKEN") # استخدام التوكن هنا
36
  )
37
  tts_model = VitsModel.from_pretrained(
38
  "facebook/mms-tts-ara",
39
- token=os.getenv("HF_TOKEN") # استخدام التوكن هنا
40
  )
41
 
42
  except Exception as e:
43
- logger.error(f"خطأ في تحميل النماذج: {str(e)}")
44
  raise
45
 
46
- # ================ دالة تحسين جودة الصوت ================
47
  def enhance_audio(input_path: str, output_path: str) -> bool:
48
- """تطبيق تأثيرات تحسين على الملف الصوتي"""
49
  try:
50
  audio = AudioSegment.from_wav(input_path)
51
- audio = audio.low_pass_filter(3000) # تقليل الضوضاء
52
- audio = audio.high_pass_filter(100) # إزالة الهوم
53
- audio = audio.normalize() # توحيد مستوى الصوت
54
- audio = audio.fade_in(150).fade_out(150) # إضافة fade
55
  audio.export(output_path, format="wav")
56
  return True
57
  except Exception as e:
58
  logger.error(f"فشل تحسين الصوت: {str(e)}")
59
  return False
60
 
61
- # ================ الدوال الرئيسية ================
62
  async def speech_to_text(audio_path: str) -> str:
63
- """تحويل الصوت إلى نص"""
64
  try:
65
  audio, sr = librosa.load(audio_path, sr=16000)
66
- sf.write("temp.wav", audio, sr) # تحويل إلى wav
67
  result = asr_pipeline("temp.wav")
68
  return result["text"]
69
  except Exception as e:
@@ -71,12 +69,11 @@ async def speech_to_text(audio_path: str) -> str:
71
  return ""
72
 
73
  async def generate_response(text: str) -> str:
74
- """توليد رد الذكاء الاصطناعي"""
75
  try:
76
  chatbot = pipeline(
77
  "text-generation",
78
  model="aubmindlab/aragpt2-base",
79
- token=os.getenv("HF_TOKEN") # استخدام التوكن هنا
80
  )
81
  response = chatbot(
82
  text,
@@ -87,32 +84,28 @@ async def generate_response(text: str) -> str:
87
  return response[0]['generated_text']
88
  except Exception as e:
89
  logger.error(f"فشل توليد الرد: {str(e)}")
90
- return "حدث خطأ في توليد الرد."
91
 
92
  async def text_to_speech(text: str) -> None:
93
- """تحويل النص إلى صوت"""
94
  try:
95
  inputs = tts_tokenizer(text, return_tensors="pt")
96
  with torch.no_grad():
97
  output = tts_model(**inputs)
98
  waveform = output.waveform[0].numpy()
99
- sf.write("bot_response.wav", waveform, tts_model.config.sampling_rate) # حفظ الملف
100
  except Exception as e:
101
  logger.error(f"فشل تحويل النص إلى صوت: {str(e)}")
102
 
103
- async def process_voice(update: Update, context) -> None:
104
- """معالجة الرسائل الصوتية الواردة"""
105
  try:
106
- # تحميل الملف الصوتي
107
  voice_file = await update.message.voice.get_file()
108
  await voice_file.download_to_drive("user_voice.ogg")
109
 
110
- # معالجة الصوت
111
  user_text = await speech_to_text("user_voice.ogg")
112
  bot_response = await generate_response(user_text)
113
  await text_to_speech(bot_response)
114
 
115
- # إرسال الرد
116
  if enhance_audio("bot_response.wav", "bot_response_enhanced.wav"):
117
  await update.message.reply_voice("bot_response_enhanced.wav")
118
  else:
@@ -122,9 +115,11 @@ async def process_voice(update: Update, context) -> None:
122
  logger.error(f"خطأ رئيسي: {str(e)}")
123
  await update.message.reply_text("⚠️ حدث خطأ غير متوقع.")
124
 
125
- # ================ تشغيل البوت ================
126
- if __name__ == "__main__":
127
- TOKEN = os.getenv("TELEGRAM_TOKEN") # تأكد من إضافة TELEGRAM_TOKEN في الإعدادات
128
- application = Application.builder().token(TOKEN).build()
129
  application.add_handler(MessageHandler(filters.VOICE, process_voice))
130
- application.run_polling()
 
 
 
 
1
  import os
2
  import logging
3
+ import asyncio
4
  import numpy as np
5
  import torch
6
  import librosa
7
  import soundfile as sf
8
  from pydub import AudioSegment
9
  from telegram import Update
10
+ from telegram.ext import ApplicationBuilder, MessageHandler, filters
11
  from transformers import pipeline, AutoTokenizer, VitsModel
12
+ from huggingface_hub import login
13
 
14
+ # ===== تهيئة التوكن وتسجيل الدخول =====
15
+ login(token=os.getenv("HF_TOKEN"))
16
 
17
+ # ===== تهيئة نظام التسجيل =====
18
  logging.basicConfig(
19
  format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
20
  level=logging.INFO
21
  )
22
  logger = logging.getLogger(__name__)
23
 
24
+ # ===== تهيئة النماذج مع التعامل مع الأخطاء =====
25
  try:
26
+ # 1. نموذج التعرف على الكلام
27
  asr_pipeline = pipeline(
28
+ "automatic-speech-recognition",
29
+ model="jonatasgrosman/wav2vec2-large-xlsr-53-arabic",
30
+ token=os.getenv("HF_TOKEN")
31
  )
32
 
33
+ # 2. نموذج توليف الصوت
34
  tts_tokenizer = AutoTokenizer.from_pretrained(
35
  "facebook/mms-tts-ara",
36
+ token=os.getenv("HF_TOKEN")
37
  )
38
  tts_model = VitsModel.from_pretrained(
39
  "facebook/mms-tts-ara",
40
+ token=os.getenv("HF_TOKEN")
41
  )
42
 
43
  except Exception as e:
44
+ logger.error(f"فشل تحميل النماذج: {str(e)}")
45
  raise
46
 
47
+ # ===== دوال معالجة الصوت =====
48
  def enhance_audio(input_path: str, output_path: str) -> bool:
 
49
  try:
50
  audio = AudioSegment.from_wav(input_path)
51
+ audio = audio.low_pass_filter(3000)
52
+ audio = audio.high_pass_filter(100)
53
+ audio = audio.normalize()
54
+ audio = audio.fade_in(150).fade_out(150)
55
  audio.export(output_path, format="wav")
56
  return True
57
  except Exception as e:
58
  logger.error(f"فشل تحسين الصوت: {str(e)}")
59
  return False
60
 
 
61
  async def speech_to_text(audio_path: str) -> str:
 
62
  try:
63
  audio, sr = librosa.load(audio_path, sr=16000)
64
+ sf.write("temp.wav", audio, sr)
65
  result = asr_pipeline("temp.wav")
66
  return result["text"]
67
  except Exception as e:
 
69
  return ""
70
 
71
  async def generate_response(text: str) -> str:
 
72
  try:
73
  chatbot = pipeline(
74
  "text-generation",
75
  model="aubmindlab/aragpt2-base",
76
+ token=os.getenv("HF_TOKEN")
77
  )
78
  response = chatbot(
79
  text,
 
84
  return response[0]['generated_text']
85
  except Exception as e:
86
  logger.error(f"فشل توليد الرد: {str(e)}")
87
+ return "عذرًا، حدث خطأ في الرد."
88
 
89
  async def text_to_speech(text: str) -> None:
 
90
  try:
91
  inputs = tts_tokenizer(text, return_tensors="pt")
92
  with torch.no_grad():
93
  output = tts_model(**inputs)
94
  waveform = output.waveform[0].numpy()
95
+ sf.write("bot_response.wav", waveform, tts_model.config.sampling_rate)
96
  except Exception as e:
97
  logger.error(f"فشل تحويل النص إلى صوت: {str(e)}")
98
 
99
+ # ===== الدالة الرئيسية للمحادثة =====
100
+ async def process_voice(update: Update, context):
101
  try:
 
102
  voice_file = await update.message.voice.get_file()
103
  await voice_file.download_to_drive("user_voice.ogg")
104
 
 
105
  user_text = await speech_to_text("user_voice.ogg")
106
  bot_response = await generate_response(user_text)
107
  await text_to_speech(bot_response)
108
 
 
109
  if enhance_audio("bot_response.wav", "bot_response_enhanced.wav"):
110
  await update.message.reply_voice("bot_response_enhanced.wav")
111
  else:
 
115
  logger.error(f"خطأ رئيسي: {str(e)}")
116
  await update.message.reply_text("⚠️ حدث خطأ غير متوقع.")
117
 
118
+ # ===== التشغيل الرئيسي مع إصلاح حدث اللوب =====
119
+ async def main():
120
+ application = ApplicationBuilder().token(os.getenv("TELEGRAM_TOKEN")).build()
 
121
  application.add_handler(MessageHandler(filters.VOICE, process_voice))
122
+ await application.run_polling()
123
+
124
+ if __name__ == "__main__":
125
+ asyncio.run(main())