| 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 |
| |
| 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: |
| |
| 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 "" |
| |
| |
| 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") |
| |
| |
| 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] |
| |
| |
| 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) |
| |
| |
| 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]) |
| |
| |
| 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}"]) |
| |
| |
| cmd_args.extend(["-map", "0:v"]) |
| |
| 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?"]) |
| |
| |
| 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?"]) |
| |
| cmd_args.extend(["-map", "0:t?"]) |
| |
| |
| if default_audio is not None: |
| cmd_args.extend(["-disposition:a", "0"]) |
| 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") |
| |
| |
| 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) |
| |
| |
| 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}") |
| |
| |
| 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 |
| |
| |
| auto_routing = settings.get('auto_channel_match', 'on') == 'on' |
| target_chat = job.get('target_channel_id') |
| channel = None |
| |
| if auto_routing: |
| |
| 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 |
| |
| |
| 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}") |
| |
| |
| 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) |
| |
| |
| 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 |
| |
| try: |
| await client.edit_message_text(chat_id, status_msg, text, parse_mode=enums.ParseMode.HTML) |
| except Exception: |
| pass |
|
|
| |
| 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...") |
| |
| |
| 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)) |
| |
| |
| |
| |
| |
| 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: |
| |
| 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 |
|
|
| |
| |
| 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") |
| |
| |
| 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") |
| |
| |
| 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.") |
| |
| |
| 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 |
| |
| |
| 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) |
| |
| |
| 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 |
| ) |
| |
| |
| 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}") |
| |
| |
| 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 |
| 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 |
| |
| |
| 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}") |
| |
| |
| 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 |
| ) |
|
|
| |
| 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: |
| |
| 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}") |
| |
| |
| finish_sticker = settings.get("finish_sticker") |
| if finish_sticker: |
| try: |
| |
| await client.send_sticker(target_chat, sticker=finish_sticker) |
| except Exception as e: |
| print(f"Error sending final sticker: {e}") |
| |
| |
| 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}") |
|
|
