Blame for plugins/process.py

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)
    
    # Retrieve user settings
    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

    # Defer destination choice until /start_process so the task can still be
    # routed manually or sent back to the user's DM without blocking /set_name.
    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 it is a document, determine if it is audio or video or other
    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:
            # Ignore other files (like JSON, text, etc.) from process queue
            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

    # Use a saved per-task destination or an existing routing match when one
    # exists. Otherwise let the user pick a channel or their private chat.
    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 triggering start process
    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)
    
    # clear 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.")