import os import sys import re import subprocess import json from io import BytesIO from telegram import Update, InputFile, InlineKeyboardButton, InlineKeyboardMarkup, Bot from telegram.ext import Application, CommandHandler, MessageHandler, CallbackQueryHandler, filters, ConversationHandler from telegram.error import TelegramError from pydub import AudioSegment from fastapi import FastAPI, Request, HTTPException from fastapi.responses import HTMLResponse import uvicorn import tempfile # برای مدیریت فایل های موقت برای ffmpeg # --- تنظیمات --- # ⚠️⚠️⚠️ بسیار مهم: این مقادیر را به عنوان متغیرهای محیطی (Environment Variables) در Hugging Face Space خود تنظیم کنید.⚠️⚠️⚠️ # برای مثال، در تنظیمات اسپیس خود، VARs را اضافه کنید: # BOT_TOKEN = "YOUR_TELEGRAM_BOT_TOKEN" # ADMIN_TELEGRAM_ID = "YOUR_ADMIN_NUMERIC_ID" (مثلا 123456789) TOKEN = os.getenv("BOT_TOKEN", "YOUR_FALLBACK_TOKEN_HERE") # اگر متغیر محیطی ست نشد، از این استفاده می کند (فقط برای توسعه) ADMIN_ID = int(os.getenv("ADMIN_TELEGRAM_ID", "123456789")) # ID ادمین را اینجا وارد کنید BOT_USERNAME = "Voice2mp3_RoBot" # نام کاربری ربات برای استفاده در نام فایل ها # CHANNELS_FILE برای ذخیره لیست کانال های اجباری در Persistence Storage اسپیس # در Hugging Face Spaces، فایل های غیر از /tmp پایدار می مانند. CHANNELS_FILE = "channels.json" # --- توابع کمکی برای مدیریت کانالها --- def load_required_channels(): """کانالهای مورد نیاز را از فایل JSON بارگذاری میکند.""" if os.path.exists(CHANNELS_FILE): try: with open(CHANNELS_FILE, 'r', encoding='utf-8') as f: return json.load(f) except json.JSONDecodeError: print(f"Error decoding JSON from {CHANNELS_FILE}. Starting with empty list.", file=sys.stderr) return [] return [] def save_required_channels(channels): """کانالهای مورد نیاز را در فایل JSON ذخیره میکند.""" with open(CHANNELS_FILE, 'w', encoding='utf-8') as f: json.dump(channels, f, indent=4, ensure_ascii=False) REQUIRED_CHANNELS = load_required_channels() # --- تعریف حالتهای مکالمه --- LANGUAGE_SELECTION, MAIN_MENU, CONVERT_AUDIO, CUT_AUDIO_FILE, CUT_AUDIO_RANGE, \ VIDEO_CONVERSION_MODE, WAITING_FOR_MEMBERSHIP, \ ADMIN_MENU, ADD_CHANNEL, LIST_REMOVE_CHANNELS = range(10) # فرهنگ لغت برای پیامها به زبانهای مختلف MESSAGES = { 'fa': { 'start_welcome': "سلام! من یک ربات تبدیل فرمت صوتی و ویدیویی هستم.\n\nبرای شروع، از منوی زیر یک قابلیت را انتخاب کنید.", 'choose_language': "زبان مورد نظر خود را انتخاب کنید:", 'processing_start': "⏳ در حال شروع پردازش...", 'file_received': "⬇️ فایل دریافت شد. در حال تبدیل...", 'conversion_done': "⚙️ تبدیل فرمت انجام شد. در حال ارسال...", 'mp3_to_voice_reply': "ویس تلگرام شما (تبدیل شده از MP3)", 'voice_to_mp3_caption': "فایل MP3 شما (تبدیل شده از ویس تلگرام)", 'error_mp3_to_voice': "❌ خطا در تبدیل MP3 به ویس تلگرام: ", 'error_voice_to_mp3': "❌ خطا در تبدیل ویس تلگرام به MP3: ", 'general_error': "متاسفم، مشکلی پیش آمد. لطفاً دوباره تلاش کنید.", 'main_menu_prompt': "چه کاری میخواهید روی فایل خود انجام دهید؟", 'btn_convert_format': "تغییر فرمت صدا 🎵", 'btn_cut_audio': "برش قسمتی از صدا ✂️", 'btn_video_conversion': "تبدیل ویدیو دایرهای 🎥", 'convert_mode_active': "شما در حالت 'تغییر فرمت صدا' هستید. حالا فایل صوتی (ویس یا MP3) خود را برای من ارسال کنید.", 'cut_mode_active_file': "شما در قسمت 'برش و کات کردن صدا' هستید.\n\nابتدا فایل صوتی (MP3 یا ویس) خود را ارسال کنید.", 'cut_mode_active_range': "حالا بازه زمانی مورد نظر برای برش را به صورت 'دقیقه.ثانیه-دقیقه.ثانیه' (مثال: 00.21-00.54) ارسال کنید.", 'invalid_time_format': "فرمت زمان وارد شده صحیح نیست. لطفاً از فرمت 'MM.SS-MM.SS' استفاده کنید. (مثال: 00.21-00.54)", 'invalid_time_range': "بازه زمانی نامعتبر است یا زمان پایان از زمان شروع کمتر است. لطفاً بازه صحیح را وارد کنید.", 'audio_cut_success': "✅ برش صدا با موفقیت انجام شد. فایل شما آماده است.", 'no_audio_for_cut': "فایلی برای برش پیدا نشد. لطفاً ابتدا فایل صوتی را ارسال کنید.", 'cut_processing': "✂️ در حال برش صدا...", 'returning_to_main_menu': "بازگشت به منوی اصلی...", 'cancel_message': "عملیات لغو شد. به منوی اصلی بازگشتید.", 'video_conversion_mode_active': "شما در حالت 'تبدیل ویدیو دایرهای' هستید.\n\nیک ویدیو معمولی یا یک ویدیو دایرهای (Video Message) برای من ارسال کنید.", 'file_received_video': "⬇️ فایل ویدیویی دریافت شد. در حال پردازش...", 'converting_video_note_to_video': "🔄 در حال تبدیل ویدیو دایرهای به ویدیو معمولی...", 'converting_video_to_video_note': "🔄 در حال تبدیل ویدیو معمولی به ویدیو دایرهای...", 'conversion_done_video': "✅ تبدیل ویدیو با موفقیت انجام شد. در حال ارسال...", 'video_note_to_video_caption': "ویدیو معمولی شما (تبدیل شده از ویدیو دایرهای)", 'video_to_video_note_reply': "ویدیو دایرهای شما (تبدیل شده از ویدیو معمولی)", 'error_video_conversion': "❌ خطا در تبدیل ویدیو: ", 'invalid_file_type_video': "لطفاً یک فایل ویدیویی یا ویدیو دایرهای ارسال کنید.", 'membership_required': "برای ادامه کار با ربات و استفاده نامحدود، لطفاً ابتدا عضو کانالهای زیر شوید:", 'btn_join_channel': "عضو شدن 🤝", 'btn_check_membership': "بررسی عضویت ✅", 'membership_success': "✅ عضویت شما تأیید شد! اکنون میتوانید به صورت نامحدود از ربات استفاده کنید.", 'membership_failed': "❌ متاسفم، شما هنوز عضو تمام کانالهای مورد نیاز نیستید. لطفاً ابتدا عضو شوید و سپس دوباره 'بررسی عضویت' را بزنید.", 'not_admin': "شما اجازه دسترسی به این بخش را ندارید.", 'admin_menu_prompt': "به پنل مدیریت لینکها خوش آمدید:", 'btn_add_channel': "افزودن لینک کانال ➕", 'btn_list_channels': "لیست کانالها و حذف 🗑️", 'send_channel_link': "لطفاً لینک (مانند @mychannel) یا آیدی عددی کانال را ارسال کنید:", 'channel_added': "✅ کانال '{channel_id}' با موفقیت اضافه شد.", 'channel_already_exists': "❗️ این کانال قبلاً اضافه شده است.", 'no_channels_configured': "هیچ کانالی برای عضویت پیکربندی نشده است.", 'channel_list_prompt': "لیست کانالهای فعلی برای عضویت اجباری:", 'btn_remove_channel': "حذف ❌", 'channel_removed': "✅ کانال '{channel_id}' با موفقیت حذف شد.", 'channel_not_found': "❗️ کانال مورد نظر یافت نشد.", 'invalid_channel_id': "آیدی/لینک کانال نامعتبر است. لطفاً @username یا آیدی عددی (مانند -1001234567890) را ارسال کنید.", 'bot_not_admin_in_channel': "ربات ادمین کانال '{channel_id}' نیست یا مجوزهای کافی برای بررسی عضویت را ندارد. لطفاً ربات را به عنوان ادمین با مجوز 'بررسی وضعیت اعضا' در کانال اضافه کنید." }, 'en': { 'start_welcome': "Hello! I am an audio and video format conversion bot.\n\nTo start, select a feature from the menu below.", 'choose_language': "Choose your preferred language:", 'processing_start': "⏳ Starting processing...", 'file_received': "⬇️ File received. Processing...", 'conversion_done': "⚙️ Conversion complete. Sending...", 'mp3_to_voice_reply': "Your Telegram voice (converted from MP3)", 'voice_to_mp3_caption': "Your MP3 file (converted from Telegram voice)", 'error_mp3_to_voice': "❌ Error converting MP3 to Telegram voice: ", 'error_voice_to_mp3': "❌ Error converting Telegram voice to MP3: ", 'general_error': "Sorry, something went wrong. Please try again.", 'main_menu_prompt': "What would you like to do with your file?", 'btn_convert_format': "Change Audio Format 🎵", 'btn_cut_audio': "Cut Part of Audio ✂️", 'btn_video_conversion': "Convert Circular Video 🎥", 'convert_mode_active': "You are now in 'Change Audio Format' mode. Send me your audio file (voice or MP3).", 'cut_mode_active_file': "You are in the 'Cut Audio' section.\n\nFirst, send your audio file (MP3 or voice).", 'cut_mode_active_range': "Now send the desired time range for cutting in 'MM.SS-MM.SS' format (example: 00.21-00.54).", 'invalid_time_format': "Invalid time format. Please use 'MM.SS-MM.SS' format. (example: 00.21-00.54)", 'invalid_time_range': "Invalid time range or end time is less than start time. Please enter a valid range.", 'audio_cut_success': "✅ Audio cut successfully. Your file is ready.", 'no_audio_for_cut': "No audio file found for cutting. Please send the audio file first.", 'cut_processing': "✂️ Cutting audio...", 'returning_to_main_menu': "Returning to main menu...", 'cancel_message': "Operation cancelled. Returned to main menu.", 'video_conversion_mode_active': "You are in 'Circular Video Conversion' mode.\n\nSend me a regular video or a circular video message (Video Message).", 'file_received_video': "⬇️ Video file received. Processing...", 'converting_video_note_to_video': "🔄 Converting circular video to regular video...", 'converting_video_to_video_note': "🔄 Converting regular video to circular video...", 'conversion_done_video': "✅ Video conversion successful. Sending...", 'video_note_to_video_caption': "Your regular video (converted from circular video)", 'video_to_video_note_reply': "Your circular video (converted from regular video)", 'error_video_conversion': "❌ Error converting video: ", 'invalid_file_type_video': "Please send a video file or a video message.", 'membership_required': "To continue using the bot and access unlimited features, please join the following channels first:", 'btn_join_channel': "Join Channel 🤝", 'btn_check_membership': "Check Membership ✅", 'membership_success': "✅ Your membership has been verified! You can now use the bot unlimitedly.", 'membership_failed': "❌ Sorry, you are not yet a member of all required channels. Please join first and then press 'Check Membership' again.", 'not_admin': "You do not have permission to access this section.", 'admin_menu_prompt': "Welcome to the link management panel:", 'btn_add_channel': "Add Channel Link ➕", 'btn_list_channels': "List Channels & Remove 🗑️", 'send_channel_link': "Please send the channel link (e.g., @mychannel) or numeric ID:", 'channel_added': "✅ Channel '{channel_id}' with successfully added.", 'channel_already_exists': "❗️ This channel has already been added.", 'no_channels_configured': "No channels configured for membership.", 'channel_list_prompt': "Current list of channels for mandatory membership:", 'btn_remove_channel': "Remove ❌", 'channel_removed': "✅ Channel '{channel_id}' successfully removed.", 'channel_not_found': "❗️ Channel not found.", 'invalid_channel_id': "Invalid channel ID/link. Please send @username or numeric ID (e.g., -1001234567890).", 'bot_not_admin_in_channel': "The bot is not an admin in channel '{channel_id}' or does not have sufficient permissions to check membership. Please add the bot as an admin with 'Check members' permission in the channel." } } def get_message(context, key, **kwargs): lang = context.user_data.get('language', 'fa') message_template = MESSAGES[lang].get(key, MESSAGES['fa'][key]) return message_template.format(**kwargs) def parse_time_to_ms(time_str): match = re.match(r'^(\d{2})\.(\d{2})$', time_str) if not match: raise ValueError("Invalid time format") minutes = int(match.group(1)) seconds = int(match.group(2)) if seconds >= 60: raise ValueError("Seconds must be between 00 and 59") return (minutes * 60 + seconds) * 1000 async def check_user_membership(update: Update, context): user_id = update.effective_user.id lang = context.user_data.get('language', 'fa') if not REQUIRED_CHANNELS: return True all_channels_joined = True for channel_id in REQUIRED_CHANNELS: try: chat_member = await context.bot.get_chat_member(chat_id=channel_id, user_id=user_id) if chat_member.status not in ['member', 'administrator', 'creator']: all_channels_joined = False break except Exception as e: print(f"Error checking membership for channel {channel_id}: {e}", file=sys.stderr) all_channels_joined = False break return all_channels_joined async def show_membership_required_message(update: Update, context): keyboard = [] if not REQUIRED_CHANNELS: return await show_main_menu(update, context) for channel_id in REQUIRED_CHANNELS: try: chat = await context.bot.get_chat(chat_id=channel_id) if chat.invite_link: keyboard.append([InlineKeyboardButton(f"{get_message(context, 'btn_join_channel')} {chat.title or channel_id}", url=chat.invite_link)]) elif chat.username: keyboard.append([InlineKeyboardButton(f"{get_message(context, 'btn_join_channel')} @{chat.username}", url=f"https://t.me/{chat.username}")]) else: keyboard.append([InlineKeyboardButton(f"{get_message(context, 'btn_join_channel')} {channel_id}", callback_data=f"no_link_{channel_id}")]) except Exception as e: print(f"Could not get chat info for {channel_id}: {e}", file=sys.stderr) keyboard.append([InlineKeyboardButton(f"{get_message(context, 'btn_join_channel')} {channel_id}", callback_data=f"no_link_{channel_id}")]) keyboard.append([InlineKeyboardButton(get_message(context, 'btn_check_membership'), callback_data='check_membership')]) reply_markup = InlineKeyboardMarkup(keyboard) if update.callback_query: await update.callback_query.edit_message_text( get_message(context, 'membership_required'), reply_markup=reply_markup ) else: await update.effective_message.reply_text( get_message(context, 'membership_required'), reply_markup=reply_markup ) return WAITING_FOR_MEMBERSHIP async def process_feature_or_check_membership(update: Update, context, feature_func, *args, **kwargs): user_id = update.effective_user.id if user_id == ADMIN_ID: return await feature_func(update, context, *args, **kwargs) if not REQUIRED_CHANNELS: return await feature_func(update, context, *args, **kwargs) if context.user_data.get('used_bot_once', False) and not context.user_data.get('is_member', False): return await show_membership_required_message(update, context) if not context.user_data.get('used_bot_once', False): context.user_data['used_bot_once'] = True is_member = await check_user_membership(update, context) context.user_data['is_member'] = is_member if not is_member: return await show_membership_required_message(update, context) result = await feature_func(update, context, *args, **kwargs) if not context.user_data.get('is_member', False): return await show_membership_required_message(update, context) return result async def check_membership_callback(update: Update, context): query = update.callback_query await query.answer() is_member = await check_user_membership(update, context) if is_member: context.user_data['is_member'] = True await query.edit_message_text(get_message(context, 'membership_success')) return await show_main_menu(update, context) else: await query.edit_message_text(get_message(context, 'membership_failed'), reply_markup=query.message.reply_markup) return WAITING_FOR_MEMBERSHIP async def start(update: Update, context): keyboard = [ [InlineKeyboardButton("فارسی 🇮🇷", callback_data='set_lang_fa')], [InlineKeyboardButton("English 🇬🇧", callback_data='set_lang_en')] ] reply_markup = InlineKeyboardMarkup(keyboard) await update.message.reply_text( "زبان مورد نظر خود را انتخاب کنید:\nChoose your preferred language:", reply_markup=reply_markup ) return LANGUAGE_SELECTION async def set_language(update: Update, context): query = update.callback_query await query.answer() lang_code = query.data.replace('set_lang_', '') context.user_data['language'] = lang_code welcome_message = get_message(context, 'start_welcome') await query.edit_message_text(text=welcome_message) return await process_feature_or_check_membership(update, context, show_main_menu) async def show_main_menu(update: Update, context): keyboard = [ [InlineKeyboardButton(get_message(context, 'btn_convert_format'), callback_data='select_convert_format')], [InlineKeyboardButton(get_message(context, 'btn_cut_audio'), callback_data='select_cut_audio')], [InlineKeyboardButton(get_message(context, 'btn_video_conversion'), callback_data='select_video_conversion')] ] reply_markup = InlineKeyboardMarkup(keyboard) if update.callback_query: await update.callback_query.edit_message_text( text=get_message(context, 'main_menu_prompt'), reply_markup=reply_markup ) else: await update.effective_message.reply_text( text=get_message(context, 'main_menu_prompt'), reply_markup=reply_markup ) return MAIN_MENU async def change_format_selected(update: Update, context): query = update.callback_query await query.answer() await query.edit_message_text(text=get_message(context, 'convert_mode_active')) return CONVERT_AUDIO async def cut_audio_selected(update: Update, context): query = update.callback_query await query.answer() await query.edit_message_text(text=get_message(context, 'cut_mode_active_file')) context.user_data['audio_for_cut_path'] = None return CUT_AUDIO_FILE async def video_conversion_selected(update: Update, context): query = update.callback_query await query.answer() await query.edit_message_text(text=get_message(context, 'video_conversion_mode_active')) return VIDEO_CONVERSION_MODE async def handle_audio(update: Update, context): if not update.message.audio: return CONVERT_AUDIO async def _perform_conversion(update: Update, context): file_id = update.message.audio.file_id file_name = update.message.audio.file_name if update.message.audio.file_name else f"audio_{file_id}.mp3" processing_message = await update.message.reply_text(get_message(context, 'processing_start')) try: new_file = await context.bot.get_file(file_id) input_bytes = BytesIO(await new_file.download_as_bytes()) await processing_message.edit_text(get_message(context, 'file_received')) audio = AudioSegment.from_mp3(input_bytes) output_buffer = BytesIO() audio.export(output_buffer, format="ogg", codec="libopus", parameters=["-b:a", "32k"]) output_buffer.seek(0) await processing_message.edit_text(get_message(context, 'conversion_done')) await update.message.reply_voice(InputFile(output_buffer, filename=f"{os.path.splitext(file_name)[0]}.ogg")) await update.message.reply_text(get_message(context, 'mp3_to_voice_reply')) await processing_message.delete() return CONVERT_AUDIO except Exception as e: print(f"Error converting MP3 to voice: {e}", file=sys.stderr) if processing_message: await processing_message.edit_text(get_message(context, 'error_mp3_to_voice') + str(e)) else: await update.message.reply_text(get_message(context, 'error_mp3_to_voice') + str(e)) return CONVERT_AUDIO return await process_feature_or_check_membership(update, context, _perform_conversion) async def handle_voice(update: Update, context): if not update.message.voice: return CONVERT_AUDIO async def _perform_conversion(update: Update, context): file_id = update.message.voice.file_id processing_message = await update.message.reply_text(get_message(context, 'processing_start')) try: new_file = await context.bot.get_file(file_id) input_bytes = BytesIO(await new_file.download_as_bytes()) await processing_message.edit_text(get_message(context, 'file_received')) audio = AudioSegment.from_file(input_bytes, format="ogg") output_buffer = BytesIO() export_tags = { 'album': BOT_USERNAME, 'artist': BOT_USERNAME } audio.export(output_buffer, format="mp3", tags=export_tags) output_buffer.seek(0) await processing_message.edit_text(get_message(context, 'conversion_done')) await update.message.reply_audio( InputFile(output_buffer, filename=f"@{BOT_USERNAME}_{file_id}.mp3"), caption=get_message(context, 'voice_to_mp3_caption') ) await processing_message.delete() return CONVERT_AUDIO except Exception as e: print(f"Error converting voice to MP3: {e}", file=sys.stderr) if processing_message: await processing_message.edit_text(get_message(context, 'error_voice_to_mp3') + str(e)) else: await update.message.reply_text(get_message(context, 'error_voice_to_mp3') + str(e)) return CONVERT_AUDIO return await process_feature_or_check_membership(update, context, _perform_conversion) async def handle_cut_audio_file(update: Update, context): audio_file = update.message.audio or update.message.voice if not audio_file: await update.message.reply_text(get_message(context, 'no_audio_for_cut')) return CUT_AUDIO_FILE async def _perform_file_receive(update: Update, context): file_id = audio_file.file_id file_type = 'mp3' if update.message.audio else 'ogg' processing_message = await update.message.reply_text(get_message(context, 'processing_start')) try: new_file = await context.bot.get_file(file_id) context.user_data['audio_for_cut_bytes'] = await new_file.download_as_bytes() context.user_data['audio_for_cut_type'] = file_type context.user_data['audio_for_cut_original_filename'] = update.message.audio.file_name if update.message.audio else f"voice_{file_id}.ogg" await processing_message.edit_text(get_message(context, 'cut_mode_active_range')) await processing_message.delete() return CUT_AUDIO_RANGE except Exception as e: print(f"Error handling cut audio file: {e}", file=sys.stderr) if processing_message: await processing_message.edit_text(get_message(context, 'general_error') + str(e)) else: await update.message.reply_text(get_message(context, 'general_error') + str(e)) return CUT_AUDIO_FILE return await process_feature_or_check_membership(update, context, _perform_file_receive) async def handle_cut_audio_range(update: Update, context): time_range_str = update.message.text audio_bytes = context.user_data.get('audio_for_cut_bytes') audio_type = context.user_data.get('audio_for_cut_type') original_filename = context.user_data.get('audio_for_cut_original_filename', "audio_cut") if not audio_bytes: await update.message.reply_text(get_message(context, 'no_audio_for_cut')) return CUT_AUDIO_FILE async def _perform_cut(update: Update, context): processing_message = await update.message.reply_text(get_message(context, 'cut_processing')) try: start_time_str, end_time_str = time_range_str.split('-') start_ms = parse_time_to_ms(start_time_str.strip()) end_ms = parse_time_to_ms(end_time_str.strip()) if start_ms >= end_ms: await processing_message.edit_text(get_message(context, 'invalid_time_range')) return CUT_AUDIO_RANGE input_buffer = BytesIO(audio_bytes) audio = AudioSegment.from_file(input_buffer, format=audio_type) cut_audio = audio[start_ms:end_ms] output_buffer = BytesIO() cut_audio.export(output_buffer, format="mp3") output_buffer.seek(0) await processing_message.edit_text(get_message(context, 'audio_cut_success')) await update.message.reply_audio(InputFile(output_buffer, filename=f"cut_{os.path.splitext(original_filename)[0]}.mp3"), caption=f"برش از {start_time_str} تا {end_time_str}") await processing_message.delete() return await show_main_menu(update, context) except ValueError as ve: print(f"Time format or range error: {ve}", file=sys.stderr) await processing_message.edit_text(get_message(context, 'invalid_time_format')) return CUT_AUDIO_RANGE except Exception as e: print(f"Error cutting audio: {e}", file=sys.stderr) if processing_message: await processing_message.edit_text(get_message(context, 'general_error') + str(e)) else: await update.message.reply_text(get_message(context, 'general_error') + str(e)) return CUT_AUDIO_FILE finally: if 'audio_for_cut_bytes' in context.user_data: del context.user_data['audio_for_cut_bytes'] if 'audio_for_cut_type' in context.user_data: del context.user_data['audio_for_cut_type'] if 'audio_for_cut_original_filename' in context.user_data: del context.user_data['audio_for_cut_original_filename'] return await process_feature_or_check_membership(update, context, _perform_cut) async def handle_video_conversion(update: Update, context): file_to_process = None is_video_note = False if update.message.video: file_to_process = update.message.video original_file_name = update.message.video.file_name if update.message.video.file_name else f"video_{file_to_process.file_id}.mp4" elif update.message.video_note: file_to_process = update.message.video_note is_video_note = True original_file_name = f"videonote_{file_to_process.file_id}.mp4" else: await update.message.reply_text(get_message(context, 'invalid_file_type_video')) return VIDEO_CONVERSION_MODE async def _perform_video_conversion(update: Update, context): file_id = file_to_process.file_id processing_message = await update.message.reply_text(get_message(context, 'processing_start')) temp_input_path = None temp_output_path = None try: new_file = await context.bot.get_file(file_id) input_video_bytes = await new_file.download_as_bytes() await processing_message.edit_text(get_message(context, 'file_received_video')) # استفاده از NamedTemporaryFile برای FFmpeg with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as temp_input_file: temp_input_file.write(input_video_bytes) temp_input_path = temp_input_file.name temp_output_path = tempfile.mktemp(suffix=".mp4") if is_video_note: await processing_message.edit_text(get_message(context, 'converting_video_note_to_video')) ffmpeg_command = [ 'ffmpeg', '-i', temp_input_path, '-c:v', 'libx264', '-crf', '23', '-preset', 'medium', '-c:a', 'aac', '-b:a', '128k', '-movflags', '+faststart', temp_output_path ] result = subprocess.run(ffmpeg_command, check=True, capture_output=True) await processing_message.edit_text(get_message(context, 'conversion_done_video')) with open(temp_output_path, 'rb') as f: output_file_bytes = BytesIO(f.read()) await update.message.reply_video( InputFile(output_file_bytes, filename=f"{os.path.splitext(original_file_name)[0]}_regular.mp4"), caption=get_message(context, 'video_note_to_video_caption') ) else: await processing_message.edit_text(get_message(context, 'converting_video_to_video_note')) ffmpeg_command = [ 'ffmpeg', '-i', temp_input_path, '-vf', 'crop=min(iw\,ih):min(iw\,ih),scale=640:640', '-c:v', 'libx264', '-crf', '28', '-preset', 'veryfast', '-c:a', 'aac', '-b:a', '64k', '-movflags', '+faststart', temp_output_path ] result = subprocess.run(ffmpeg_command, check=True, capture_output=True) await processing_message.edit_text(get_message(context, 'conversion_done_video')) with open(temp_output_path, 'rb') as f: output_file_bytes = BytesIO(f.read()) await update.message.reply_video_note(InputFile(output_file_bytes, filename=f"{os.path.splitext(original_file_name)[0]}_note.mp4")) await update.message.reply_text(get_message(context, 'video_to_video_note_reply')) await processing_message.delete() return VIDEO_CONVERSION_MODE except subprocess.CalledProcessError as e: print(f"FFmpeg error (stdout): {e.stdout.decode()}\nFFmpeg error (stderr): {e.stderr.decode()}", file=sys.stderr) if processing_message: await processing_message.edit_text(get_message(context, 'error_video_conversion') + f"FFmpeg Error: {e.stderr.decode()}") else: await update.message.reply_text(get_message(context, 'error_video_conversion') + f"FFmpeg Error: {e.stderr.decode()}") return VIDEO_CONVERSION_MODE except Exception as e: print(f"General video conversion error: {e}", file=sys.stderr) if processing_message: await processing_message.edit_text(get_message(context, 'error_video_conversion') + str(e)) else: await update.message.reply_text(get_message(context, 'error_video_conversion') + str(e)) return VIDEO_CONVERSION_MODE finally: if temp_input_path and os.path.exists(temp_input_path): os.remove(temp_input_path) if temp_output_path and os.path.exists(temp_output_path): os.remove(temp_output_path) return await process_feature_or_check_membership(update, context, _perform_video_conversion) async def cancel(update: Update, context): await update.message.reply_text(get_message(context, 'cancel_message')) if 'audio_for_cut_bytes' in context.user_data: del context.user_data['audio_for_cut_bytes'] if 'audio_for_cut_type' in context.user_data: del context.user_data['audio_for_cut_type'] if 'audio_for_cut_original_filename' in context.user_data: del context.user_data['audio_for_cut_original_filename'] context.user_data['is_member'] = False context.user_data['used_bot_once'] = False return await show_main_menu(update, context) async def error_handler(update: Update, context): print(f"Update {update} caused error {context.error}", file=sys.stderr) if update.message: await update.message.reply_text(get_message(context, 'general_error')) return ConversationHandler.END # --- Admin Functionality --- async def admin_link_command(update: Update, context): if update.effective_user.id != ADMIN_ID: await update.message.reply_text(get_message(context, 'not_admin')) return ConversationHandler.END keyboard = [ [InlineKeyboardButton(get_message(context, 'btn_add_channel'), callback_data='admin_add_channel')], [InlineKeyboardButton(get_message(context, 'btn_list_channels'), callback_data='admin_list_channels')] ] reply_markup = InlineKeyboardMarkup(keyboard) await update.message.reply_text(get_message(context, 'admin_menu_prompt'), reply_markup=reply_markup) return ADMIN_MENU async def admin_add_channel_prompt(update: Update, context): query = update.callback_query await query.answer() await query.edit_message_text(get_message(context, 'send_channel_link')) return ADD_CHANNEL async def admin_handle_add_channel(update: Update, context): channel_input = update.message.text.strip() if not (channel_input.startswith('@') or channel_input.startswith('-100')): await update.message.reply_text(get_message(context, 'invalid_channel_id')) return ADD_CHANNEL try: chat_member = await context.bot.get_chat_member(chat_id=channel_input, user_id=context.bot.id) if not (chat_member.status == 'administrator' and chat_member.can_restrict_members): # از can_restrict_members استفاده می کنیم await update.message.reply_text(get_message(context, 'bot_not_admin_in_channel', channel_id=channel_input)) return ADD_CHANNEL except TelegramError as e: print(f"Telegram error checking bot's admin status in channel {channel_input}: {e}", file=sys.stderr) await update.message.reply_text(get_message(context, 'invalid_channel_id') + f"\nError: {e}") return ADD_CHANNEL except Exception as e: print(f"Error checking bot's admin status in channel {channel_input}: {e}", file=sys.stderr) await update.message.reply_text(get_message(context, 'invalid_channel_id') + f"\nError: {e}") return ADD_CHANNEL if channel_input not in REQUIRED_CHANNELS: REQUIRED_CHANNELS.append(channel_input) save_required_channels(REQUIRED_CHANNELS) await update.message.reply_text(get_message(context, 'channel_added', channel_id=channel_input)) else: await update.message.reply_text(get_message(context, 'channel_already_exists')) return await admin_link_command(update, context) async def admin_list_channels(update: Update, context): query = update.callback_query await query.answer() if not REQUIRED_CHANNELS: await query.edit_message_text(get_message(context, 'no_channels_configured')) return ADMIN_MENU keyboard = [] message_text = get_message(context, 'channel_list_prompt') + "\n\n" for i, channel_id in enumerate(REQUIRED_CHANNELS): try: chat = await context.bot.get_chat(chat_id=channel_id) channel_name = chat.title if chat.title else channel_id message_text += f"{i+1}. {channel_name} (`{channel_id}`)\n" keyboard.append([InlineKeyboardButton(f"{get_message(context, 'btn_remove_channel')} {channel_name}", callback_data=f'remove_channel_{channel_id}')]) except Exception as e: message_text += f"{i+1}. {channel_id} (خطا در دریافت اطلاعات: {e})\n" keyboard.append([InlineKeyboardButton(f"{get_message(context, 'btn_remove_channel')} {channel_id}", callback_data=f'remove_channel_{channel_id}')]) reply_markup = InlineKeyboardMarkup(keyboard) await query.edit_message_text(message_text, reply_markup=reply_markup, parse_mode='Markdown') return LIST_REMOVE_CHANNELS async def admin_handle_remove_channel(update: Update, context): query = update.callback_query await query.answer() channel_id_to_remove = query.data.replace('remove_channel_', '') if channel_id_to_remove in REQUIRED_CHANNELS: REQUIRED_CHANNELS.remove(channel_id_to_remove) save_required_channels(REQUIRED_CHANNELS) await query.edit_message_text(get_message(context, 'channel_removed', channel_id=channel_id_to_remove)) else: await query.edit_message_text(get_message(context, 'channel_not_found')) return await admin_link_command(update, context) # --- ادغام با FastAPI برای Webhook --- WEBHOOK_PATH = "/webhook" # اضافه کردن خط دیباگ در ابتدای فایل برای لاگهای Build/Startup print("DEBUG: app.py is starting up.") application = Application.builder().token(TOKEN).build() app = FastAPI( title="Telegram Bot Webhook", description="Backend for Telegram Bot running on Hugging Face Spaces with Webhook", version="1.0.0" ) # تعریف هندلرها conv_handler = ConversationHandler( entry_points=[ CommandHandler("start", start), CommandHandler("link", admin_link_command) ], states={ LANGUAGE_SELECTION: [ CallbackQueryHandler(set_language, pattern='^set_lang_') ], MAIN_MENU: [ CallbackQueryHandler(change_format_selected, pattern='^select_convert_format$'), CallbackQueryHandler(cut_audio_selected, pattern='^select_cut_audio$'), CallbackQueryHandler(video_conversion_selected, pattern='^select_video_conversion$') ], CONVERT_AUDIO: [ MessageHandler(filters.AUDIO & ~filters.COMMAND, handle_audio), MessageHandler(filters.VOICE & ~filters.COMMAND, handle_voice), CommandHandler("start", start), CommandHandler("cancel", cancel) ], CUT_AUDIO_FILE: [ MessageHandler(filters.AUDIO & ~filters.COMMAND, handle_cut_audio_file), MessageHandler(filters.VOICE & ~filters.COMMAND, handle_cut_audio_file), CommandHandler("start", start), CommandHandler("cancel", cancel) ], CUT_AUDIO_RANGE: [ MessageHandler(filters.TEXT & ~filters.COMMAND, handle_cut_audio_range), CommandHandler("start", start), CommandHandler("cancel", cancel) ], VIDEO_CONVERSION_MODE: [ MessageHandler(filters.VIDEO & ~filters.COMMAND, handle_video_conversion), MessageHandler(filters.VIDEO_NOTE & ~filters.COMMAND, handle_video_conversion), CommandHandler("start", start), CommandHandler("cancel", cancel) ], WAITING_FOR_MEMBERSHIP: [ CallbackQueryHandler(check_membership_callback, pattern='^check_membership$'), CommandHandler("start", start), CommandHandler("cancel", cancel) ], ADMIN_MENU: [ CallbackQueryHandler(admin_add_channel_prompt, pattern='^admin_add_channel$'), CallbackQueryHandler(admin_list_channels, pattern='^admin_list_channels$'), CommandHandler("start", start), CommandHandler("cancel", cancel) ], ADD_CHANNEL: [ MessageHandler(filters.TEXT & ~filters.COMMAND, admin_handle_add_channel), CommandHandler("start", start), CommandHandler("cancel", cancel) ], LIST_REMOVE_CHANNELS: [ CallbackQueryHandler(admin_handle_remove_channel, pattern='^remove_channel_'), CommandHandler("start", start), CommandHandler("cancel", cancel) ] }, fallbacks=[CommandHandler("cancel", cancel), CommandHandler("start", start)], allow_reentry=True ) application.add_handler(conv_handler) application.add_error_handler(error_handler) # FastAPI app events @app.on_event("startup") async def on_startup(): print("Application startup...") # Initialize the PTB application. # این خط لازم است تا application هندلرها را بشناسد و آماده پردازش آپدیت ها باشد. await application.initialize() @app.on_event("shutdown") async def on_shutdown(): print("Application shutdown...") # Terminate PTB application gracefully await application.shutdown() @app.post(WEBHOOK_PATH) async def bot_webhook(request: Request): """ Endpoint for Telegram Webhook. Receives updates from Telegram. """ # اضافه کردن خط دیباگ در لحظه دریافت آپدیت print("DEBUG: Received webhook update.") json_data = await request.json() print(f"DEBUG: Update JSON: {json.dumps(json_data, indent=2)}") # لاگ کردن محتوای آپدیت برای عیب یابی update = Update.de_json(json_data, application.bot) # Process the update using the PTB application await application.process_update(update) return {"ok": True} @app.get("/") async def root(): """Home endpoint to show bot status or a welcome message.""" return HTMLResponse("
Send messages to your bot on Telegram.
")