import asyncio import csv import os from typing import Any from telethon import TelegramClient, functions, types from telethon.errors import FloodWaitError, RPCError API_ID = int(os.getenv("TG_API_ID", "0")) API_HASH = os.getenv("TG_API_HASH", "") BOT_USERNAME = os.getenv("BOT_USERNAME", "") VIP_CHAT = os.getenv("VIP_CHAT", "") OUTPUT_FILE = "hasil_recovery_vip.csv" def normalize_chat_reference(value: str) -> str | int: cleaned = value.strip() if cleaned.lstrip("-").isdigit(): return int(cleaned) return cleaned def format_datetime(value: Any) -> str: if not value: return "" try: return value.isoformat() except AttributeError: return str(value) async def request_with_flood_wait(client, request): while True: try: return await client(request) except FloodWaitError as error: wait_seconds = int(error.seconds) + 1 print(f"Telegram meminta jeda {wait_seconds} detik. Script akan melanjutkan otomatis.") await asyncio.sleep(wait_seconds) async def get_all_invite_links(client, chat, link_creator, revoked: bool): invite_links = [] offset_date = None offset_link = None while True: result = await request_with_flood_wait( client, functions.messages.GetExportedChatInvitesRequest( peer=chat, admin_id=link_creator, offset_date=offset_date, offset_link=offset_link, limit=100, revoked=revoked, ), ) batch = list(result.invites) if not batch: break invite_links.extend(batch) print(f"Berhasil membaca {len(invite_links)} link {'revoked' if revoked else 'aktif'}.") if len(batch) < 100: break last_invite = batch[-1] offset_date = last_invite.date offset_link = last_invite.link return invite_links async def get_link_importers(client, chat, invite_link: str): return await request_with_flood_wait( client, functions.messages.GetChatInviteImportersRequest( peer=chat, link=invite_link, q=None, offset_date=0, offset_user=types.InputUserEmpty(), limit=10, requested=False, subscription_expired=False, ), ) async def main(): if not API_ID or not API_HASH: raise RuntimeError("TG_API_ID dan TG_API_HASH belum diisi.") if not BOT_USERNAME: raise RuntimeError("BOT_USERNAME belum diisi.") if not VIP_CHAT: raise RuntimeError("VIP_CHAT belum diisi.") chat_reference = normalize_chat_reference(VIP_CHAT) async with TelegramClient("recovery_session", API_ID, API_HASH) as client: print("Login akun Telegram berhasil.") await client.get_dialogs() chat = await client.get_input_entity(chat_reference) bot_creator = await client.get_input_entity(BOT_USERNAME) print("Mengambil link aktif...") active_links = await get_all_invite_links(client, chat, bot_creator, revoked=False) print("Mengambil link revoked...") revoked_links = await get_all_invite_links(client, chat, bot_creator, revoked=True) all_links = [("ACTIVE", invite) for invite in active_links] + [("REVOKED", invite) for invite in revoked_links] print(f"Total link ditemukan: {len(all_links)}") rows = [] for index, (link_status, invite) in enumerate(all_links, start=1): invite_url = invite.link usage = getattr(invite, "usage", 0) or 0 link_name = getattr(invite, "title", "") or "" print(f"[{index}/{len(all_links)}] Membaca pengguna: {invite_url}") try: result = await get_link_importers(client, chat, invite_url) users_by_id = {user.id: user for user in result.users} if not result.importers: rows.append({ "invite_link": invite_url, "link_name": link_name, "link_status": link_status, "usage": usage, "user_id": "", "username": "", "first_name": "", "last_name": "", "joined_at": "", "link_created_at": format_datetime(getattr(invite, "date", None)), "recovery_status": "BELUM_DIPAKAI" if usage == 0 else "USER_TIDAK_DITEMUKAN", "error": "", }) else: for importer in result.importers: user_id = importer.user_id user = users_by_id.get(user_id) rows.append({ "invite_link": invite_url, "link_name": link_name, "link_status": link_status, "usage": usage, "user_id": user_id, "username": getattr(user, "username", "") or "" if user else "", "first_name": getattr(user, "first_name", "") or "" if user else "", "last_name": getattr(user, "last_name", "") or "" if user else "", "joined_at": format_datetime(getattr(importer, "date", None)), "link_created_at": format_datetime(getattr(invite, "date", None)), "recovery_status": "BERHASIL", "error": "", }) except RPCError as error: rows.append({ "invite_link": invite_url, "link_name": link_name, "link_status": link_status, "usage": usage, "user_id": "", "username": "", "first_name": "", "last_name": "", "joined_at": "", "link_created_at": format_datetime(getattr(invite, "date", None)), "recovery_status": "GAGAL", "error": f"{type(error).__name__}: {error}", }) await asyncio.sleep(0.4) fieldnames = [ "invite_link", "link_name", "link_status", "usage", "user_id", "username", "first_name", "last_name", "joined_at", "link_created_at", "recovery_status", "error", ] with open(OUTPUT_FILE, "w", newline="", encoding="utf-8-sig") as csv_file: writer = csv.DictWriter(csv_file, fieldnames=fieldnames) writer.writeheader() writer.writerows(rows) recovered_ids = {str(row["user_id"]) for row in rows if row["user_id"]} unused_links = sum(1 for row in rows if row["recovery_status"] == "BELUM_DIPAKAI") failed_links = sum(1 for row in rows if row["recovery_status"] == "GAGAL") print() print("========== HASIL ==========") print(f"Total link dibaca : {len(all_links)}") print(f"User ID ditemukan: {len(recovered_ids)}") print(f"Link belum dipakai: {unused_links}") print(f"Link gagal dibaca : {failed_links}") print(f"File hasil : {OUTPUT_FILE}") if __name__ == "__main__": asyncio.run(main())