import os import time import tempfile import subprocess import json import logging import telebot import mimetypes import re import requests import urllib.parse from acrcloud.recognizer import ACRCloudRecognizer from telebot.types import InlineKeyboardMarkup, InlineKeyboardButton from functools import wraps TELEGRAM_BOT_TOKEN = "8486418359:AAEUz31G6JQoobRcwaag2pWSPD6KJWBhgTk" CHANNEL_ID = "-1003006093715" CHANNEL_USERNAME = "t.me/SourceKodo" DEVELOPER_URL = "https://t.me/Evan1_a" DEVELOPER_ID = 7923164784 USERS_FILE = 'users.json' ACR_CONFIG = { 'host': 'identify-ap-southeast-1.acrcloud.com', 'access_key': '4a86ccf529d1ad4097733441f7cfb248', 'access_secret': '8gW9OSST8vazuICNotp4SHwLfjx1NsTPWwSkg4bt', 'timeout': 10 } bot = telebot.TeleBot(TELEGRAM_BOT_TOKEN) recognizer = ACRCloudRecognizer(ACR_CONFIG) logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def load_users(): if not os.path.exists(USERS_FILE): return {} try: with open(USERS_FILE, 'r', encoding='utf-8') as f: return json.load(f) except (IOError, json.JSONDecodeError): return {} def save_users(users_data): try: with open(USERS_FILE, 'w', encoding='utf-8') as f: json.dump(users_data, f, indent=4, ensure_ascii=False) except IOError as e: logger.error(f"Error saving users data: {e}") def save_user_data(user): users = load_users() user_id_str = str(user.id) if user_id_str not in users: user_info = { 'name': user.first_name + (f" {user.last_name}" if user.last_name else ""), 'username': user.username, 'user_id': user.id } users[user_id_str] = user_info save_users(users) logger.info(f"New user saved: {user.id}") def check_subscription(user_id): try: member = bot.get_chat_member(CHANNEL_ID, user_id) return member.status in ['member', 'administrator', 'creator'] except Exception as e: logger.error(f"Error checking subscription...") return False def developer_button(): keyboard = InlineKeyboardMarkup() keyboard.add(InlineKeyboardButton("Developer", url=DEVELOPER_URL)) return keyboard def subscription_required(func): @wraps(func) def wrapped(message, *args, **kwargs): if not check_subscription(message.from_user.id): bot.send_message(message.chat.id, f"يجب عليك الاشتراك في القناة أولاً لاستخدام البوت:\n\n- {CHANNEL_USERNAME}", parse_mode='HTML') return return func(message, *args, **kwargs) return wrapped def developer_only(func): @wraps(func) def wrapped(message, *args, **kwargs): if message.from_user.id != DEVELOPER_ID: bot.reply_to(message, "عذراً، هذا الأمر مخصص للمطور فقط.", parse_mode='HTML') return return func(message, *args, **kwargs) return wrapped @bot.message_handler(commands=['start']) @subscription_required def handle_start(message): save_user_data(message.from_user) bot.reply_to(message, "مرحباً بك في البوت! أرسل لي رسالة صوتية أو ملف صوتي أو قم بتوجية رسالة صوتية (audio/document/voice) للتعرف على الموسيقى التي تبحث عنها.", parse_mode='HTML') def dev_menu_markup(): markup = InlineKeyboardMarkup() markup.row( InlineKeyboardButton("الإحصائيات", callback_data='dev_stats'), InlineKeyboardButton("إذاعة بالخاص", callback_data='dev_broadcast_start') ) return markup def send_dev_menu(chat_id, message_id=None): text = "عزيزي هذه لوحة المطور الخاصة بك:" markup = dev_menu_markup() if message_id: bot.edit_message_text(text, chat_id, message_id, reply_markup=markup, parse_mode='HTML') else: bot.send_message(chat_id, text, reply_markup=markup, parse_mode='HTML') @bot.message_handler(commands=['help']) @subscription_required def handle_help(message): help_text = ( "مرحباً بك في بوت التعرف على الموسيقى!\n\n" "كيف يعمل البوت؟\n" "يقوم البوت بتحليل المقاطع الصوتية التي ترسلها للتعرف على اسم الأغنية والفنان.\n\n" "للحصول على أفضل النتائج:\n" "1. أرسل مقطعاً صوتياً واضحاً.\n" "2. يفضل أن يكون المقطع الصوتي بين 10 إلى 15 ثانية.\n" "3. يمكنك إرسال ملف صوتي (Audio) أو رسالة صوتية (Voice Message) أو قم بتوجية رسالة صوتية.\n\n" "الأوامر المتاحة:\n" "/start - لبدء استخدام البوت.\n" "/help - لعرض هذه الرسالة." ) markup = InlineKeyboardMarkup() markup.add(InlineKeyboardButton("Source Channel", url=CHANNEL_USERNAME)) bot.send_message(message.chat.id, help_text, reply_markup=markup, parse_mode='HTML') @bot.message_handler(commands=['dev']) @developer_only def handle_dev(message): send_dev_menu(message.chat.id) @bot.callback_query_handler(func=lambda call: call.data.startswith('dev_')) @developer_only def handle_dev_callback(call): bot.answer_callback_query(call.id) chat_id = call.message.chat.id message_id = call.message.message_id data = call.data if data == 'dev_menu': bot.clear_step_handler_by_chat_id(chat_id) send_dev_menu(chat_id, message_id) return back_markup = InlineKeyboardMarkup() back_markup.add(InlineKeyboardButton("الرجوع للقائمة الرئيسية", callback_data='dev_menu')) if data == 'dev_stats': users = load_users() count = len(users) text = f"الاحصائيات:\n\nعدد المستخدمين: {count} مستخدم." bot.edit_message_text(text, chat_id, message_id, reply_markup=back_markup, parse_mode='HTML') elif data == 'dev_broadcast_start': text = "أرسل الآن الرسالة التي تريد بثها لجميع المستخدمين." bot.edit_message_text(text, chat_id, message_id, reply_markup=back_markup, parse_mode='HTML') bot.register_next_step_handler(call.message, process_broadcast_message, message_id_to_edit=message_id) def process_broadcast_message(message, message_id_to_edit): chat_id = message.chat.id if message.from_user.id != DEVELOPER_ID: bot.send_message(chat_id, "تم إلغاء العملية. هذا الأمر مخصص للمطور فقط.", parse_mode='HTML') return broadcast_text = message.text if not broadcast_text: bot.edit_message_text("لم يتم إرسال نص. تم إلغاء الإذاعة.", chat_id, message_id_to_edit, reply_markup=dev_menu_markup(), parse_mode='HTML') return users = load_users() total_users = len(users) sent_count = 0 failed_count = 0 bot.edit_message_text(f"بدء عملية الإذاعة لـ {total_users} مستخدم...", chat_id, message_id_to_edit, parse_mode='HTML') for user_id_str, user_info in users.items(): try: bot.send_message(user_info['user_id'], broadcast_text, parse_mode='HTML') sent_count += 1 time.sleep(1) except Exception as e: logger.warning(f"Failed to send broadcast to user {user_id_str}: {e}") failed_count += 1 result_text = f"انتهت عملية الإذاعة:\n\nتم الإرسال بنجاح إلى: {sent_count} مستخدم.\nفشل الإرسال إلى: {failed_count} مستخدم." back_markup = InlineKeyboardMarkup() back_markup.add(InlineKeyboardButton("الرجوع للقائمة الرئيسية", callback_data='dev_menu')) bot.edit_message_text(result_text, chat_id, message_id_to_edit, reply_markup=back_markup, parse_mode='HTML') @bot.message_handler(func=lambda message: True) @subscription_required def handle_other_messages(message): bot.send_message(message.chat.id, "أرسل لي رسالة صوتية أو ملف صوتي أو قم بتوجية رسالة صوتية (audio/document/voice) للتعرف على الموسيقى.", parse_mode='HTML') def ffmpeg_convert(input_path, output_path, duration_s=12, srate=8000, channels=1): cmd = [ 'ffmpeg', '-y', '-hide_banner', '-loglevel', 'error', '-i', input_path, '-t', str(duration_s), '-ar', str(srate), '-ac', str(channels), '-sample_fmt', 's16', output_path ] subprocess.run(cmd, check=True) def acr_recognize_file(wav_path, start_seconds=0, rec_length=12): try: res = recognizer.recognize_by_file(wav_path, start_seconds, rec_length) if isinstance(res, str): return json.loads(res) return res except Exception: logger.exception("ACRCloud recognition error") return None def extract_track_info(acr_result): try: metadata = acr_result.get('metadata', {}) musics = metadata.get('music', []) if not musics: return None first = musics[0] title = first.get('title') artists = first.get('artists', []) artist_names = ", ".join([a.get('name') for a in artists if a.get('name')]) or None return { 'title': title, 'artist': artist_names } except Exception: logger.exception("Error parsing ACR result") return None def youtube_search_link(query): try: encoded_query = urllib.parse.quote_plus(query) url = f"https://www.youtube.com/results?search_query={encoded_query}" headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36', 'Accept-Language': 'en-US,en;q=0.9' } response = requests.get(url, headers=headers, timeout=10) video_ids = re.findall(r"watch\?v=(\S{11})", response.text) if video_ids: return f"https://www.youtube.com/watch?v={video_ids[0]}" return None except Exception: logger.exception("YouTube search failed") return None @bot.message_handler(content_types=['voice', 'audio', 'document', 'video_note']) @subscription_required def handle_media(message): og_path = None acr_wav = None try: bot.send_chat_action(message.chat.id, 'typing') file_id = None ext = None if message.voice: file_id = message.voice.file_id ext = '.ogg' elif message.video_note: file_id = message.video_note.file_id ext = '.ogg' elif message.audio: file_id = message.audio.file_id ext = os.path.splitext(getattr(message.audio, 'file_name', '') or '')[1] or mimetypes.guess_extension(message.audio.mime_type or '') or '.mp3' elif message.document: doc = message.document fname = doc.file_name or '' mime = (doc.mime_type or '').lower() if mime.startswith('audio') or fname.lower().endswith(('.mp3', '.wav', '.ogg', '.m4a', '.flac', '.aac', '.oga')): file_id = doc.file_id ext = os.path.splitext(fname)[1] or mimetypes.guess_extension(mime) or '.mp3' else: bot.reply_to(message, "الملف المرسل ليس ملف صوتي. أرسل ملف صوتي أو رسالة صوتية.", parse_mode='HTML') return if not file_id: bot.reply_to(message, "لم أجد ملف صوتي في الرسالة.", parse_mode='HTML') return file_info = bot.get_file(file_id) file_bytes = bot.download_file(file_info.file_path) if not ext: ext = os.path.splitext(file_info.file_path)[1] or '.mp3' with tempfile.NamedTemporaryFile(suffix=ext, delete=False) as f: f.write(file_bytes) og_path = f.name acr_wav = og_path + '.acr.wav' ffmpeg_convert(og_path, acr_wav, duration_s=12, srate=8000, channels=1) acr_raw = acr_recognize_file(acr_wav, start_seconds=0, rec_length=12) logger.info("ACR response: %s", acr_raw) if not acr_raw: bot.reply_to(message, "حدث خطأ أثناء محاولة التعرف على الصوت!\nتواصل مع المطور لحل المشكلة:", reply_markup=developer_button(), parse_mode='HTML') return status = acr_raw.get('status', {}) code = status.get('code') if code == 0: track = extract_track_info(acr_raw) if not track: bot.reply_to(message, "تم الاتصال بخدمة التعرف لكن لم تُرجع تفاصيل للموسيقى.\nتواصل مع المطور لحل المشكلة:", reply_markup=developer_button(), parse_mode='HTML') return title = track.get('title') or "Unknown" artist = track.get('artist') or "Unknown" search_query = f"{title} {artist}".strip() youtube_link = youtube_search_link(search_query) reply_text = ( f"The music has been identified:\n\n" f"Music name: {title}\n" f"Artist name: {artist}" ) markup = InlineKeyboardMarkup() if youtube_link: markup.add(InlineKeyboardButton("Watch on YouTube", url=youtube_link)) bot.reply_to(message, reply_text, reply_markup=markup, parse_mode='HTML') return elif code == 1001: bot.reply_to(message, "اسف جداً لم يتم التعرف على أي موسيقى!\nجرّب مقطعاً أطول أو أوضح (10-15 ثانية).\n\nتواصل مع المطور إن واجهتك مشكلة:", reply_markup=developer_button(), parse_mode='HTML') return else: bot.reply_to(message, f"خطأ في خدمة التعرف على الموسيقى!\nتواصل مع المطور لحل المشكلة:", reply_markup=developer_button(), parse_mode='HTML') return except subprocess.CalledProcessError: logger.exception("ffmpeg failed") bot.reply_to(message, "خطأ في معالجة الصوت!\nتواصل مع المطور لحل المشكلة:", reply_markup=developer_button(), parse_mode='HTML') except Exception as e: logger.exception("General error") bot.reply_to(message, f"حدث خطأ أثناء المعالجة!\nتواصل مع المطور لحل المشكلة:", reply_markup=developer_button(), parse_mode='HTML') finally: for p in [og_path, acr_wav]: try: if p and os.path.exists(p): os.remove(p) except Exception: pass if __name__ == "__main__": print("Starting bot with polling mode...") try: bot.delete_webhook() print("Existing webhook deleted successfully.") except Exception as e: print(f"Could not delete webhook: {e}") bot.polling(none_stop=True, interval=0, timeout=20) print("Bot polling stopped.")