#!/usr/bin/env python3 # -*- coding: utf-8 -*- ####################################################### # PRIVATE LICENSE NOTICE # # © 2025 AKN-DEV. All Rights Reserved. # # # # This software is licensed under PRIVATE LICENSE. # # Unauthorized use, modification, distribution, # # or replication is strictly prohibited. # # # # LEGAL OWNER: AKN-DEV TEAM # # CONTACT: @aknuserbot (Telegram) # # # # VIOLATIONS WILL RESULT IN: # # - Automatic bot termination # # - Legal action under DMCA/EU Copyright Law # # - Permanent blacklist from all products # ####################################################### import re import os import logging import time import asyncio import sys import traceback from pyrogram import idle, errors from akn import assistant, app from config import * from typing import List from akn.utils.database import db from akn.clientmulti_bot import * from akn.utils.version_utils import validate_version from akn.utils.logger import LOGS from traceback import print_exc import traceback # ---- CONFIGURATION ---- # WARNING: Editing below this line voids your license! VALID_CODE_LICENSE = "D81B4FBB" VALID_DATE_LICENSE = "20250411" VALID_USERNAME_BOT_LICENSE = "aknuserbot" # ---- FUNCTIONS PRIVATE ---- def uvloop_ultra_new(): try: import uvloop uvloop.install() except ImportError: pass def logging_info_warning(): logging.basicConfig(level=logging.INFO) logging.getLogger("pyrogram.syncer").setLevel(logging.WARNING) logging.getLogger("pyrogram.client").setLevel(logging.WARNING) def get_event_loop(): return asyncio.get_event_loop() async def check_export_getenv(): import os envs = "\n".join([f"{key}={value}" for key, value in os.environ.items()]) with open("envs.txt", "w") as f: f.write(envs) await assistant.send_document(PRIVATE_LOGS, "envs.txt", caption="Environment Variables Private") async def main(): try: await db.connect() await license_private_start_bot(app, assistant) await idle() except Exception as e: LOGS.error(f"Error in main: {e}") traceback.print_exc() finally: tasks = [task for task in asyncio.all_tasks() if task is not asyncio.current_task()] [task.cancel() for task in tasks] await asyncio.gather(*tasks, return_exceptions=True) LOGS.info("All tasks completed successfully!") def faster_launcher_loaded(loop): start_time = time.monotonic() loop.run_until_complete(main()) duration = time.monotonic() - start_time LOGS.info(f"Clean shutdown completed in {duration:.2f}s") def _shutdown_loop(loop): try: pending = asyncio.all_tasks(loop) for task in pending: task.cancel() loop.run_until_complete( asyncio.gather(*pending, return_exceptions=True) ) loop.run_until_complete(loop.shutdown_asyncgens()) loop.close() LOGS.info("Event loop resources released") except Exception as e: LOGS.error(f"Shutdown error: {str(e)}") os._exit(1) async def client_multi_bot(): bots = [ start_user, start_gemini_bot, start_magic_bot, start_meta_bot, start_youtube_bot, start_session_bot, start_captcha_bot, start_all_downloader_bot ] running_tasks: List[asyncio.Task] = [] try: for bot in bots: task = asyncio.create_task(_safe_bot_launch(bot)) running_tasks.append(task) await asyncio.gather(*running_tasks) except Exception as e: LOGS.critical(f"⛔ Multi-bot crash: {type(e).__name__}") print_exc() await _emergency_cleanup(running_tasks) finally: LOGS.info("🔌 All bots terminated") async def _safe_bot_launch(bot_func): try: is_valid, msg = validate_version(bot_func.__name__) if not is_valid: LOGS.error(f"❌ {bot_func.__name__}: {msg}") return False await bot_func() LOGS.info(f"✅ {bot_func.__name__} launched & version checked: {msg}") return True except asyncio.CancelledError: LOGS.warning(f"🛑 {bot_func.__name__} forced shutdown") except Exception as e: LOGS.error(f"⚠️ {bot_func.__name__} failed: {str(e)}") return False async def _emergency_cleanup(tasks: List[asyncio.Task]): LOGS.warning("🚨 Initiating emergency cleanup...") for task in tasks: if not task.done(): task.cancel() await asyncio.sleep(1) for task in tasks: if not task.done(): LOGS.error(f"💀 Force-killing {task.get_name()}") del task async def license_private_start_bot(app, assistant): try: await asyncio.gather( app.start(), assistant.start() ) is_valid, reason = await validate_license(assistant) if not is_valid: LOGS.error(f"🚨 LICENSE FAILED: {reason}") await emergency_shutdown(app, assistant) return if assistant.me.username != VALID_USERNAME_BOT_LICENSE: LOGS.error(f"🆔 UNAUTHORIZED BOT: {assistant.me.username}") await emergency_shutdown(app, assistant) return LOGS.info(f"✅ Licensed Bot Active: @{assistant.me.username}") # await check_export_getenv() await client_multi_bot() asyncio.create_task( connection_watchdog(assistant), name=f"health_dev_{assistant.me.id}" ) except errors.FloodWait as e: LOGS.error(f"⏳ FloodWait: {e.value} seconds") await asyncio.sleep(e.value) # except errors.RPCError as e: LOGS.error(f"⚠️ RPC Error: {str(e)}") await emergency_shutdown(app, assistant) except Exception as e: LOGS.error(f"⚡ CRITICAL ERROR: {str(e)}") traceback.print_exc() await emergency_shutdown(app, assistant) async def emergency_shutdown(app, assistant): try: await asyncio.gather( app.stop(), assistant.stop() ) finally: sys.exit(1) async def validate_license(c): try: doc_lice = await c.get_messages("LicenseAknBotDB", 3) if not doc_lice or not doc_lice.document: return False, "License document not found" filename = doc_lice.document.file_name match = re.match(r"^(.+)_AKNUSERBOT-(\d{8})-([A-F0-9]{8})\.pdf$", filename) if not match: return False, "Invalid filename format" _, date, license_code = match.groups() if license_code != VALID_CODE_LICENSE: return False, "License code mismatch" if date != VALID_DATE_LICENSE: return False, "License expired" return True, "Valid license" except Exception as e: return False, f"Validation error: {str(e)}" ####################################################### # DO NOT REMOVE THIS LICENSE NOTICE! # # This code contains proprietary trade secrets of # # AKN-DEV. Any removal will result in legal action. # #######################################################