Blame for utils/job_queue.py

import asyncio
from config import Config
from plugins.file_rename import process_thumbnail
from utils.ffmpeg import get_video_resolution, strip_and_mux_audio, get_media_info
from helper.database import codeflixbots as db
import os
import shutil
from utils.sub_watermark import extract_and_watermark_subs
import time
import re
import html
from urllib.parse import unquote
from pyrogram import enums
from pyrogram.types import InlineKeyboardMarkup, InlineKeyboardButton

SEASON_EPISODE_PATTERNS = [
    (re.compile(r'S(\d+)(?:E|EP)(\d+)'), ('season', 'episode')),
    (re.compile(r'S(\d+)[\s-]*(?:E|EP)(\d+)'), ('season', 'episode')),
    (re.compile(r'Season\s*(\d+)\s*Episode\s*(\d+)', re.IGNORECASE), ('season', 'episode')),
    (re.compile(r'\[S(\d+)\]\[E(\d+)\]'), ('season', 'episode')),
    (re.compile(r'S(\d+)[^\d]*(\d+)'), ('season', 'episode')),
    (re.compile(r'(?:E|EP|Episode)\s*(\d+)', re.IGNORECASE), (None, 'episode')),
    (re.compile(r'\b(\d+)\b'), (None, 'episode'))
]

def extract_season_episode(filename):
    for pattern, (season_group, episode_group) in SEASON_EPISODE_PATTERNS:
        match = pattern.search(filename)
        if match:
            season = match.group(1) if season_group else None
            episode = match.group(2 if season_group else 1)
            return season, episode
    return None, None

def extract_quality_from_filename(filename):
    match = re.search(r'(360p|480p|720p|1080p|1440p|2160p|4k)', filename, re.IGNORECASE)
    if match:
        return match.group(1).lower()
    return "-"

def humanbytes(size):
    if not size:
        return "0 B"
    power = 2**10
    n = 0
    Dic_powerN = {0: 'B', 1: 'KB', 2: 'MB', 3: 'GB', 4: 'TB'}
    while size > power and n < 4:
        size /= power
        n += 1
    return f"{round(size, 2)} {Dic_powerN[n]}"

process_queue = asyncio.Queue()
USER_CANCELLATIONS = set()

async def worker(client):
    while True:
        job = await process_queue.get()
        user_id = job.get('user_id')
        if user_id in USER_CANCELLATIONS:
            print(f"Skipping job for cancelled user {user_id}")
            try:
                await client.send_message(job['chat_id'], f"🛑 **Queued task skipped (Cancelled).**")
                await client.delete_messages(job['chat_id'], job['status_msg_id'])
            except Exception:
                pass
            process_queue.task_done()
            continue
            
        try:
            await process_job(client, job)
        except Exception as e:
            print(f"Error processing job: {e}")
        finally:
            process_queue.task_done()

def get_progress_text(ep, state, anime, s_and_e, quality, filename):
    return (
        f"**[ Episode : {ep} ]**\n"
        f"**state** - {state}\n"
        f"**anime name** - {anime}\n"
        f"**season and episode** - {s_and_e}\n"
        f"**quality** - {quality}\n"
        f"**Filename** - `{filename}`"
    )

def sort_key(res_str):
    if "480" in res_str: return 1
    if "720" in res_str: return 2
    if "1080" in res_str: return 3
    if "4k" in res_str.lower(): return 4
    return 10

def safe_format(template, **kwargs):
    temp_tpl = template
    # Force 02d formatting specifiers for season and episode placeholders
    temp_tpl = re.sub(r'\{season(?::[^}]+)?\}', '{season:02d}', temp_tpl)
    temp_tpl = re.sub(r'\{episode(?::[^}]+)?\}', '{episode:02d}', temp_tpl)
    
    formatted_kwargs = dict(kwargs)
    try:
        if 'season' in formatted_kwargs:
            formatted_kwargs['season'] = int(str(formatted_kwargs['season']).strip())
    except Exception:
        pass
    try:
        if 'episode' in formatted_kwargs:
            formatted_kwargs['episode'] = int(str(formatted_kwargs['episode']).strip())
    except Exception:
        pass
        
    try:
        return temp_tpl.format(**formatted_kwargs)
    except Exception:
        # Fallback to simple replace
        res = template
        for k, v in kwargs.items():
            val_str = str(v)
            if k in ['season', 'episode']:
                try:
                    if int(v) < 10:
                        val_str = f"{int(v):02d}"
                except Exception:
                    pass
            res = re.sub(r'\{' + k + r'(?::[^}]+)?\}', val_str, res)
        return res

