""" Telegram Downloader — Universal (Channels, Private Groups, Topics & Direct Chats) --------------------------------------------------------------------------------- Supports: - web.telegram.org/a/#7564523801 (Direct User/Bot Chats) - web.telegram.org/a/#-1004422937739_2 (Forum Topics) - t.me/c/4422937739/2 (Topic Links) - t.me/channelname or @channelname (Public Channels) Features: - Fast chunk streaming (powered by cryptg C-acceleration) - Auto-names folder using the chat's actual name/title - Sequential ordered numbering (0001_, 0002_, ...) - Resume support (skips already finished files) """ import asyncio import re import sys from pathlib import Path try: from telethon import TelegramClient from telethon.errors import FloodWaitError from telethon.tl.types import MessageMediaDocument, MessageMediaPhoto from tqdm import tqdm except ImportError: print("❌ Missing packages. Run: pip install telethon tqdm cryptg") sys.exit(1) try: import cryptg # Telethon uses this automatically when installed. CRYPTG_ENABLED = True except ImportError: CRYPTG_ENABLED = False # ════════════════════════ CONFIG ════════════════════════ API_ID = 22910426 API_HASH = "57990461eb01f4d8171ba06f8fb70cd2" BASE_DIR = Path(r"C:\Users\nobit\Documents\Telegram Downloader\Downloads") # Download concurrency. 4 is a good starting point for a fast connection. # If Telegram starts returning FloodWait often, reduce this to 3. PARALLEL_FILES = 4 # Chunk size per request (512 KB = Telegram max) CHUNK = 512 * 1024 # Automatic retry settings MAX_RETRIES = 5 RETRY_BASE_DELAY = 3 # ── Allowed file types ────────────────────────────────── ALLOWED_MIME = { # Video "video/mp4", "video/x-matroska", "video/webm", "video/quicktime", "video/x-msvideo", "video/mpeg", "video/3gpp", "video/x-flv", # Archives "application/zip", "application/x-zip-compressed", "application/x-rar-compressed", "application/x-7z-compressed", "application/x-tar", "application/gzip", "application/x-bzip2", # Documents / PDFs "application/pdf", "application/octet-stream", } ALLOWED_EXT = { ".mp4", ".mkv", ".webm", ".mov", ".avi", ".mpeg", ".3gp", ".flv", ".zip", ".rar", ".7z", ".tar", ".gz", ".bz2", ".pdf", ".doc", ".docx", ".ppt", ".pptx", } # ════════════════════════════════════════════════════════ def safe_filename(name: str) -> str: return re.sub(r'[\\/:*?"<>|]', "_", name).strip() or "file" def resolve_channel(channel: str): """ Parses any Telegram link or identifier: - https://web.telegram.org/a/#-1004422937739_2 -> (-1004422937739, 2) - https://web.telegram.org/a/#7564523801 -> (7564523801, None) - https://web.telegram.org/k/#@username -> ('username', None) - https://t.me/c/4422937739/2 -> (-1004422937739, 2) - 7564523801 -> (7564523801, None) - @channelname -> ('channelname', None) """ clean = channel.strip() # 1. web.telegram.org/a/#... or web.telegram.org/k/#... m_web = re.search(r"web\.telegram\.org/[ak]/#(-?\d+)(?:_(\d+))?", clean) if m_web: raw_id = m_web.group(1) topic_id = int(m_web.group(2)) if m_web.group(2) else None entity = int(raw_id) return entity, topic_id # 2. t.me/c/{chat_id}/{topic_id} m_c = re.search(r"t\.me/c/(\d+)(?:/(\d+))?", clean) if m_c: entity = int(f"-100{m_c.group(1)}") topic_id = int(m_c.group(2)) if m_c.group(2) else None return entity, topic_id # 3. Raw number (e.g. 7564523801 or -1004422937739) if re.match(r"^-?\d+$", clean): return int(clean), None # 4. Standard public link: t.me/username or @username name = clean.strip("/").split("/")[-1].lstrip("@") return name, None def is_wanted(media) -> bool: if not media: return False if isinstance(media, MessageMediaPhoto): return True if isinstance(media, MessageMediaDocument) and media.document: doc = media.document mime = (doc.mime_type or "").lower() if mime in ALLOWED_MIME: return True for attr in doc.attributes: if type(attr).__name__ in ("DocumentAttributeVideo", "DocumentAttributeAudio"): return True if hasattr(attr, "file_name") and attr.file_name: if Path(attr.file_name).suffix.lower() in ALLOWED_EXT: return True return False def get_filename(msg, idx: int) -> str: date_str = msg.date.strftime("%Y%m%d_%H%M%S") media = msg.media if isinstance(media, MessageMediaPhoto): return f"{idx:04d}_photo_{date_str}_id{msg.id}.jpg" if isinstance(media, MessageMediaDocument) and media.document: doc = media.document for attr in doc.attributes: if hasattr(attr, "file_name") and attr.file_name: return f"{idx:04d}_{safe_filename(attr.file_name.strip())}" ext = (doc.mime_type or "application/octet-stream").split("/")[-1] return f"{idx:04d}_file_{date_str}_id{msg.id}.{ext}" return f"{idx:04d}_media_{date_str}_id{msg.id}.bin" # ── Global stats ────────────────────────────────────────── stats = {"done": 0, "skipped": 0, "failed": 0} stats_lock = asyncio.Lock() async def download_one(client, msg, out_path: Path, label: str, pos: int, overall: tqdm): """ Download a single media item. Improvements over the original version: - resumes an incomplete file from the last complete 512 KB chunk - keeps partial files instead of deleting them after a network error - retries temporary failures automatically - handles Telegram FloodWait cleanly """ doc = getattr(msg.media, "document", None) size_bytes = getattr(doc, "size", 0) or 0 # ── Skip if already downloaded ──────────────────────── if out_path.exists() and size_bytes and out_path.stat().st_size >= size_bytes - 1024: async with stats_lock: stats["skipped"] += 1 overall.set_postfix(done=stats["done"], skip=stats["skipped"], fail=stats["failed"]) overall.update(1) return # If Telegram did not report a size and a non-empty file already exists, # leave it alone rather than risk overwriting an existing download. if out_path.exists() and not size_bytes and out_path.stat().st_size > 0: async with stats_lock: stats["skipped"] += 1 overall.set_postfix(done=stats["done"], skip=stats["skipped"], fail=stats["failed"]) overall.update(1) return existing = out_path.stat().st_size if out_path.exists() else 0 # iter_download works best when the resume offset is aligned to the request size. # If Windows/network interruption left a partial chunk, trim only that last chunk. resume_from = existing - (existing % CHUNK) if existing else 0 if out_path.exists() and existing != resume_from: with open(out_path, "r+b") as f: f.truncate(resume_from) file_bar = tqdm( total=size_bytes or None, initial=resume_from, desc=label, unit="B", unit_scale=True, unit_divisor=1024, position=pos, leave=False, ncols=90, colour="cyan", ) completed = False last_error = None for attempt in range(1, MAX_RETRIES + 1): try: current = out_path.stat().st_size if out_path.exists() else 0 # Re-align after any interrupted attempt. aligned = current - (current % CHUNK) if current else 0 if out_path.exists() and current != aligned: with open(out_path, "r+b") as f: f.truncate(aligned) # Keep the progress bar synchronized with what is actually on disk. file_bar.n = aligned file_bar.refresh() mode = "ab" if aligned else "wb" with open(out_path, mode) as f: async for chunk in client.iter_download( msg.media, offset=aligned, request_size=CHUNK, ): f.write(chunk) file_bar.update(len(chunk)) final_size = out_path.stat().st_size if out_path.exists() else 0 # If Telegram supplied a size, verify it. Otherwise reaching the end # of iter_download counts as success. if size_bytes and final_size < size_bytes - 1024: raise IOError( f"Incomplete download: {final_size}/{size_bytes} bytes" ) completed = True file_bar.set_description(f"✅ {label}") async with stats_lock: stats["done"] += 1 break except FloodWaitError as e: last_error = e wait_for = int(getattr(e, "seconds", 0)) + 1 tqdm.write( f" ⏳ FloodWait [{label}] — Telegram asked us to wait " f"{wait_for}s. Resuming afterward..." ) await asyncio.sleep(wait_for) except Exception as e: last_error = e if attempt < MAX_RETRIES: delay = RETRY_BASE_DELAY * attempt tqdm.write( f" ⚠️ Retry {attempt}/{MAX_RETRIES} [{label}] in {delay}s: {e}" ) await asyncio.sleep(delay) else: break if not completed: file_bar.set_description(f"❌ {label}") async with stats_lock: stats["failed"] += 1 tqdm.write( f" ❌ Failed [{label}] after {MAX_RETRIES} attempts: {last_error}" ) tqdm.write( " Partial data was kept, so the next run can resume this file." ) file_bar.close() async with stats_lock: overall.set_postfix(done=stats["done"], skip=stats["skipped"], fail=stats["failed"]) overall.update(1) async def main(channel: str): entity_input, topic_id = resolve_channel(channel) session_path = str(Path(__file__).parent / "session_downloader") client = TelegramClient(session_path, API_ID, API_HASH, connection_retries=10, retry_delay=2) await client.start() print("✅ Connected to Telegram!\n") # Resolve entity info and determine folder name try: chat_entity = await client.get_entity(entity_input) title = getattr(chat_entity, "title", None) or getattr(chat_entity, "first_name", None) or str(entity_input) folder_name = safe_filename(title) if topic_id: folder_name += f"_topic_{topic_id}" except Exception: chat_entity = entity_input folder_name = f"chat_{str(entity_input).replace('-', '_')}" if topic_id: folder_name += f"_topic_{topic_id}" out_dir = BASE_DIR / folder_name out_dir.mkdir(parents=True, exist_ok=True) print(f"📁 Saving to : {out_dir}") print(f"⚡ Parallel files: {PARALLEL_FILES} at a time") print(f"🚀 cryptg : {'enabled' if CRYPTG_ENABLED else 'NOT installed (downloads may be slower)' }") if topic_id: print(f"🧵 Topic ID : {topic_id} (Forum Topic)") print() # ── Scan messages ───────────────────────────────────── print(f"📡 Scanning chat messages...") wanted = [] async for msg in client.iter_messages(chat_entity, reverse=True): if topic_id is not None: belongs = (msg.id == topic_id) if not belongs and msg.reply_to: top = getattr(msg.reply_to, "reply_to_top_id", None) direct = getattr(msg.reply_to, "reply_to_msg_id", None) belongs = (top == topic_id or direct == topic_id) if not belongs: continue if is_wanted(msg.media): wanted.append(msg) total = len(wanted) print(f"🗂️ Found {total} media file(s) to download\n") if total == 0: print("Nothing found. Make sure there is media in this chat.") await client.disconnect() return # ── Download files ──────────────────────────────────── semaphore = asyncio.Semaphore(PARALLEL_FILES) overall_bar = tqdm(total=total, desc="📦 Overall", unit="file", position=0, colour="green", ncols=80) async def guarded_download(msg, idx): filename = get_filename(msg, idx) out_path = out_dir / filename label = filename[:35] slot = (idx - 1) % PARALLEL_FILES + 1 async with semaphore: await download_one(client, msg, out_path, label, slot, overall_bar) tasks = [guarded_download(msg, i) for i, msg in enumerate(wanted, 1)] await asyncio.gather(*tasks) overall_bar.close() await client.disconnect() print("\n" + "═" * 55) print(f"✅ Downloaded : {stats['done']}") print(f"⏭️ Skipped : {stats['skipped']} (already existed)") print(f"❌ Failed : {stats['failed']}") print(f"📁 Saved to : {out_dir}") print("═" * 55) if __name__ == "__main__": if len(sys.argv) < 2: print("Usage: python download_videos.py ") sys.exit(1) asyncio.run(main(sys.argv[1]))