| from pyrogram import Client, filters |
| from pyrogram.types import Message, InlineKeyboardMarkup, InlineKeyboardButton, CallbackQuery |
| from helper.database import codeflixbots as db |
| from utils.job_queue import process_queue |
| import os |
| import asyncio |
| from pyromod.exceptions import ListenerTimeout |
| from config import Config |
|
|
| command_filter = filters.create(lambda _, __, msg: bool(msg.text and msg.text.startswith("/"))) |
|
|
| @Client.on_message(filters.command("set_name") & filters.private) |
| async def set_name_cmd(client: Client, message: Message): |
| if not await db.is_user_authorized(message.from_user.id): |
| return await message.reply_text("<b>❌ You are not authorized to use this bot. Please contact the Owner/Admin to authorize you.</b>") |
| is_uploader_mode = await db.get_auto_upload(message.from_user.id) |
| if not is_uploader_mode: |
| return await message.reply_text("Auto Upload mode is OFF. Turn it ON in /settings to use this.") |
| args = message.text.split(" ", 1) |
| if len(args) < 2: |
| await message.reply_text("Usage: `/set_name <anime name>`") |
| return |
| |
| anime_name = args[1].strip() |
| user_id = message.from_user.id |
| chat_id = message.chat.id |
| await client.stop_listening(chat_id=chat_id, user_id=user_id) |
| |
| |
| settings = await db.get_user_settings(user_id) |
| auto_match_enabled = settings.get("auto_channel_match", "on") == "on" |
| |
| state = await db.get_user_state(user_id) |
| job = state.get("current_job", {}) |
| job["anime_name"] = anime_name |
| |
| matched_channel = None |
| if auto_match_enabled: |
| matched_channel = await db.match_channel(user_id, anime_name) |
| |
| if matched_channel: |
| job["target_channel_id"] = matched_channel["channel_id"] |
| job["target_channel_title"] = matched_channel.get("title", f"Channel {matched_channel['channel_id']}") |
| |
| state["current_job"] = job |
| await db.update_user_state(user_id, state) |
| |
| await message.reply_text( |
| f"✅ **Anime Name Set:** `{anime_name}`\n" |
| f"📺 **Auto-matched Target Channel:** {job['target_channel_title']} (`{job['target_channel_id']}`)\n\n" |
| "Use `/start_process` to start processing." |
| ) |
| return |
|
|
| |
| |
| state["current_job"] = job |
| await db.update_user_state(user_id, state) |
| await message.reply_text( |
| f"✅ Anime name set: `{anime_name}`.\n" |
| "Use `/start_process`; if no routing channel matches, you can choose a channel or DM upload." |
| ) |
| return |
|
|
| @Client.on_message(filters.command("set_details") & filters.private) |
| async def set_details_cmd(client: Client, message: Message): |
| if not await db.is_user_authorized(message.from_user.id): |
| return await message.reply_text("<b>❌ You are not authorized to use this bot. Please contact the Owner/Admin to authorize you.</b>") |
| is_uploader_mode = await db.get_auto_upload(message.from_user.id) |
| if not is_uploader_mode: |
| return await message.reply_text("Auto Upload mode is OFF. Turn it ON in /settings to use this.") |
| await client.stop_listening(chat_id=message.chat.id, user_id=message.from_user.id) |
| args = message.text.split(" ", 1) |
| if len(args) < 2: |
| guide = ( |
| "📝 **Set Details Command Guide**\n\n" |
| "This command helps you forcibly define the episode and season in case they are missing from the filename.\n\n" |
| "**Format:**\n" |
| "`/set_details <episode>, <season>, <target_channel_id (optional)>`\n\n" |
| "**Example:**\n" |
| "`/set_details 3, 1` (Set as Episode 3, Season 1)\n" |
| "*Note: Comma `,` is required to separate the values!*" |
| ) |
| await message.reply_text(guide) |
| return |
| |
| parts = [p.strip() for p in args[1].split(",")] |
| if len(parts) < 2: |
| await message.reply_text("❌ Please provide at least episode, and season separated by commas. Example: `/set_details 3, 1`") |
| return |
| |
| state = await db.get_user_state(message.from_user.id) |
| job = state.get("current_job", {}) |
| job["episode"] = parts[0] |
| job["season"] = parts[1] |
| |
| if len(parts) > 2: |
| job["target_channel_id"] = parts[2] |
| |
| state["current_job"] = job |
| await db.update_user_state(message.from_user.id, state) |
| await message.reply_text(f"Details saved: E{job['episode']} S{job['season']}") |
|
|
| @Client.on_message((filters.video | filters.document | filters.audio | filters.text) & filters.private & ~filters.command([c for c in ["start", "settings", "set_name", "clear_queue", "setaudioorder", "resetaudioorder", "set_auto_audio_order", "reset_auto_audio_order", "set_leech_audio_order", "reset_leech_audio_order", "set_default_audio", "resetdefaultaudio", "set_sub_watermark", "set_details", "cancel_task", "start_process", "extract"]])) |
| async def handle_files(client: Client, message: Message): |
| if not await db.is_user_authorized(message.from_user.id): |
| return await message.reply_text("<b>❌ You are not authorized to use this bot. Please contact the Owner/Admin to authorize you.</b>") |
| is_uploader_mode = await db.get_auto_upload(message.from_user.id) |
| if not is_uploader_mode: |
| message.continue_propagation() |
| return |
| |
| is_audio = getattr(message, "audio", None) is not None |
| is_video = getattr(message, "video", None) is not None |
| is_url = False |
| |
| if message.text: |
| text = message.text.strip() |
| if text.startswith("http://") or text.startswith("https://"): |
| is_url = True |
| else: |
| message.continue_propagation() |
| return |
| |
| |
| if message.document: |
| mime = message.document.mime_type or "" |
| name = message.document.file_name or "" |
| ext = os.path.splitext(name)[1].lower() |
| |
| audio_exts = {".mp3", ".wav", ".m4a", ".flac", ".ogg", ".opus", ".aac"} |
| video_exts = {".mp4", ".mkv", ".avi", ".mov", ".webm", ".m4v", ".ts"} |
| |
| if mime.startswith("audio/") or ext in audio_exts: |
| is_audio = True |
| elif mime.startswith("video/") or ext in video_exts: |
| is_video = True |
| else: |
| |
| message.continue_propagation() |
| |
| if not is_audio and not is_video and not is_url: |
| message.continue_propagation() |
| return |
| |
| settings = await db.get_user_settings(message.from_user.id) |
| process_mode = settings.get("process_mode", "merge") |
| |
| if is_video and process_mode == "extract": |
| from plugins.extract import initiate_extraction |
| await initiate_extraction(client, message) |
| return |
| |
| if is_audio: |
| audio_info = {"chat_id": message.chat.id, "msg_id": message.id} |
| await db.set_current_job_audio(message.from_user.id, audio_info) |
| resp = "Audio added to current job." |
| else: |
| video_info = {"chat_id": message.chat.id, "msg_id": message.id} |
| if is_url: |
| video_info["url"] = message.text.strip() |
| |
| await db.push_current_job_video(message.from_user.id, video_info) |
| |
| state = await db.get_user_state(message.from_user.id) |
| job = state.get("current_job", {}) |
| videos = job.get("video_msgs", []) |
| resp = f"{'Link' if is_url else 'Video'} added to current job. Total videos: {len(videos)} \n Use /set_name to set anime name for the file" |
| |
| await message.reply_text(resp) |
|
|
| @Client.on_message(filters.command("clear_queue") & filters.private) |
| async def clear_queue_cmd(client: Client, message: Message): |
| is_uploader_mode = await db.get_auto_upload(message.from_user.id) |
| if not is_uploader_mode: |
| return await message.reply_text("Auto Upload mode is OFF. Turn it ON in /settings to use this.") |
| return |
| state = await db.get_user_state(message.from_user.id) |
| state["current_job"] = {} |
| await db.update_user_state(message.from_user.id, state) |
| await message.reply_text("✅ **Your queue has been cleared!** You can now send new files.") |
|
|
| @Client.on_message(filters.command("cancel_task") & filters.private) |
| async def cancel_task_cmd(client: Client, message: Message): |
| is_uploader_mode = await db.get_auto_upload(message.from_user.id) |
| if not is_uploader_mode: |
| return await message.reply_text("Auto Upload mode is OFF. Turn it ON in /settings to use this.") |
| return |
| from utils.job_queue import USER_CANCELLATIONS |
| USER_CANCELLATIONS.add(message.from_user.id) |
| state = await db.get_user_state(message.from_user.id) |
| state["current_job"] = {} |
| await db.update_user_state(message.from_user.id, state) |
| await message.reply_text("🛑 **Cancellation Requested!**\nAll currently running and queued tasks have been forcibly stopped and cleared.") |
|
|
| @Client.on_message(filters.command("start_process") & filters.private) |
| async def start_process_cmd(client: Client, message: Message): |
| if not await db.is_user_authorized(message.from_user.id): |
| return await message.reply_text("<b>❌ You are not authorized to use this bot. Please contact the Owner/Admin to authorize you.</b>") |
| is_uploader_mode = await db.get_auto_upload(message.from_user.id) |
| if not is_uploader_mode: |
| return await message.reply_text("Auto Upload mode is OFF. Turn it ON in /settings to use this.") |
| |
| from utils.job_queue import USER_CANCELLATIONS |
| USER_CANCELLATIONS.discard(message.from_user.id) |
| |
| await client.stop_listening(chat_id=message.chat.id, user_id=message.from_user.id) |
| settings = await db.get_user_settings(message.from_user.id) |
| process_mode = settings.get("process_mode", "merge") |
| state = await db.get_user_state(message.from_user.id) |
| job = state.get("current_job", {}) |
| if "anime_name" not in job: |
| job["anime_name"] = None |
| if "video_msgs" not in job or not job["video_msgs"]: |
| await message.reply_text("No videos added.") |
| return |
| if process_mode == "merge": |
| all_urls = all("url" in vm for vm in job.get("video_msgs", [])) |
| if not all_urls and "audio_msg" not in job: |
| await message.reply_text("No audio added.") |
| return |
|
|
| |
| |
| target = job.get("target_channel_id") or job.get("target_chat_id") |
| if not target and settings.get("auto_channel_match", "on") == "on": |
| match_text = str(job.get("anime_name") or "") |
| if not match_text: |
| try: |
| vm = job["video_msgs"][0] |
| if "url" in vm: |
| from plugins.file_rename import resolve_remote_filename |
| match_text = await resolve_remote_filename(vm["url"], fallback="") |
| if match_text: |
| vm["file_name"] = match_text |
| else: |
| media_msg = await client.get_messages(vm["chat_id"], vm["msg_id"]) |
| media = media_msg.video or media_msg.document |
| match_text = getattr(media, "file_name", "") or "" |
| except Exception: |
| pass |
| matched = await db.match_channel(message.from_user.id, match_text) if match_text else None |
| if matched: |
| job["target_channel_id"] = matched["channel_id"] |
| job["target_channel_title"] = matched.get("title", "Routing channel") |
| target = matched["channel_id"] |
| if not target: |
| await db.update_user_state(message.from_user.id, state) |
| return await message.reply_text( |
| "📤 Routing channel match kidaikkala. Upload destination choose pannunga:", |
| reply_markup=InlineKeyboardMarkup([ |
| [InlineKeyboardButton("📺 Set Channel", callback_data="process_destination_channel")], |
| [InlineKeyboardButton("💬 Upload to Bot DM", callback_data="process_destination_dm")], |
| ]) |
| ) |
| |
| |
| log_msg_id = None |
| if Config.LOG_CHANNEL: |
| try: |
| log_msg = ( |
| f"🚀 **Bot Job Triggered**\n\n" |
| f"• **Anime:** `{job.get('anime_name')}`\n" |
| f"• **Videos Count:** `{len(job.get('video_msgs', []))}`" |
| ) |
| sent_log_msg = await client.send_message(Config.LOG_CHANNEL, log_msg) |
| log_msg_id = sent_log_msg.id |
| except Exception as e: |
| print(f"Error logging job start: {e}") |
| |
| status_msg = await message.reply_text("Queuing job...") |
| |
| full_job = { |
| "user_id": message.from_user.id, |
| "chat_id": message.chat.id, |
| "status_msg_id": status_msg.id, |
| "log_msg_id": log_msg_id, |
| **job |
| } |
| |
| await process_queue.put(full_job) |
| |
| |
| state["current_job"] = {} |
| await db.update_user_state(message.from_user.id, state) |
| |
| await status_msg.edit_text("Job queued successfully! Position in queue: " + str(process_queue.qsize())) |
|
|
|
|
| async def _queue_destination_choice(client, query, destination=None): |
| user_id = query.from_user.id |
| if not await db.is_user_authorized(user_id) or not await db.get_auto_upload(user_id): |
| return await query.answer("Auto Upload is off or access is not authorized.", show_alert=True) |
| state = await db.get_user_state(user_id) |
| job = state.get("current_job", {}) |
| if not job.get("video_msgs"): |
| return await query.answer("No pending video task found.", show_alert=True) |
| if destination: |
| job["target_channel_id"] = destination.id |
| job["target_channel_title"] = getattr(destination, "title", "Routing channel") |
| else: |
| job["target_channel_id"] = query.message.chat.id |
| job["target_channel_title"] = "Your Bot DM" |
| status = await query.message.reply_text("Queuing job...") |
| full_job = {"user_id": user_id, "chat_id": query.message.chat.id, |
| "status_msg_id": status.id, **job} |
| await process_queue.put(full_job) |
| state["current_job"] = {} |
| await db.update_user_state(user_id, state) |
| await query.message.edit_text("✅ Job queued! Position: " + str(process_queue.qsize())) |
| await query.answer() |
|
|
|
|
| @Client.on_callback_query(filters.regex(r"^process_destination_dm$")) |
| async def process_destination_dm_cb(client: Client, query: CallbackQuery): |
| await _queue_destination_choice(client, query) |
|
|
|
|
| @Client.on_callback_query(filters.regex(r"^process_destination_channel$")) |
| async def process_destination_channel_cb(client: Client, query: CallbackQuery): |
| user_id, chat_id = query.from_user.id, query.message.chat.id |
| if not await db.is_user_authorized(user_id) or not await db.get_auto_upload(user_id): |
| return await query.answer("Auto Upload is off or access is not authorized.", show_alert=True) |
| await query.message.edit_text( |
| "Send channel ID, public @username/link, or forward a post from the channel.", |
| reply_markup=InlineKeyboardMarkup([[InlineKeyboardButton("Cancel", callback_data="process_destination_cancel")]]) |
| ) |
| await query.answer() |
| try: |
| response = await client.listen(chat_id=chat_id, user_id=user_id, timeout=300) |
| from utils.channel_resolver import resolve_routing_channel |
| forwarded = getattr(response, "forward_from_chat", None) |
| if not forwarded and getattr(response, "forward_origin", None): |
| forwarded = getattr(response.forward_origin, "chat", None) |
| channel = await resolve_routing_channel(client, response.text if response.text else None, forwarded) |
| member = await client.get_chat_member(channel.id, "me") |
| member_status = getattr(member, "status", "") |
| member_status = str(getattr(member_status, "value", member_status)).lower().rsplit(".", 1)[-1] |
| if member_status not in {"administrator", "owner", "creator"}: |
| raise ValueError("Bot must be an admin in that channel to upload there.") |
| await response.delete() |
| await _queue_destination_choice(client, query, channel) |
| except Exception as exc: |
| await query.message.reply_text(f"❌ Channel setup failed: {exc}\nRun `/start_process` and try again.") |
|
|
|
|
| @Client.on_callback_query(filters.regex(r"^process_destination_cancel$")) |
| async def process_destination_cancel_cb(client: Client, query: CallbackQuery): |
| await query.answer("Task is still saved. You can run /start_process again.") |
|
|