async def apply_metadata(input_path, output_path, user_id, client=None):
    settings = await db.get_user_settings(user_id)
    
    metadata_on = settings.get("metadata", "Off") != "Off"
    metadata_txt = settings.get("metadata_txt", "") if metadata_on else ""
    
    # FETCH PROPERLY:
    audio_order = await db.get_audio_order(user_id)
    default_audio = await db.get_audio_default(user_id)
    keep_audio = await db.get_keep_audio(user_id)
    sub_watermark = settings.get("sub_watermark")
    
    # Fast copy if no processing is needed
    if not (metadata_txt or audio_order or keep_audio or default_audio is not None or sub_watermark):
        shutil.copy2(input_path, output_path)
        return True
        
    cmd_args = ["ffmpeg", "-y", "-i", input_path]
    
    # 1. Handle Subtitle Watermark Extractor
    watermarked_subs = []
    if sub_watermark:
        temp_dir = os.path.dirname(input_path)
        watermarked_subs = await extract_and_watermark_subs(input_path, sub_watermark, temp_dir)
        
        # DEBUG FALLBACK: Tell the user if it mysteriously failed returning None
        if not watermarked_subs or not any(watermarked_subs):
            if client:
                try:
                    await client.send_message(user_id, f"⚠️ Subtitle Debug: Extractor returned `{watermarked_subs}`. Subtitle mapping aborted.")
                except Exception:
                    pass
                    
        for sub_file in watermarked_subs:
            if sub_file:
                cmd_args.extend(["-i", sub_file])
                
    # 2. Assign Metadata Text Configs
    if metadata_txt:
        cmd_args.extend(["-metadata", f"title={metadata_txt}"])
        cmd_args.extend(["-metadata", f"artist={metadata_txt}"])
        cmd_args.extend(["-metadata", f"author={metadata_txt}"])
        cmd_args.extend(["-metadata:s:v", f"title={metadata_txt}"])
        cmd_args.extend(["-metadata:s:a", f"title={metadata_txt}"])
        cmd_args.extend(["-metadata:s:s", f"title={metadata_txt}"])
        
    # 3. Handle Map streams (Audio Reorder)
    cmd_args.extend(["-map", "0:v"]) # Primary video
    
    if keep_audio:
        from utils.audio_tracks import resolve_keep_audio
        kept_tracks = await resolve_keep_audio(input_path, keep_audio)
        for track in kept_tracks:
            cmd_args.extend(["-map", f"0:a:{track}?"])
    elif audio_order:
        try:
            from utils.audio_tracks import probe_audio_streams
            order = [part.strip() for part in str(audio_order).split(",") if part.strip()]
            streams = await probe_audio_streams(input_path)
            seen = set()
            for at in order:
                if at.isdigit() and int(at) < len(streams) and at not in seen:
                    cmd_args.extend(["-map", f"0:a:{at}?"])
                    seen.add(at)
            for idx in range(len(streams)):
                if str(idx) not in seen:
                    cmd_args.extend(["-map", f"0:a:{idx}?"])
        except Exception:
            cmd_args.extend(["-map", "0:a?"])
    else:
        cmd_args.extend(["-map", "0:a?"]) # Map all audios
        
    # Inject Custom Watermarked Subs or Standard Subs
    if watermarked_subs and any(watermarked_subs):
        input_idx = 1
        for sf in watermarked_subs:
            if sf:
                cmd_args.extend(["-map", f"{input_idx}:0?"])
                input_idx += 1
    else:
        cmd_args.extend(["-map", "0:s?"]) # standard map
        
    cmd_args.extend(["-map", "0:t?"]) # Fonts & Attachments
    
    # 4. Enforce Selected Default Audio Display disposition
    if default_audio is not None:
        cmd_args.extend(["-disposition:a", "0"]) # Turn off all audios default
        try:
            cmd_args.extend([f"-disposition:a:{str(default_audio).strip()}", "default"])
        except Exception:
            pass
            
    cmd_args.extend(["-c", "copy", output_path])
    
    process = await asyncio.create_subprocess_exec(
        *cmd_args,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )
    
    while True:
        try:
            stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=2.0)
            break
        except asyncio.TimeoutError:
            if user_id in USER_CANCELLATIONS:
                try:
                    process.kill()
                except Exception:
                    pass
                raise RuntimeError("TaskCancelledByUser")
    
    # Subtitle Cleanup
    for sf in watermarked_subs:
        if sf and os.path.exists(sf):
            try:
                os.remove(sf)
            except Exception:
                pass
                
    return process.returncode == 0

