Spaces:
Running
Running
# -*- coding: utf-8 -*- | |
import os | |
import sys | |
import re | |
import subprocess | |
import json | |
from io import BytesIO | |
from flask import Flask, request, jsonify | |
from telegram import Update, InputFile, InlineKeyboardButton, InlineKeyboardMarkup | |
from telegram.ext import ( | |
Application, | |
CommandHandler, | |
MessageHandler, | |
CallbackQueryHandler, | |
filters, | |
ConversationHandler | |
) | |
from telegram.constants import ParseMode | |
from pydub import AudioSegment | |
# --- 1. تنظیمات و متغیرهای سراسری --- | |
# ⚠️⚠️⚠️ بسیار مهم: این توکن را با توکن واقعی ربات خود که از BotFather دریافت کردهاید جایگزین کنید. | |
# میتوانید توکن را به عنوان یک متغیر محیطی (Environment Variable) به نام TELEGRAM_BOT_TOKEN نیز تعریف کنید. | |
TOKEN = os.getenv("TELEGRAM_BOT_TOKEN", "8035336072:AAHG9REotvM4u8DgreC7hu8gIAroxhS-N-M") | |
if "YOUR_TELEGRAM_BOT_TOKEN_HERE" in TOKEN: | |
print("خطا: لطفا توکن ربات تلگرام خود را در کد جایگزین کنید.", file=sys.stderr) | |
sys.exit(1) | |
# نام کاربری ربات و آیدی ادمین | |
BOT_USERNAME = os.getenv("TELEGRAM_BOT_USERNAME", "Voice2mp3_RoBot") | |
# ⚠️⚠️⚠️ این را به آیدی عددی تلگرام خودتان تغییر دهید. | |
ADMIN_ID = int(os.getenv("TELEGRAM_ADMIN_ID", "684173337")) | |
# مسیرها | |
# در محیطهای سرورلس مانند Hugging Face Spaces، استفاده از /tmp توصیه میشود | |
DOWNLOAD_DIR = "/tmp/downloads" | |
OUTPUT_DIR = "/tmp/outputs" | |
CHANNELS_FILE = "/tmp/channels.json" | |
os.makedirs(DOWNLOAD_DIR, exist_ok=True) | |
os.makedirs(OUTPUT_DIR, exist_ok=True) | |
# --- 2. توابع مدیریت کانالهای عضویت اجباری --- | |
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"خطا در خواندن فایل JSON کانالها: {CHANNELS_FILE}", 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() | |
# --- 3. تعریف حالتهای مکالمه و پیامها --- | |
# تعریف حالتها | |
(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}' 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." | |
} | |
} | |
# --- 4. تمام توابع کمکی و منطقی ربات --- | |
def get_message(context, key, **kwargs): | |
"""پیام مناسب را بر اساس زبان کاربر برمیگرداند.""" | |
lang = context.user_data.get('language', 'fa') | |
# Fallback to Persian if language or key is not found | |
message_template = MESSAGES.get(lang, MESSAGES['fa']).get(key, MESSAGES['fa'].get(key, "Message key not found")) | |
return message_template.format(**kwargs) | |
def parse_time_to_ms(time_str): | |
"""رشته 'MM.SS' را به میلیثانیه تبدیل میکند.""" | |
match = re.match(r'^(\d{2})\.(\d{2})$', time_str) | |
if not match: | |
raise ValueError("Invalid time format") | |
minutes, seconds = int(match.group(1)), 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 | |
if not REQUIRED_CHANNELS: | |
return 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']: | |
return False | |
except Exception as e: | |
print(f"خطا در بررسی عضویت کانال {channel_id}: {e}", file=sys.stderr) | |
return False | |
return True | |
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) | |
url = chat.invite_link or (f"https://t.me/{chat.username}" if chat.username else None) | |
if url: | |
keyboard.append([InlineKeyboardButton(f"{get_message(context, 'btn_join_channel')} {chat.title or channel_id}", url=url)]) | |
except Exception as e: | |
print(f"ناتوان در دریافت اطلاعات کانال {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) | |
message_sender = update.callback_query.edit_message_text if update.callback_query else update.effective_message.reply_text | |
await message_sender(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): | |
"""یک میانافزار برای بررسی عضویت قبل از اجرای قابلیتها.""" | |
if update.effective_user.id == ADMIN_ID or not REQUIRED_CHANNELS: | |
return await feature_func(update, context, *args, **kwargs) | |
is_member = await check_user_membership(update, context) | |
if is_member: | |
context.user_data['is_member'] = True | |
return await feature_func(update, context, *args, **kwargs) | |
else: | |
context.user_data['is_member'] = False | |
return await show_membership_required_message(update, context) | |
# --- 5. تمام توابع Handler (مدیریت دستورات و پیامها) --- | |
async def start(update: Update, context): | |
"""دستور /start را مدیریت کرده و درخواست انتخاب زبان میکند.""" | |
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() | |
context.user_data['language'] = query.data.replace('set_lang_', '') | |
await query.edit_message_text(text=get_message(context, 'start_welcome')) | |
return await show_main_menu(update, context) | |
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) | |
message_sender = update.callback_query.edit_message_text if update.callback_query else update.message.reply_text | |
await message_sender(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.pop('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): | |
"""فایل MP3 را به ویس تلگرام تبدیل میکند.""" | |
async def _perform_conversion(update, context): | |
file_id = update.message.audio.file_id | |
file_name = update.message.audio.file_name or f"audio_{file_id}.mp3" | |
processing_message = await update.message.reply_text(get_message(context, 'processing_start')) | |
download_path = os.path.join(DOWNLOAD_DIR, f"in_{file_id}.mp3") | |
output_ogg_path = os.path.join(OUTPUT_DIR, f"out_{file_id}.ogg") | |
try: | |
new_file = await context.bot.get_file(file_id) | |
await new_file.download_to_drive(download_path) | |
await processing_message.edit_text(get_message(context, 'file_received')) | |
audio = AudioSegment.from_file(download_path) | |
audio.export(output_ogg_path, format="ogg", codec="libopus", parameters=["-b:a", "32k"]) | |
await processing_message.edit_text(get_message(context, 'conversion_done')) | |
with open(output_ogg_path, 'rb') as f: | |
await update.message.reply_voice(f, reply_to_message_id=update.message.message_id) | |
await processing_message.delete() | |
except Exception as e: | |
print(f"خطا در تبدیل MP3 به ویس: {e}", file=sys.stderr) | |
await processing_message.edit_text(get_message(context, 'error_mp3_to_voice') + str(e)) | |
finally: | |
if os.path.exists(download_path): os.remove(download_path) | |
if os.path.exists(output_ogg_path): os.remove(output_ogg_path) | |
return CONVERT_AUDIO | |
return await process_feature_or_check_membership(update, context, _perform_conversion) | |
async def handle_voice(update: Update, context): | |
"""ویس تلگرام را به فایل MP3 تبدیل میکند.""" | |
async def _perform_conversion(update, context): | |
file_id = update.message.voice.file_id | |
processing_message = await update.message.reply_text(get_message(context, 'processing_start')) | |
download_path = os.path.join(DOWNLOAD_DIR, f"in_{file_id}.ogg") | |
output_mp3_path = os.path.join(OUTPUT_DIR, f"@{BOT_USERNAME}_{file_id}.mp3") | |
try: | |
new_file = await context.bot.get_file(file_id) | |
await new_file.download_to_drive(download_path) | |
await processing_message.edit_text(get_message(context, 'file_received')) | |
audio = AudioSegment.from_file(download_path, format="ogg") | |
audio.export(output_mp3_path, format="mp3", tags={'album': BOT_USERNAME, 'artist': BOT_USERNAME}) | |
await processing_message.edit_text(get_message(context, 'conversion_done')) | |
with open(output_mp3_path, 'rb') as f: | |
await update.message.reply_audio(f, caption=get_message(context, 'voice_to_mp3_caption'), reply_to_message_id=update.message.message_id) | |
await processing_message.delete() | |
except Exception as e: | |
print(f"خطا در تبدیل ویس به MP3: {e}", file=sys.stderr) | |
await processing_message.edit_text(get_message(context, 'error_voice_to_mp3') + str(e)) | |
finally: | |
if os.path.exists(download_path): os.remove(download_path) | |
if os.path.exists(output_mp3_path): os.remove(output_mp3_path) | |
return CONVERT_AUDIO | |
return await process_feature_or_check_membership(update, context, _perform_conversion) | |
async def handle_cut_audio_file(update: Update, context): | |
"""فایل صوتی برای برش را دریافت میکند.""" | |
async def _perform_file_receive(update, context): | |
audio_file = update.message.audio or update.message.voice | |
file_id = audio_file.file_id | |
ext = 'mp3' if update.message.audio else 'ogg' | |
download_path = os.path.join(DOWNLOAD_DIR, f"cut_in_{file_id}.{ext}") | |
try: | |
new_file = await context.bot.get_file(file_id) | |
await new_file.download_to_drive(download_path) | |
context.user_data['audio_for_cut_path'] = download_path | |
context.user_data['audio_for_cut_type'] = ext | |
await update.message.reply_text(get_message(context, 'cut_mode_active_range')) | |
return CUT_AUDIO_RANGE | |
except Exception as e: | |
print(f"خطا در دریافت فایل صوتی برای برش: {e}", file=sys.stderr) | |
await update.message.reply_text(get_message(context, 'general_error')) | |
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): | |
"""بازه زمانی را دریافت کرده و صدا را برش میدهد.""" | |
async def _perform_cut(update, context): | |
time_range_str = update.message.text | |
audio_path = context.user_data.get('audio_for_cut_path') | |
audio_type = context.user_data.get('audio_for_cut_type') | |
if not audio_path or not os.path.exists(audio_path): | |
await update.message.reply_text(get_message(context, 'no_audio_for_cut')) | |
return CUT_AUDIO_FILE | |
processing_message = await update.message.reply_text(get_message(context, 'cut_processing')) | |
output_cut_path = os.path.join(OUTPUT_DIR, f"cut_out_{os.path.basename(audio_path)}.mp3") | |
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 | |
audio = AudioSegment.from_file(audio_path, format=audio_type) | |
cut_audio = audio[start_ms:end_ms] | |
cut_audio.export(output_cut_path, format="mp3") | |
await processing_message.edit_text(get_message(context, 'audio_cut_success')) | |
with open(output_cut_path, 'rb') as f: | |
await update.message.reply_audio(f, caption=f"برش از {start_time_str} تا {end_time_str}") | |
await processing_message.delete() | |
return await show_main_menu(update, context) | |
except ValueError: | |
await processing_message.edit_text(get_message(context, 'invalid_time_format')) | |
return CUT_AUDIO_RANGE | |
except Exception as e: | |
print(f"خطا در برش صدا: {e}", file=sys.stderr) | |
await processing_message.edit_text(get_message(context, 'general_error')) | |
return await show_main_menu(update, context) | |
finally: | |
if os.path.exists(audio_path): os.remove(audio_path) | |
if os.path.exists(output_cut_path): os.remove(output_cut_path) | |
context.user_data.pop('audio_for_cut_path', None) | |
context.user_data.pop('audio_for_cut_type', None) | |
return await process_feature_or_check_membership(update, context, _perform_cut) | |
async def handle_video_conversion(update: Update, context): | |
"""تبدیل بین ویدیو معمولی و دایرهای را انجام میدهد.""" | |
async def _perform_video_conversion(update, context): | |
is_video_note = bool(update.message.video_note) | |
file_to_process = update.message.video_note if is_video_note else update.message.video | |
file_id = file_to_process.file_id | |
download_path = os.path.join(DOWNLOAD_DIR, f"vid_in_{file_id}.mp4") | |
output_path = os.path.join(OUTPUT_DIR, f"vid_out_{file_id}.mp4") | |
processing_message = await update.message.reply_text(get_message(context, 'processing_start')) | |
try: | |
new_file = await context.bot.get_file(file_id) | |
await new_file.download_to_drive(download_path) | |
await processing_message.edit_text(get_message(context, 'file_received_video')) | |
if is_video_note: | |
await processing_message.edit_text(get_message(context, 'converting_video_note_to_video')) | |
ffmpeg_command = ['ffmpeg', '-i', download_path, '-c:v', 'libx264', '-crf', '23', '-preset', 'medium', '-c:a', 'aac', '-b:a', '128k', '-movflags', '+faststart', output_path] | |
else: | |
await processing_message.edit_text(get_message(context, 'converting_video_to_video_note')) | |
ffmpeg_command = ['ffmpeg', '-i', download_path, '-vf', 'crop=min(iw\,ih):min(iw\,ih),scale=640:640', '-c:v', 'libx264', '-crf', '28', '-preset', 'veryfast', '-an', output_path] | |
subprocess.run(ffmpeg_command, check=True, capture_output=True) | |
await processing_message.edit_text(get_message(context, 'conversion_done_video')) | |
with open(output_path, 'rb') as f: | |
if is_video_note: | |
await update.message.reply_video(f, caption=get_message(context, 'video_note_to_video_caption')) | |
else: | |
await update.message.reply_video_note(f) | |
await processing_message.delete() | |
except subprocess.CalledProcessError as e: | |
print(f"خطای FFmpeg: {e.stderr.decode()}", file=sys.stderr) | |
await processing_message.edit_text(get_message(context, 'error_video_conversion') + "خطای FFmpeg") | |
except Exception as e: | |
print(f"خطای کلی تبدیل ویدیو: {e}", file=sys.stderr) | |
await processing_message.edit_text(get_message(context, 'error_video_conversion') + str(e)) | |
finally: | |
if os.path.exists(download_path): os.remove(download_path) | |
if os.path.exists(output_path): os.remove(output_path) | |
return VIDEO_CONVERSION_MODE | |
return await process_feature_or_check_membership(update, context, _perform_video_conversion) | |
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.answer(get_message(context, 'membership_failed'), show_alert=True) | |
return WAITING_FOR_MEMBERSHIP | |
async def admin_link_command(update: Update, context): | |
"""دستور /link برای ادمین.""" | |
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')] | |
] | |
await update.message.reply_text(get_message(context, 'admin_menu_prompt'), reply_markup=InlineKeyboardMarkup(keyboard)) | |
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: | |
await context.bot.get_chat(channel_input) # Check if bot can access the 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')) | |
except Exception as e: | |
await update.message.reply_text(get_message(context, 'bot_not_admin_in_channel', channel_id=channel_input) + f"\nError: {e}") | |
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 = [] | |
for channel_id in REQUIRED_CHANNELS: | |
keyboard.append([InlineKeyboardButton(f"{get_message(context, 'btn_remove_channel')} {channel_id}", callback_data=f'remove_channel_{channel_id}')]) | |
await query.edit_message_text(get_message(context, 'channel_list_prompt'), reply_markup=InlineKeyboardMarkup(keyboard)) | |
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) | |
async def cancel(update: Update, context): | |
"""عملیات را لغو کرده و به منوی اصلی برمیگردد.""" | |
message = update.message or update.callback_query.message | |
await message.reply_text(get_message(context, 'cancel_message')) | |
context.user_data.clear() | |
return await show_main_menu(update, context) | |
async def error_handler(update: object, context: object) -> None: | |
"""لاگ کردن خطاها.""" | |
print(f"Update {update} caused error {context.error}", file=sys.stderr) | |
# --- 6. بخش مربوط به Flask و Webhook --- | |
app = Flask(__name__) | |
_application_instance = None | |
async def get_telegram_application(): | |
"""اپلیکیشن تلگرام را مقداردهی اولیه کرده و ConversationHandler کامل را به آن اضافه میکند.""" | |
global _application_instance | |
if _application_instance is None: | |
print("Initializing Telegram bot Application for this worker...") | |
application = Application.builder().token(TOKEN).build() | |
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), | |
], | |
CUT_AUDIO_FILE: [MessageHandler((filters.AUDIO | filters.VOICE) & ~filters.COMMAND, handle_cut_audio_file)], | |
CUT_AUDIO_RANGE: [MessageHandler(filters.TEXT & ~filters.COMMAND, handle_cut_audio_range)], | |
VIDEO_CONVERSION_MODE: [MessageHandler((filters.VIDEO | filters.VIDEO_NOTE) & ~filters.COMMAND, handle_video_conversion)], | |
WAITING_FOR_MEMBERSHIP: [CallbackQueryHandler(check_membership_callback, pattern='^check_membership$')], | |
ADMIN_MENU: [ | |
CallbackQueryHandler(admin_add_channel_prompt, pattern='^admin_add_channel$'), | |
CallbackQueryHandler(admin_list_channels, pattern='^admin_list_channels$'), | |
], | |
ADD_CHANNEL: [MessageHandler(filters.TEXT & ~filters.COMMAND, admin_handle_add_channel)], | |
LIST_REMOVE_CHANNELS: [CallbackQueryHandler(admin_handle_remove_channel, pattern='^remove_channel_')], | |
}, | |
fallbacks=[CommandHandler("cancel", cancel), CommandHandler("start", start)], | |
allow_reentry=True | |
) | |
application.add_handler(conv_handler) | |
application.add_error_handler(error_handler) | |
await application.initialize() | |
_application_instance = application | |
print("Telegram bot Application initialized.") | |
return _application_instance | |
def index(): | |
"""یک صفحه ساده برای اطمینان از بالا بودن سرور.""" | |
return "Hello, I am the Telegram bot server. The bot is running." | |
async def webhook(): | |
"""این مسیر آپدیتها را از تلگرام دریافت میکند.""" | |
try: | |
application = await get_telegram_application() | |
update = Update.de_json(request.get_json(force=True), application.bot) | |
await application.process_update(update) | |
return "ok" | |
except Exception as e: | |
print(f"Error in webhook: {e}", file=sys.stderr) | |
return "error", 500 | |
async def set_webhook_route(): | |
"""یک مسیر برای تنظیم وبهوک به صورت خودکار.""" | |
webhook_url = os.getenv("WEBHOOK_URL") | |
if not webhook_url: | |
return jsonify({"status": "error", "message": "WEBHOOK_URL environment variable not set."}), 500 | |
if not webhook_url.endswith("/webhook"): | |
webhook_url = f"{webhook_url.rstrip('/')}/webhook" | |
try: | |
application = await get_telegram_application() | |
await application.bot.set_webhook(url=webhook_url) | |
return jsonify({"status": "success", "message": f"Webhook set to {webhook_url}"}) | |
except Exception as e: | |
print(f"Failed to set webhook: {e}", file=sys.stderr) | |
return jsonify({"status": "error", "message": f"Failed to set webhook: {e}"}), 500 | |
if __name__ == "__main__": | |
# این بخش برای اجرای محلی (local) سرور Flask است | |
# در محیطهای production مانند Hugging Face، یک وب سرور مانند Gunicorn این فایل را اجرا میکند. | |
app.run(host='0.0.0.0', port=int(os.environ.get("PORT", 7860))) | |