import os import asyncio import threading from telethon import TelegramClient, events from telethon.errors import SessionPasswordNeededError from modules.config import ConfigManager from modules.logger import logger from modules.database import query_db, execute_db SESSION_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'sessions') SESSION_FILE = os.path.join(SESSION_DIR, 'telegram_session') class TelegramManager: _instance = None _lock = threading.Lock() def __new__(cls): with cls._lock: if cls._instance is None: cls._instance = super(TelegramManager, cls).__new__(cls) cls._instance._init_manager() return cls._instance def _init_manager(self): self.loop = None self.thread = None self.client = None self.phone_code_hash = None self.phone = None self.api_id = None self.api_hash = None if not os.path.exists(SESSION_DIR): os.makedirs(SESSION_DIR, exist_ok=True) # Start background event loop thread self._start_loop() def _start_loop(self): self.loop = asyncio.new_event_loop() self.thread = threading.Thread(target=self._run_loop, daemon=True) self.thread.start() logger.info("Background asyncio event loop thread started.") def _run_loop(self): asyncio.set_event_loop(self.loop) self.loop.run_forever() def run_coroutine(self, coro): """Safely submits a coroutine to the background event loop and waits for results.""" future = asyncio.run_coroutine_threadsafe(coro, self.loop) return future.result() async def _init_client(self, api_id, api_hash): """Creates the Telethon TelegramClient instance.""" if self.client: await self.client.disconnect() # Initialize client with SQLite session storage self.client = TelegramClient(SESSION_FILE, int(api_id), api_hash) await self.client.connect() def is_connected(self) -> bool: """Checks if the client is connected to Telegram.""" if not self.client: return False try: return self.run_coroutine(self._is_connected_async()) except Exception: return False async def _is_connected_async(self): return self.client and await self.client.is_user_authorized() def start_bot(self): """Initializes and runs the Telegram auto-reply listener bot.""" config = ConfigManager() api_id = config.get('tg_api_id') api_hash = config.get('tg_api_hash') if not api_id or not api_hash: logger.warning("Telegram client cannot start: API ID/Hash missing.") return False try: self.run_coroutine(self._start_bot_async(api_id, api_hash)) logger.info("Telegram Auto-Reply listener successfully started.") return True except Exception as e: logger.error(f"Failed to start Telegram Auto-Reply client: {e}") return False async def _start_bot_async(self, api_id, api_hash): await self._init_client(api_id, api_hash) if await self.client.is_user_authorized(): # Register message handlers from modules.autoreply import register_handlers register_handlers(self.client) logger.info("Autoreply message handlers registered.") else: logger.warning("Telegram client is connected but NOT authorized.") def stop_bot(self): """Disconnects the Telethon client.""" if self.client: try: self.run_coroutine(self.client.disconnect()) logger.info("Telegram client disconnected.") except Exception as e: logger.error(f"Error disconnecting client: {e}") def send_otp(self, api_id, api_hash, phone) -> str: """Initiates the connection and requests an OTP code.""" self.api_id = api_id self.api_hash = api_hash self.phone = phone return self.run_coroutine(self._send_otp_async(api_id, api_hash, phone)) async def _send_otp_async(self, api_id, api_hash, phone): await self._init_client(api_id, api_hash) # Send code request result = await self.client.send_code_request(phone) self.phone_code_hash = result.phone_code_hash return result.phone_code_hash def verify_otp(self, code) -> dict: """Verifies the received OTP. If 2FA is needed, throws a specific error code.""" try: return self.run_coroutine(self._verify_otp_async(code)) except SessionPasswordNeededError: logger.info("2FA password is required to login.") return {"status": "2fa_required"} except Exception as e: logger.error(f"OTP verification failed: {e}") return {"status": "error", "message": str(e)} async def _verify_otp_async(self, code): try: user = await self.client.sign_in(self.phone, code, phone_code_hash=self.phone_code_hash) # Save configs config = ConfigManager() config.set('tg_api_id', self.api_id) config.set('tg_api_hash', self.api_hash) config.set('tg_phone', self.phone) # Start handlers from modules.autoreply import register_handlers register_handlers(self.client) return {"status": "success", "user": user.to_dict() if hasattr(user, 'to_dict') else str(user)} except SessionPasswordNeededError: raise def verify_2fa(self, password) -> dict: """Submits 2FA password to complete authentication.""" try: return self.run_coroutine(self._verify_2fa_async(password)) except Exception as e: logger.error(f"2FA login failed: {e}") return {"status": "error", "message": str(e)} async def _verify_2fa_async(self, password): user = await self.client.sign_in(password=password) # Save configs config = ConfigManager() config.set('tg_api_id', self.api_id) config.set('tg_api_hash', self.api_hash) config.set('tg_phone', self.phone) # Start handlers from modules.autoreply import register_handlers register_handlers(self.client) return {"status": "success", "user": user.to_dict() if hasattr(user, 'to_dict') else str(user)} def get_me(self): """Returns details about the authenticated user profile.""" if not self.is_connected(): return None try: return self.run_coroutine(self._get_me_async()) except Exception as e: logger.error(f"Failed to fetch profile info: {e}") return None async def _get_me_async(self): me = await self.client.get_me() if me: return { "id": me.id, "first_name": me.first_name, "last_name": me.last_name, "username": me.username, "phone": me.phone } return None def logout(self): """Logs out from Telegram, removes sessions and clears configuration.""" if self.client: try: self.run_coroutine(self._logout_async()) except Exception as e: logger.error(f"Error logging out client: {e}") # Reset config values config = ConfigManager() config.set('tg_api_id', '') config.set('tg_api_hash', '') config.set('tg_phone', '') # Delete session files for filename in os.listdir(SESSION_DIR): if filename.startswith('telegram_session'): try: os.remove(os.path.join(SESSION_DIR, filename)) except Exception as e: logger.error(f"Could not delete session file {filename}: {e}") async def _logout_async(self): if await self.client.is_user_authorized(): await self.client.log_out() await self.client.disconnect() self.client = None