import asyncio import logging import random import re import pyrogram from pyrogram.raw import functions, types from database import db_session, LiveConfig, log_system_event logger = logging.getLogger("LiveJoiner") def sanitize_chat_target(chat_target: str) -> str: """Clean Telegram channel link / username / ID into valid chat identifier""" if not chat_target: return "" chat_target = chat_target.strip() if "?livestream=" in chat_target: chat_target = chat_target.split("?")[0] if "t.me/" in chat_target: parts = chat_target.split("t.me/") chat_target = parts[1].strip("/") if not chat_target.startswith("@") and not chat_target.startswith("-100") and not chat_target.lstrip("-").isdigit(): chat_target = f"@{chat_target}" return chat_target def parse_multiple_chat_targets(chat_target_raw: str) -> list: """Parse comma, space, or newline separated multi-channel targets""" if not chat_target_raw: return [] raw_list = re.split(r'[,\n\s]+', chat_target_raw.strip()) clean_list = [] for item in raw_list: if item.strip(): clean = sanitize_chat_target(item.strip()) if clean and clean not in clean_list: clean_list.append(clean) return clean_list class TGVoiceCallManager: def __init__(self, user_clients_dict: dict): self.user_clients = user_clients_dict # Dict of account_id -> Pyrogram Client self.pytgcalls_instances = {} # account_id -> PyTgCalls instance self.joined_accounts = set() # Set of (acc_id, chat_identifier) self.joined_call_ids = {} # (acc_id, chat_identifier) -> Telegram call ID self.current_chats = set() self.silence_feeders = {} # (acc_id, chat_identifier) -> asyncio.Task def _stop_silence_feeder(self, pair): task = self.silence_feeders.pop(pair, None) if task and not task.done(): task.cancel() async def _feed_external_silence(self, pair, py_call, chat_id): """Keep an EXTERNAL audio source alive without spawning ffmpeg. PyTgCalls consumes signed 16-bit little-endian PCM. At 48 kHz mono, one 20 ms frame is 960 samples / 1920 bytes. This path is deliberately process-free so it works on shared cPanel hosts where Boost.Process cannot launch ffmpeg/ffprobe. """ from pytgcalls.types import Device silent_frame = b"\x00" * 1920 loop = asyncio.get_running_loop() next_frame_at = loop.time() try: while True: try: await py_call.send_frame( chat_id, Device.MICROPHONE, silent_frame, ) except asyncio.CancelledError: raise except Exception as exc: # Negotiation can briefly report ConnectionNotFound just # after play(). Retry for the next frame; the connection # verifier below remains the authority on join success. logger.debug(f"External silence feeder notice for {pair}: {exc}") next_frame_at += 0.020 await asyncio.sleep(max(0.001, next_frame_at - loop.time())) except asyncio.CancelledError: pass async def _play_external_stream(self, pair, py_call, chat_id, audio_stream): """Start frame production before native EXTERNAL-media negotiation.""" self._stop_silence_feeder(pair) self.silence_feeders[pair] = asyncio.create_task( self._feed_external_silence(pair, py_call, chat_id), name=f"tg-silence-{pair[0]}-{chat_id}", ) try: operation = ( py_call.play(chat_id, audio_stream) if hasattr(py_call, "play") else py_call.join_group_call(chat_id, audio_stream) ) await asyncio.wait_for(operation, timeout=30) except Exception: self._stop_silence_feeder(pair) raise async def _native_call_is_active(self, py_call, chat_id, wait_seconds=0): """Return True only when ntgcalls reports a real media connection. ``play()`` schedules a connection and can return before WebRTC has finished negotiating. Treating that return value as success produced false 10/10 reports while Telegram showed zero participants. """ deadline = asyncio.get_running_loop().time() + max(0, wait_seconds) while True: try: binding = getattr(py_call, "_binding", None) calls_method = getattr(binding, "calls", None) if not callable(calls_method): return None active_calls = await calls_method() if isinstance(active_calls, dict): active_ids = active_calls.keys() else: active_ids = active_calls or [] if chat_id in active_ids or str(chat_id) in {str(item) for item in active_ids}: return True except Exception as exc: logger.warning(f"Could not verify native call {chat_id}: {exc}") return None if asyncio.get_running_loop().time() >= deadline: return False await asyncio.sleep(1) async def _telegram_reports_participant(self, client, call_obj, wait_seconds=0): """Ask Telegram itself whether this account is in the group call.""" deadline = asyncio.get_running_loop().time() + max(0, wait_seconds) input_call = types.InputGroupCall( id=call_obj.id, access_hash=call_obj.access_hash ) me_peer = await client.resolve_peer("me") while True: try: result = await client.invoke( functions.phone.GetGroupParticipants( call=input_call, ids=[me_peer], sources=[], offset="", limit=10 ) ) if getattr(result, "participants", None): return True except Exception as exc: logger.warning(f"Telegram participant verification notice: {exc}") if asyncio.get_running_loop().time() >= deadline: return False await asyncio.sleep(1) async def start_all(self): try: from pytgcalls import PyTgCalls log_system_event( "PyTgCalls initialization: module available; " "media=direct external PCM (no ffmpeg/ffprobe subprocess)" ) for acc_id, client in self.user_clients.items(): if acc_id not in self.pytgcalls_instances: py_call = PyTgCalls(client) try: await py_call.start() except Exception as ex: logger.warning(f"PyTgCalls start notice for #{acc_id}: {ex}") log_system_event( f"PyTgCalls start failed for Account #{acc_id}: " f"{type(ex).__name__} - {ex}", level="ERROR" ) self.pytgcalls_instances[acc_id] = py_call logger.info(f"PyTgCalls ready for {len(self.pytgcalls_instances)} account sessions.") log_system_event( f"PyTgCalls instances ready: {len(self.pytgcalls_instances)}/{len(self.user_clients)}" ) except Exception as e: logger.warning(f"PyTgCalls init notice: {e}") log_system_event( f"PyTgCalls initialization failed: {type(e).__name__} - {e}", level="ERROR" ) async def debug_live_join(self, chat_identifier_raw: str): """Run real-time deep debugging of Live Voice Chat join for Account #1 across targets""" targets = parse_multiple_chat_targets(chat_identifier_raw) out = [] out.append(f"๐Ÿ” === TELEGRAM MULTI-CHANNEL LIVE DEBUGGER ===") out.append(f"๐ŸŽฏ Target Input Channels: {targets}") if not self.user_clients: out.append("โŒ ERROR: No userbot accounts loaded in memory!") return "\n".join(out) out.append(f"โœ… Total Accounts Loaded in Memory: {len(self.user_clients)}") acc_id, client = next(iter(self.user_clients.items())) out.append(f"\n๐Ÿงช Testing Sample Account ID #{acc_id}...") # 1. Connection check try: me = await client.get_me() out.append(f" โœ… Account Connected: {me.first_name} (@{me.username or 'N/A'}, Phone: +{me.phone_number})") except Exception as e: out.append(f" โŒ Account Connection Failed: {e}") return "\n".join(out) for chat_identifier in targets: out.append(f"\n๐Ÿ“ก Testing Target Stream: {chat_identifier}") # 2. Get Chat check chat = None try: join_result = await client.join_chat(chat_identifier) # Kurigram 2.2+ may return ChatJoinResultSuccess instead of a # Chat object. Resolve the actual Chat before reading id/title. chat = join_result if hasattr(join_result, "id") else await client.get_chat(chat_identifier) out.append(f" โœ… Joined Chat: {chat.title} (ID: {chat.id}, Type: {chat.type})") except Exception as j_err: out.append(f" โš ๏ธ Join Chat Notice: {j_err}") try: chat = await client.get_chat(chat_identifier) out.append(f" โœ… Get Chat OK: {chat.title} (ID: {chat.id}, Type: {chat.type})") except Exception as g_err: out.append(f" โŒ Get Chat Failed: {g_err}") continue # 3. Call Object Resolution (Channel & Group) call_obj = None call_type = "" try: peer = await client.resolve_peer(chat.id) full = await client.invoke(functions.channels.GetFullChannel(channel=peer)) call_obj = getattr(full.full_chat, "call", None) call_type = "Channel/Supergroup" except Exception as c_err: out.append(f" โ„น๏ธ Channel GetFullChannel notice: {c_err}") try: full = await client.invoke(functions.messages.GetFullChat(chat_id=chat.id)) call_obj = getattr(full.full_chat, "call", None) call_type = "Basic Group" except Exception as g_err: out.append(f" โŒ Group GetFullChat error: {g_err}") if not call_obj: out.append(f" โš ๏ธ NO ACTIVE VOICE CHAT DETECTED in {chat.title}!") out.append(" ๐Ÿ’ก Make sure Voice Chat / Live Stream is STARTED in Telegram app before clicking Join.") continue out.append(f" โœ… Live Call Object Found via {call_type}! (Call ID: {call_obj.id}, AccessHash: {call_obj.access_hash})") # 4. Join Group Call Test with Unique SSRC Payload try: input_call = types.InputGroupCall(id=call_obj.id, access_hash=call_obj.access_hash) ssrc = random.randint(1000000, 99999999) params_json = f'{{"ssrc": {ssrc}, "muted": true, "video_stopped": true}}' res = await client.invoke( functions.phone.JoinGroupCall( call=input_call, join_as=await client.resolve_peer("me"), params=types.DataJSON(data=params_json), muted=True ) ) out.append(f" ๐ŸŽ‰ SUCCESS! Account #{acc_id} Joined Live Voice Chat in {chat.title}! (Response: {type(res).__name__})") except Exception as raw_err: out.append(f" โŒ JoinGroupCall MTProto Error: {raw_err}") return "\n".join(out) async def run_live_diagnostics(self, chat_identifier: str): """Run complete step-by-step diagnostic test for Live Voice Chat joining""" return True, await self.debug_live_join(chat_identifier) async def join_single_account(self, acc_id, client, chat_identifier, muted, stagger_delay=0.0): """Join Voice Chat for a single userbot account with FloodWait handling & duplicate skipping""" if stagger_delay > 0: await asyncio.sleep(stagger_delay) try: # Step 1: Ensure account has joined the target channel/group first try: join_result = await client.join_chat(chat_identifier) # New Kurigram returns ChatJoinResultSuccess for a successful # join. It is not a Chat and intentionally has no `.id`. chat = join_result if hasattr(join_result, "id") else await client.get_chat(chat_identifier) except Exception: try: chat = await client.get_chat(chat_identifier) except Exception as g_err: log_system_event(f"Account #{acc_id} could not get chat {chat_identifier}: {g_err}", level="WARNING") return False, str(g_err) chat_id = chat.id # Resolve the currently active Telegram call before touching the # media connection. A Live restart creates a new call ID. call_obj = None try: peer = await client.resolve_peer(chat_id) full_chat = await client.invoke(functions.channels.GetFullChannel(channel=peer)) call_obj = getattr(full_chat.full_chat, "call", None) except Exception: try: full_chat = await client.invoke(functions.messages.GetFullChat(chat_id=chat_id)) call_obj = getattr(full_chat.full_chat, "call", None) except Exception as chat_err: logger.warning(f"Account #{acc_id} call detection notice: {chat_err}") pair = (acc_id, chat_identifier) if not call_obj: self._stop_silence_feeder(pair) self.joined_accounts.discard(pair) self.joined_call_ids.pop(pair, None) return False, "No active Live Voice Chat running" # Heartbeat and cron may both check repeatedly. Never call play() # again for the same account and same active Telegram call. if self.joined_call_ids.get(pair) == call_obj.id: py_call = self.pytgcalls_instances.get(acc_id) still_connected = None # Current PyTgCalls exposes active native calls through its # binding. Detect a finished/dropped media connection instead # of trusting stale in-memory state forever. if py_call and hasattr(py_call, "_binding") and hasattr(py_call._binding, "calls"): try: still_connected = await self._native_call_is_active(py_call, chat_id) except Exception: still_connected = None if still_connected is True: server_confirmed = await self._telegram_reports_participant( client, call_obj, wait_seconds=3 ) if server_confirmed: return True, f"Account #{acc_id} confirmed by Telegram in current Live" still_connected = False # With no native state API, do not repeatedly call play() for a # cached connection. This compatibility path is only used by # older PyTgCalls releases. if still_connected is None and pair in self.joined_accounts: server_confirmed = await self._telegram_reports_participant( client, call_obj, wait_seconds=3 ) if server_confirmed: return True, f"Account #{acc_id} confirmed by Telegram (legacy media state)" self.joined_accounts.discard(pair) self.joined_call_ids.pop(pair, None) self._stop_silence_feeder(pair) # Step 2: Try PyTgCalls joining joined = False if acc_id in self.pytgcalls_instances: try: from pytgcalls.types import ExternalMedia, MediaStream from pytgcalls.types.raw import AudioParameters py_call = self.pytgcalls_instances[acc_id] # Use PyTgCalls' native EXTERNAL source. File/SHELL sources # invoke ffmpeg through Boost.Process and fail on this # shared cPanel host with default_launcher ENOENT. audio_stream = MediaStream( ExternalMedia.AUDIO, audio_parameters=AudioParameters(48000, 1), video_flags=MediaStream.Flags.IGNORE, ) # EXTERNAL media has no producer of its own. Start the PCM # feeder before play(): play() may wait for the first # external frames during native negotiation. Starting it # afterwards creates a circular wait with no exception. await self._play_external_stream( pair, py_call, chat_id, audio_stream ) if muted: try: if hasattr(py_call, "mute"): await py_call.mute(chat_id) else: await py_call.mute_stream(chat_id) except Exception: pass verified = await self._native_call_is_active(py_call, chat_id, wait_seconds=10) if verified is False: raise RuntimeError("PyTgCalls returned but no native media connection appeared") server_confirmed = await self._telegram_reports_participant( client, call_obj, wait_seconds=12 ) if not server_confirmed: raise RuntimeError( "Telegram server did not list this account as a Live participant" ) joined = True except Exception as py_err: self._stop_silence_feeder(pair) logger.warning(f"Account #{acc_id} PyTgCalls join notice: {py_err}") log_system_event( f"Account #{acc_id} voice connection failed for {chat_identifier}: {py_err}", level="ERROR" ) # A raw JoinGroupCall request without a real WebRTC transport only # creates a short-lived/ghost participant. Never count that as a # successful Live connection. if not joined: if acc_id not in self.pytgcalls_instances: return False, ( f"Account #{acc_id}: PyTgCalls instance is unavailable " f"({len(self.pytgcalls_instances)}/{len(self.user_clients)} initialized)" ) return False, "Native PyTgCalls media connection was not established" if joined: self.joined_accounts.add(pair) self.joined_call_ids[pair] = call_obj.id self.current_chats.add(chat_identifier) log_system_event(f"๐Ÿš€ Account #{acc_id} joined TG Live Voice Chat in {chat_identifier}") return True, f"Account #{acc_id} joined {chat_identifier}" return False, "Failed to join live" except pyrogram.errors.FloodWait as fw: log_system_event(f"โš ๏ธ Account #{acc_id} Telegram FloodWait ({fw.value}s) for {chat_identifier}", level="WARNING") return False, f"FloodWait {fw.value}s" except Exception as e: logger.error(f"Account #{acc_id} live join failed for {chat_identifier}: {e}") log_system_event(f"Account #{acc_id} live join error for {chat_identifier}: {e}", level="ERROR") return False, str(e) async def join_live(self, chat_target_raw: str, muted: bool = True): """Join TG Live Voice Chat for all accounts across MULTIPLE target channels smoothly with staggered delays""" targets = parse_multiple_chat_targets(chat_target_raw) if not targets: log_system_event("Live Join Error: Target chat username/ID not specified", level="ERROR") return False, "Target chat not specified" if not self.user_clients: log_system_event("Live Join Error: No active accounts loaded", level="ERROR") cfg = db_session.query(LiveConfig).first() if cfg: cfg.last_status = "Failed: No active accounts in memory/database" db_session.commit() return False, "No active userbot accounts loaded" results = [] task_pairs = [] for target in targets: for acc_id, client in self.user_clients.items(): task_pairs.append((acc_id, target)) # Shared cPanel enforces a strict native-thread quota. Starting # five WebRTC negotiations together raises "can't start new # thread". Negotiate one account fully before starting next. try: result = await self.join_single_account( acc_id, client, target, muted ) except Exception as exc: result = exc results.append(result) await asyncio.sleep(0.25) if not task_pairs: return True, f"All {len(self.user_clients)} accounts already joined in Live Voice Chat ({targets})" log_system_event( f"Completed direct-PCM serial Live Join check for targets {targets} " f"across {len(task_pairs)} connections" ) # Reconcile cached joins with the real result of this check. Without # this, an ended Live remained falsely reported as 10/10 connected. success_count = 0 failure_reasons = [] for pair, result in zip(task_pairs, results): succeeded = ( not isinstance(result, Exception) and isinstance(result, tuple) and bool(result[0]) ) if succeeded: self.joined_accounts.add(pair) success_count += 1 else: self.joined_accounts.discard(pair) self.joined_call_ids.pop(pair, None) self._stop_silence_feeder(pair) if isinstance(result, Exception): failure_reasons.append(str(result)) elif isinstance(result, tuple) and len(result) > 1: failure_reasons.append(str(result[1])) active_targets = {target for _, target in self.joined_accounts} self.current_chats.intersection_update(active_targets) total_attempted = len(targets) * len(self.user_clients) # Update DB State cfg = db_session.query(LiveConfig).first() if cfg: cfg.is_joined = (success_count > 0) if success_count: cfg.last_status = f"Joined {success_count}/{total_attempted} stream connections across {len(targets)} channels." else: reason = failure_reasons[0] if failure_reasons else "No active Live Voice Chat detected" cfg.last_status = f"Waiting for Live: {reason}" db_session.commit() if success_count: status_msg = f"๐ŸŽ‰ Successfully connected accounts into Live Voice Chats! ({success_count}/{total_attempted} active connections across {len(targets)} channels)" log_system_event(status_msg) else: reason = failure_reasons[0] if failure_reasons else "No active Live Voice Chat detected" status_msg = f"โณ Auto-join waiting: {reason}" log_system_event(status_msg) return (success_count > 0), status_msg async def leave_live(self): """Leave TG Live Voice Chat for all accounts across all active channels""" if not self.current_chats and not self.joined_accounts: return True, "Not currently in any live voice chat" tasks = [] for (acc_id, chat_identifier) in list(self.joined_accounts): if acc_id in self.user_clients: client = self.user_clients[acc_id] tasks.append(self._leave_single_account(acc_id, client, chat_identifier)) if tasks: await asyncio.gather(*tasks, return_exceptions=True) self.joined_accounts.clear() self.joined_call_ids.clear() self.current_chats.clear() for pair in list(self.silence_feeders): self._stop_silence_feeder(pair) cfg = db_session.query(LiveConfig).first() if cfg: cfg.is_joined = False cfg.last_status = "Disconnected" db_session.commit() log_system_event("Disconnected all accounts from TG Live Voice Chats.") return True, "Disconnected all accounts from Live Voice Chats." async def _leave_single_account(self, acc_id, client, chat_identifier): """Leave voice call for single account""" self._stop_silence_feeder((acc_id, chat_identifier)) if acc_id in self.pytgcalls_instances: try: py_call = self.pytgcalls_instances[acc_id] chat = await client.get_chat(chat_identifier) await py_call.leave_call(chat.id) except Exception: pass try: chat = await client.get_chat(chat_identifier) peer = await client.resolve_peer(chat.id) full_chat = None try: full_chat = await client.invoke(functions.channels.GetFullChannel(channel=peer)) except Exception: full_chat = await client.invoke(functions.messages.GetFullChat(chat_id=chat.id)) call_obj = getattr(full_chat.full_chat, "call", None) if call_obj: input_call = types.InputGroupCall(id=call_obj.id, access_hash=call_obj.access_hash) await client.invoke(functions.phone.LeaveGroupCall(call=input_call, source=0)) except Exception as e: logger.warning(f"Account #{acc_id} LeaveGroupCall error: {e}") async def toggle_mute(self, muted: bool): """Toggle mute/unmute state across active voice calls""" for acc_id, py_call in self.pytgcalls_instances.items(): for chat in self.current_chats: try: c_obj = await self.user_clients[acc_id].get_chat(chat) if muted: if hasattr(py_call, "mute"): await py_call.mute(c_obj.id) else: await py_call.mute_stream(c_obj.id) else: if hasattr(py_call, "unmute"): await py_call.unmute(c_obj.id) else: await py_call.unmute_stream(c_obj.id) except Exception: pass log_system_event(f"Toggled microphone state: {'Muted' if muted else 'Unmuted'}") return True, f"Audio microphone {'muted' if muted else 'unmuted'}"