async def process_job(client, job):
    user_id = job['user_id']
    status_msg = job['status_msg_id']
    chat_id = job['chat_id']
    settings = await db.get_user_settings(user_id)
    
    # 1. Fetch first video details context for filename extraction
    original_name = "Unknown.mkv"
    try:
        first_vm = job['video_msgs'][0]
        if "url" in first_vm:
            url = first_vm["url"]
            try:
                from plugins.file_rename import resolve_remote_filename
                resolved = await resolve_remote_filename(url, fallback="Unknown.mkv")
                if resolved:
                    original_name = resolved
                    print(f"Resolved remote filename: {original_name}")
            except Exception as e:
                print(f"Could not resolve remote filename: {e}")
        else:
            first_v_msg = await client.get_messages(first_vm['chat_id'], first_vm['msg_id'])
            original_name = getattr(first_v_msg.video or first_v_msg.document, "file_name", "Unknown.mkv")
    except Exception as e:
        print(f"Error fetching first video message profile: {e}")
        
    # 2. Extract season and episode dynamically
    season_ext, episode_ext = extract_season_episode(original_name)
    if season_ext:
        job['season'] = season_ext
    if episode_ext:
        job['episode'] = episode_ext
        
    for key in ['season', 'episode']:
        val = job.get(key, '01')
        try:
            if int(val) < 10:
                job[key] = f"{int(val):02d}"
            else:
                job[key] = str(int(val))
        except (ValueError, TypeError):
            pass
        
    # 3. Channel resolution based on auto routing settings
    auto_routing = settings.get('auto_channel_match', 'on') == 'on'
    target_chat = job.get('target_channel_id')
    channel = None
    
    if auto_routing:
        # If not set in the job already, try auto matching
        channel = await db.match_channel(user_id, original_name)
        if channel and not target_chat:
            target_chat = channel['channel_id']
            
    if not target_chat:
        target_chat = chat_id
        
    if isinstance(target_chat, str) and (target_chat.startswith("-") or target_chat.isdigit()):
        try:
            target_chat = int(target_chat)
        except ValueError:
            pass
            
    # 4. Pre-post "Episode : {episode_number}" to target chat
    try:
        await client.send_message(target_chat, f"**Episode : {job.get('episode', '01')}**")
    except Exception as e:
        print(f"Error pre-posting episode header to chat: {e}")
        
    # Create the progress status message DIRECTLY inside the target channel!
    status_msg_node = None
    try:
        status_msg_node = await client.send_message(target_chat, text="⏳ Starting process...")
    except Exception as e:
        print(f"Error sending progress message to channel: {e}")
        
    status_msg_node_id = status_msg_node.id if status_msg_node else None
    
    ep = str(job.get('episode', '?')).zfill(2)
    s = str(job.get('season', '?')).zfill(2)
    
    # Dynamic anime name guessing
    anime = job.get('anime_name')
    if not anime:
        if channel:
            anime = channel.get('title', 'Unknown')
        else:
            try:
                parts = original_name.split(' - ')
                if len(parts) > 1:
                    s_first, ep_first = extract_season_episode(parts[0])
                    if s_first or ep_first:
                        anime = parts[1].split('[')[0].strip()
                    else:
                        anime = parts[0].split('[')[0].strip()
                else:
                    anime = original_name.split('.')[0].strip()
            except Exception:
                anime = "Unknown"
    s_e = f"S{s}E{ep}"
    
    async def update_state(state_header, progress_bar="", quality="-", filename="-"):
        disp_filename = filename
            
        esc_header = html.escape(str(state_header))
        esc_anime = html.escape(str(anime))
        esc_quality = html.escape(str(quality))
        esc_ep = html.escape(str(ep))
        esc_filename = html.escape(str(disp_filename))
        
        text = (
            f"<blockquote>{esc_header}</blockquote>\n\n"
            f"<blockquote>🎬 <b>𝗔𝗻𝗶𝗺𝗲 :</b> {esc_anime}\n"
            f"📦 <b>𝗤𝘂𝗮𝗹𝗶𝘁𝘆 :</b> {esc_quality}\n"
            f"🔢 <b>𝗘𝗽𝗶𝘀𝗼𝗱𝗲 :</b> {esc_ep}\n"
            f"📁 <b>𝗙𝗶𝗹𝗲 :</b> <code>{esc_filename}</code>"
        )
        if progress_bar:
            esc_progress = html.escape(str(progress_bar))
            text += f"\n\n<b>{esc_progress}</b>"
        text += "</blockquote>"
            
        if status_msg_node_id:
            try:
                await client.edit_message_text(target_chat, status_msg_node_id, text, parse_mode=enums.ParseMode.HTML)
                return
            except Exception:
                pass
        # Fallback to PM if channel edit fails
        try:
            await client.edit_message_text(chat_id, status_msg, text, parse_mode=enums.ParseMode.HTML)
        except Exception:
            pass

    # Process start
    loop = asyncio.get_running_loop()
    dl_start = [0]
    last_dl_update = [0]
    
    def dl_progress_cb(current, total, file_label="", quality_label="-"):
        if job.get('user_id') in USER_CANCELLATIONS:
            raise RuntimeError("TaskCancelledByUser")
        now = time.time()
        if dl_start[0] == 0:
            dl_start[0] = now
        if now - last_dl_update[0] < 8 and current < total:
            return
        last_dl_update[0] = now
        
        percent = (current / total) * 100 if total else 0
        filled = int(percent / 10)
        bar = "■" * filled + "□" * (10 - filled)
        
        elapsed = now - dl_start[0]
        speed = current / elapsed if elapsed > 0 else 0
        eta = ""
        if speed > 0:
            eta_seconds = (total - current) / speed
            eta = f" | ETA: {int(eta_seconds)}s"
            
        speed_text = f"{humanbytes(speed)}/s" if speed else "—/s"
        progress_bar = f"[{bar}] {percent:.1f}% ({humanbytes(current)} of {humanbytes(total)} @ {speed_text}{eta})"
        
        asyncio.run_coroutine_threadsafe(
            update_state("📥 𝗗𝗼𝘄𝗻𝗹𝗼𝗮𝗱𝗶𝗻𝗴 𝗔𝗻𝗶𝗺𝗲 𝗙𝗶𝗹𝗲...", progress_bar, quality_label, file_label),
            loop
        )

    await update_state("⏳ Starting resource downloads...")
    
    # 2. Fetch video messages sequentially by message ID (preserving chronological forwarding order)
    resolved_videos = []
    in_order_msgs = sorted(job['video_msgs'], key=lambda x: x['msg_id'])
    for vm in in_order_msgs:
        try:
            v_msg = await client.get_messages(vm['chat_id'], vm['msg_id'])
            v_orig_name = getattr(v_msg.video or v_msg.document, "file_name", "")
            q = extract_quality_from_filename(v_orig_name)
        except Exception:
            v_msg = None
            q = "-"
        resolved_videos.append((vm, q, v_msg))
        
    # Preserve the exact sequence the user sent the videos in
    # (Removed quality sorting as requested by user)
    
    # Update job['video_msgs'] and cache messages list
    job['video_msgs'] = [rv[0] for rv in resolved_videos]
    cached_video_msgs = [rv[2] for rv in resolved_videos]
    
    original_qualities = {}
    predicted_filenames = {}
    
    prefix = settings.get('prefix', '')
    suffix = settings.get('suffix', '')
    fmt = settings.get("rename_format", "{anime} - S{season:02d}E{episode:02d} [{language}] {quality} @suffix.mkv")
    
    ext = ".mkv"
    base_fmt = fmt
    if fmt.lower().endswith(".mkv"):
        base_fmt = fmt[:-4]
        ext = ".mkv"
    elif fmt.lower().endswith(".mp4"):
        base_fmt = fmt[:-4]
        ext = ".mp4"
        
    for i, rv in enumerate(resolved_videos):
        q = rv[1]
        original_qualities[i] = q
        try:
            filename_base = safe_format(
                base_fmt,
                anime=anime,
                season=job.get('season', '01'),
                episode=job.get('episode', '01'),
                language=job.get('language', 'Unknown'),
                quality=q
            )
            if prefix: filename_base = f"{prefix} {filename_base}"
            if suffix: filename_base = f"{filename_base} {suffix}"
            predicted_filenames[i] = f"{filename_base}{ext}"
        except Exception:
            predicted_filenames[i] = "video.mkv"
            
    job_id = f"job_{user_id}_{status_msg}"
    work_dir = os.path.join("temp", job_id)
    os.makedirs(work_dir, exist_ok=True)
    
    custom_thumb = settings.get("thumbnail")
    thumb_path = None
    if custom_thumb:
        try:
            download_path = await client.download_media(message=custom_thumb, file_name=os.path.join(work_dir, "thumb.jpg"))
            if download_path and os.path.exists(download_path):
                if os.path.getsize(download_path) > 0:
                    thumb_size = await db.get_thumb_size(user_id)
                    thumb_path = await process_thumbnail(download_path, thumb_size)
                else:
                    print("Thumbnail download resulted in a 0-byte file.")
                    os.remove(download_path)
        except Exception as e:
            print(f"Error downloading user thumbnail: {e}")
            
    if custom_thumb and not thumb_path:
        # It means the saved thumbnail failed to download correctly (e.g. FileReferenceExpired)
        try: 
            await client.send_message(chat_id, "⚠️ **Notice:** Your custom thumbnail is expired or could not be downloaded. It was ignored. Please set a new one in `/settings`.")
        except Exception:
            pass
            
    try:
        process_mode = settings.get("process_mode", "merge")
        audio_path = None

        # If ALL video sources are direct URLs, the MKV already has audio embedded.
        # Skip the audio merge step automatically — treat as rename_only.
        all_urls = all("url" in vm for vm in job.get("video_msgs", []))
        if all_urls and process_mode == "merge":
            process_mode = "rename_only"

        if process_mode != "rename_only":
            try:
                audio_msg = await client.get_messages(job['audio_msg']['chat_id'], job['audio_msg']['msg_id'])
                if audio_msg.empty:
                    raise ValueError("Audio message deleted")
                dl_start[0] = time.time()
                last_dl_update[0] = 0
                audio_path = await audio_msg.download(
                    file_name=os.path.join(work_dir, "audio.m4a"),
                    progress=lambda c, t: dl_progress_cb(c, t, "audio_stream", "-")
                )
            except Exception as e:
                raise RuntimeError("Audio message not found or deleted by user. Please /clear_queue and try again.")
        
        upload_type = settings.get('upload_type', 'document')
        button_mode = settings.get('button_mode', 'off') == 'on'
        prefix = settings.get('prefix', '')
        suffix = settings.get('suffix', '')
        
        filestore_username = settings.get("filestore_username", "").strip()
        dump_channel_id = settings.get("dump_channel_id")
        
        # Verify button parameters before continuing, fallback to regular if not configured
        if button_mode and (not filestore_username or not dump_channel_id):
            button_mode = False
            try:
                await client.send_message(chat_id, "⚠️ **Warning:** Button Mode is enabled, but File Store Bot credentials are not configured in settings. Falling back to regular upload.")
            except Exception:
                pass
                
        generated_links = []
            
        uploaded_success_count = 0 
            
        for i, rv in enumerate(resolved_videos):
            vm = rv[0]
            q = rv[1]
            v_msg = rv[2]
            
            try:
                if "url" in vm:
                    v_orig_name = f"video_{i}.mkv"
                else:
                    if not v_msg:
                        v_msg = await client.get_messages(vm['chat_id'], vm['msg_id'])
                    v_orig_name = getattr(v_msg.video or v_msg.document or v_msg, "file_name", f"video_{i}.mkv")
                
                # Update progress info for downloading this specific quality path
                dl_start[0] = time.time()
                last_dl_update[0] = 0
                await update_state("📥 𝗗𝗼𝘄𝗻𝗹𝗼𝗮𝗱𝗶𝗻𝗴 𝗩𝗶𝗱𝗲𝗼...", "", q, predicted_filenames.get(i, v_orig_name))
                
                if "url" in vm:
                    from utils.downloader import download_url_with_progress
                    async def url_prog(c, t):
                        dl_progress_cb(c, t, predicted_filenames.get(i, v_orig_name), q)
                    v_path = await download_url_with_progress(
                        vm["url"],
                        os.path.join(work_dir, f"video_{i}.mkv"),
                        progress_callback=url_prog
                    )
                else:
                    v_path = await v_msg.download(
                        file_name=os.path.join(work_dir, f"video_{i}.mkv"),
                        progress=lambda c, t: dl_progress_cb(c, t, predicted_filenames.get(i, v_orig_name), q)
                    )
                    
                if os.path.exists(v_path) and os.path.getsize(v_path) == 0:
                    raise RuntimeError("Downloaded file size is 0 bytes. The URL or file might be restricted/blocked.")
                
                # Probe video resolution if not found in filename
                if q != "-":
                    res = q
                else:
                    await update_state("🔍 𝗣𝗿𝗼𝗯𝗶𝗻𝗴 𝗥𝗲𝘀𝗼𝗹𝘂𝘁𝗶𝗼𝗻...", "", q, predicted_filenames.get(i, v_orig_name))
                    res = await get_video_resolution(v_path)
                    if res == "Unknown":
                        res = q
                    
                # Format final filename
                fmt = settings.get("rename_format", "{anime} - S{season:02d}E{episode:02d} [{language}] {quality} @suffix.mkv")
                ext = ".mkv"
                base_fmt = fmt
                if fmt.lower().endswith(".mkv"):
                    base_fmt = fmt[:-4]
                    ext = ".mkv"
                elif fmt.lower().endswith(".mp4"):
                    base_fmt = fmt[:-4]
                    ext = ".mp4"
                    
                filename_base = safe_format(
                    base_fmt,
                    anime=anime,
                    season=job.get('season', '01'),
                    episode=job.get('episode', '01'),
                    language=job.get('language', 'Unknown'),
                    quality=res
                )
                if prefix: filename_base = f"{prefix} {filename_base}"
                if suffix: filename_base = f"{filename_base} {suffix}"
                filename = f"{filename_base}{ext}"
                filename = filename.replace("/", "-").replace("\\", "-")
                if process_mode == "rename_only":
                    await update_state("⚙️ 𝗘𝗻𝗰𝗼𝗱𝗶𝗻𝗴 𝗩𝗶𝗱𝗲𝗼...", "", res, filename)
                    out_path = os.path.join(work_dir, f"out_{res}.mkv")
                    shutil.copy2(v_path, out_path)
                    success = True
                else:
                    await update_state("⚙️ 𝗘𝗻𝗰𝗼𝗱𝗶𝗻𝗴 𝗩𝗶𝗱𝗲𝗼...", "", res, filename)
                    out_path = os.path.join(work_dir, f"out_{res}.mkv")
                    success = await strip_and_mux_audio(v_path, audio_path, out_path, user_id)
                
                if success:
                    final_path = os.path.join(work_dir, filename)
                    meta_ok = await apply_metadata(out_path, final_path, user_id, client)
                    if not meta_ok:
                        os.rename(out_path, final_path)
                    
                    upload_label = "📤 𝗨𝗽𝗹𝗼𝗮𝗱𝗶𝗻𝗴 𝗧𝗼 𝗗𝘂𝗺𝗽..." if button_mode else "📤 𝗨𝗽𝗹𝗼𝗮𝗱𝗶𝗻𝗴 𝗧𝗼 𝗖𝗵𝗮𝗻𝗻𝗲𝗹..."
                    await update_state(upload_label, "", res, filename)
                    
                    # Calculate new metadata formats
                    languages_eng = "Unknown"
                    languages_reg = "Unknown"
                    duration_str = "00:00"
                    duration_sec = 0
                    file_size_str = "0 B"
                    
                    if not os.path.exists(final_path) or os.path.getsize(final_path) == 0:
                        raise RuntimeError(f"Encoding failed! Created output file is empty/missing. Could be an IP block from server CDN.")
                                                
                    try:
                        file_size_str = humanbytes(os.path.getsize(final_path))
                        m_info = await get_media_info(final_path)
                        
                        f_info = m_info.get("format", {})
                        if "duration" in f_info:
                            duration_sec = int(float(f_info["duration"]))
                            hours = duration_sec // 3600
                            minutes = (duration_sec % 3600) // 60
                            seconds = duration_sec % 60
                            if hours > 0:
                                duration_str = f"{hours}:{minutes:02d}:{seconds:02d}"
                            else:
                                duration_str = f"{minutes}:{seconds:02d}"
                                
                        streams = m_info.get("streams", [])
                        found_langs = []
                        for s in streams:
                            if s.get("codec_type") == "audio":
                                lang = s.get("tags", {}).get("language", "und").lower()
                                if lang != "und":
                                    found_langs.append(lang)
                                    
                        if found_langs:
                            seen = set()
                            unique_langs = [x for x in found_langs if not (x in seen or seen.add(x))]
                            
                            lang_map_eng = {"tam": "Tamil", "hin": "Hindi", "tel": "Telugu", "eng": "English", "jpn": "Japanese", "mal": "Malayalam", "kan": "Kannada", "kor": "Korean"}
                            lang_map_reg = {"tam": "தமிழ்", "hin": "हिन्दी", "tel": "తెలుగు", "eng": "English", "jpn": "日本語", "mal": "മലയാളం", "kan": "ಕನ್ನಡ", "kor": "한국어"}
                            
                            languages_eng = ", ".join([lang_map_eng.get(l, l.title()) for l in unique_langs])
                            languages_reg = ", ".join([lang_map_reg.get(l, lang_map_eng.get(l, l.title())) for l in unique_langs])
                    except Exception as e:
                        print(f"Error parsing advanced media info: {e}")
                    
                    cap_fmt = settings.get("caption_format", "<b>{filename}</b>")
                    if not cap_fmt or cap_fmt == "{filename}":
                        cap_fmt = "<b>{filename}</b>"
                    caption = safe_format(
                        cap_fmt,
                        filename=filename,
                        anime=anime,
                        season=job.get('season', '01'),
                        episode=job.get('episode', '01'),
                        language=job.get('language', 'Unknown'),
                        quality=res,
                        size=file_size_str,
                        duration=duration_str,
                        languages=languages_eng,
                        languagesr=languages_reg
                    )
                    
                    # Upload with progress bar
                    uploaded_msg = None
                    try:
                        upload_start = time.time()
                        last_update_time = [0]
                        
                        def progress_cb(current, total):
                            if job.get('user_id') in USER_CANCELLATIONS:
                                raise RuntimeError("TaskCancelledByUser")
                            if total < 2 * 1024 * 1024:
                                return
                            now = time.time()
                            if now - last_update_time[0] < 8 and current < total:
                                return
                            last_update_time[0] = now
                            
                            percent = (current / total) * 100 if total else 0
                            filled = int(percent / 10)
                            bar = "■" * filled + "□" * (10 - filled)
                            
                            elapsed = now - upload_start
                            speed = current / elapsed if elapsed > 0 else 0
                            eta = ""
                            if speed > 0:
                                eta_seconds = (total - current) / speed
                                eta = f" | ETA: {int(eta_seconds)}s"
                                
                            speed_text = f"{humanbytes(speed)}/s" if speed else "—/s"
                            progress_bar_text = f"[{bar}] {percent:.1f}% ({humanbytes(current)} of {humanbytes(total)} @ {speed_text}{eta})"
                            
                            asyncio.run_coroutine_threadsafe(
                                update_state(upload_label, progress_bar_text, res, filename),
                                loop
                            )
                        
                        upload_chat_id = int(dump_channel_id) if button_mode else target_chat
                        actual_type = upload_type
                        
                        if actual_type == "video":
                            uploaded_msg = await client.send_video(
                                chat_id=upload_chat_id,
                                video=final_path,
                                caption=caption,
                                parse_mode=enums.ParseMode.HTML,
                                thumb=thumb_path,
                                duration=duration_sec,
                                progress=progress_cb
                            )
                        else:
                            uploaded_msg = await client.send_document(
                                chat_id=upload_chat_id,
                                document=final_path,
                                caption=caption,
                                parse_mode=enums.ParseMode.HTML,
                                thumb=thumb_path,
                                progress=progress_cb
                            )
                    except Exception as e:
                        print(f"Error during progress upload: {e}. Retrying standard upload...")
                        upload_chat_id = int(dump_channel_id) if button_mode else target_chat
                        actual_type = upload_type
                        if actual_type == "video":
                            uploaded_msg = await client.send_video(upload_chat_id, video=final_path, caption=caption, parse_mode=enums.ParseMode.HTML, thumb=thumb_path, duration=duration_sec)
                        else:
                            uploaded_msg = await client.send_document(upload_chat_id, document=final_path, caption=caption, parse_mode=enums.ParseMode.HTML, thumb=thumb_path)
                    
                    if button_mode and uploaded_msg:
                        import base64
                        msg_id_val = uploaded_msg.id
                        payload_str = f"get-{msg_id_val * abs(int(dump_channel_id))}"
                        payload_bytes = payload_str.encode("ascii")
                        base64_bytes = base64.urlsafe_b64encode(payload_bytes)
                        base64_string = base64_bytes.decode("ascii").strip("=")
                        
                        bot_link = f"https://t.me/{filestore_username}?start={base64_string}"
                        generated_links.append((res, bot_link))
                        
                    if uploaded_msg and Config.GLOBAL_DUMP_CHANNEL:
                        try:
                            upload_chat_id = int(dump_channel_id) if button_mode else target_chat
                            if Config.GLOBAL_DUMP_CHANNEL != upload_chat_id:
                                uinfo = await client.get_users(user_id)
                                global_cap = caption + f"\n\n👤 **Processed By:** {uinfo.mention} (`{uinfo.id}`)"
                                await client.copy_message(
                                    chat_id=Config.GLOBAL_DUMP_CHANNEL,
                                    from_chat_id=upload_chat_id,
                                    message_id=uploaded_msg.id,
                                    caption=global_cap,
                                    parse_mode=enums.ParseMode.HTML
                                )
                        except Exception as e:
                            print(f"Global dump failed: {e}")
                        
                # Clean up this resolution files immediately to conserve disk space
                try:
                    if os.path.exists(v_path): os.remove(v_path)
                    if os.path.exists(out_path): os.remove(out_path)
                    if success and os.path.exists(final_path): os.remove(final_path)
                except Exception:
                    pass
                
                uploaded_success_count += 1

                    
            except Exception as e:
                if "TaskCancelledByUser" in str(e):
                    print(f"Task for user {user_id} cancelled by request. Breaking loop.")
                    try:
                        await client.send_message(chat_id, "🛑 **Task has been forcefully cancelled as requested.**")
                    except Exception:
                        pass
                    break  # Break out to cancel the rest of the job completely
                print(f"Error processing video resolution quality {q}: {e}")
                try:
                    await client.send_message(chat_id, f"⚠️ **Error processing/uploading {q} resolution:** `{e}`")
                except Exception:
                    pass
                    
        # Delete temporary status message from channel to keep channel feed clean
        if status_msg_node_id:
            try:
                await client.delete_messages(target_chat, status_msg_node_id)
            except Exception as e:
                print(f"Error deleting status message: {e}")
                
        # Send buttons message/poster if button_mode is on
        if button_mode and generated_links:
            btn_post_fmt = settings.get("button_post_format")
            if not btn_post_fmt:
                btn_post_fmt = "<b>{anime} | Tamil Dubbed #Official</b>\n\n<b>Season : {season} | Episode : {episode}</b>\n\n<b>‼️Note - Click The Below Button to Get Episodes 👇</b>"
            
            post_caption = safe_format(
                btn_post_fmt,
                anime=anime,
                season=job.get('season', '01'),
                episode=job.get('episode', '01')
            )
            
            inline_buttons = []
            for qual, link in generated_links:
                inline_buttons.append(InlineKeyboardButton(f"📥 {qual}", url=link))
                
            rows = [inline_buttons[x:x+2] for x in range(0, len(inline_buttons), 2)]
            reply_markup = InlineKeyboardMarkup(rows)
            button_post_thumb = custom_thumb
            if settings.get("dest_poster_mode") == "on":
                button_post_thumb = await db.match_poster(user_id, anime) or custom_thumb
            
            if button_post_thumb:
                try:
                    await client.send_photo(
                        chat_id=target_chat,
                        photo=button_post_thumb,
                        caption=post_caption,
                        parse_mode=enums.ParseMode.HTML,
                        reply_markup=reply_markup
                    )
                except Exception as e:
                    print(f"Error posting photo to channel: {e}. Falling back to text.")
                    await client.send_message(
                        chat_id=target_chat,
                        text=post_caption,
                        parse_mode=enums.ParseMode.HTML,
                        reply_markup=reply_markup
                    )
            else:
                await client.send_message(
                    chat_id=target_chat,
                    text=post_caption,
                    parse_mode=enums.ParseMode.HTML,
                    reply_markup=reply_markup
                )

        # Announce the completed upload in the user's configured update channel.
        if uploaded_success_count > 0 and settings.get("main_channel_mode") == "on" and settings.get("main_channel_id"):
            try:
                main_chat = settings["main_channel_id"]
                if isinstance(main_chat, str) and main_chat.lstrip("-").isdigit():
                    main_chat = int(main_chat)
                destination_link = generated_links[0][1] if generated_links else ""
                channel_link = ""
                if channel and channel.get("link"):
                    channel_link = channel["link"]
                else:
                    dest_info = await client.get_chat(target_chat)
                    channel_link = (f"https://t.me/{dest_info.username}" if dest_info.username
                                    else (dest_info.invite_link or ""))
                if not destination_link:
                    destination_link = channel_link

                episode_value = str(job.get("episode", "01"))
                season_value = str(job.get("season", "01"))
                try:
                    current_episode = int(re.search(r"\d+", episode_value).group())
                except Exception:
                    current_episode = None
                current_qualities = [q for q, _ in generated_links] if generated_links else ["-"]
                previous = await db.get_last_main_post(user_id, anime, season_value) if current_episode is not None else None
                episode_display = episode_value
                qualities = current_qualities
                range_first = range_last = current_episode
                edit_existing = False
                if previous and current_episode is not None and settings.get("main_batch_update", "on") == "on":
                    age = time.time() - previous.get("timestamp", 0)
                    if str(previous.get("channel_id")) == str(main_chat) and 0 <= age < 14400:
                        old_last = int(previous.get("last_episode", current_episode))
                        if old_last != current_episode:
                            range_first = min(int(previous.get("first_episode", old_last)), current_episode)
                            range_last = max(old_last, current_episode)
                            qualities = list(dict.fromkeys(previous.get("quals", []) + current_qualities))
                        else:
                            range_first = int(previous.get("first_episode", current_episode))
                            range_last = old_last
                            qualities = list(dict.fromkeys(previous.get("quals", []) + current_qualities))
                        episode_display = (f"{range_first:02d}" if range_first == range_last
                                           else f"{range_first:02d} - {range_last:02d}")
                        edit_existing = True
                quality_text = ", ".join(qualities)
                caption_template = settings.get("main_channel_caption") or "<b>{anime}</b>\nSeason {season} • Episode {episode}\n{quality}"
                episode_token = "__MAIN_CHANNEL_EPISODE__"
                caption_template = re.sub(r"\{episode(?::[^}]+)?\}", episode_token, caption_template)
                main_caption = safe_format(
                    caption_template, anime=anime, season=season_value, episode=current_episode or 0,
                    quality=quality_text, language=job.get("language", "")
                ).replace(episode_token, episode_display)
                main_buttons = []
                for cfg in settings.get("main_channel_buttons", []):
                    url = str(cfg.get("url", "{dest_post_link}"))
                    url = url.replace("{dest_post_link}", destination_link).replace("{channel_link}", channel_link)
                    url = url.replace("{anime}", str(anime)).replace("{season}", str(job.get("season", "01")))
                    url = url.replace("{episode}", episode_display)
                    if url:
                        label = str(cfg.get("text", "Open"))
                        if settings.get("colored_buttons", "on") == "on":
                            style = cfg.get("style")
                            if not style:
                                legacy_color = str(cfg.get("color", "" )).lower()
                                style = {"🔵": "primary", "blue": "primary", "primary": "primary",
                                         "🟢": "success", "green": "success", "success": "success",
                                         "🔴": "danger", "red": "danger", "danger": "danger"}.get(legacy_color)
                            style_emoji = {"primary": "🔵", "success": "🟢", "danger": "🔴"}.get(style, "")
                            if style_emoji and not label.startswith(style_emoji):
                                label = f"{style_emoji} {label}"
                        main_buttons.append(InlineKeyboardButton(label, url=url))
                layout = []
                try:
                    widths = [max(1, int(n.strip())) for n in settings.get("main_buttons_layout", "1, 2").split(",")]
                    pos = 0
                    for width in widths:
                        if pos >= len(main_buttons):
                            break
                        layout.append(main_buttons[pos:pos + width])
                        pos += width
                    if pos < len(main_buttons):
                        layout.extend([main_buttons[i:i + 2] for i in range(pos, len(main_buttons), 2)])
                except Exception:
                    layout = [main_buttons[i:i + 2] for i in range(0, len(main_buttons), 2)]
                markup = InlineKeyboardMarkup(layout) if layout else None
                poster = await db.match_poster(user_id, anime) or custom_thumb
                edited = False
                if edit_existing and previous:
                    try:
                        if previous.get("is_photo"):
                            await client.edit_message_caption(main_chat, previous["message_id"], main_caption,
                                parse_mode=enums.ParseMode.HTML, reply_markup=markup)
                        else:
                            await client.edit_message_text(main_chat, previous["message_id"], main_caption,
                                parse_mode=enums.ParseMode.HTML, reply_markup=markup)
                        edited = True
                    except Exception as edit_error:
                        # Avoid duplicate announcements if Telegram rejects an edit.
                        print(f"Could not edit existing main channel post: {edit_error}")
                if not edit_existing:
                    if poster:
                        try:
                            sent = await client.send_photo(main_chat, poster, caption=main_caption,
                                parse_mode=enums.ParseMode.HTML, reply_markup=markup)
                            is_photo = True
                        except Exception:
                            sent = await client.send_message(main_chat, main_caption,
                                parse_mode=enums.ParseMode.HTML, reply_markup=markup)
                            is_photo = False
                    else:
                        sent = await client.send_message(main_chat, main_caption,
                            parse_mode=enums.ParseMode.HTML, reply_markup=markup)
                        is_photo = False
                    previous = {"message_id": sent.id, "is_photo": is_photo}
                if current_episode is not None and (not edit_existing or edited):
                    await db.save_last_main_post(user_id, anime, season_value, {
                        "channel_id": main_chat,
                        "message_id": previous["message_id"],
                        "is_photo": previous.get("is_photo", False),
                        "first_episode": range_first,
                        "last_episode": range_last,
                        "timestamp": time.time(),
                        "quals": qualities,
                        "dest_post_link": destination_link
                    })
            except Exception as e:
                print(f"Error posting upload announcement to main channel: {e}")
                
        # 5. Post completion sticker if configured
        finish_sticker = settings.get("finish_sticker")
        if finish_sticker:
            try:
                # Send to target chat
                await client.send_sticker(target_chat, sticker=finish_sticker)
            except Exception as e:
                print(f"Error sending final sticker: {e}")
                
        # Final notification in user's PM
        try:
            if job.get('user_id') in USER_CANCELLATIONS:
                await client.edit_message_text(chat_id, status_msg, "❌ **Task was cancelled by user.**")
            elif uploaded_success_count > 0:
                await client.edit_message_text(chat_id, status_msg, f"✅ **Processing Completed!**\nFiles successfully posted in target channel.")
            else:
                await client.edit_message_text(chat_id, status_msg, f"❌ **Processing Failed!**\nNo files were successfully uploaded.")
        except Exception:
            pass

    except Exception as re_err:
        if str(re_err) == "TaskCancelledByUser":
            try:
                await client.edit_message_text(chat_id, status_msg, "❌ **Task Cancelled.** Moving next...")
            except Exception: pass
            print(f"Job cancelled for user {job.get('user_id')}")
        else:
            try:
                await client.edit_message_text(chat_id, status_msg, f"❌ **Error during process:** `{re_err}`")
            except Exception: pass
            print(f"Job crashed with error: {re_err}")
    finally:
        shutil.rmtree(work_dir, ignore_errors=True)
        log_msg_id = job.get("log_msg_id")
        if log_msg_id and Config.LOG_CHANNEL:
            try:
                await client.delete_messages(Config.LOG_CHANNEL, log_msg_id)
            except Exception as e:
                print(f"Error deleting job trigger log message: {e}")