diff --git "a/app.py" "b/app.py" --- "a/app.py" +++ "b/app.py" @@ -1,13 +1,22 @@ """ -Advanced Stream Bot — v3.0 "Polished Edition" -- Easy navigation: commands first, inline buttons as supplements -- Full advanced settings: every parameter configurable -- Default/restore mode -- Anti-freeze: stream error recovery + watchdog -- New message on command response (no stale edits) -- Auto stream-state change notifications -- Context-aware keyboards (streaming vs idle) -- Improved UX flows +Advanced Stream Bot — v3.3.0 +- Fixed: frames_encoded now counts decoded input frames (encoder buffering no longer causes 0) +- Fixed: live_log_lines_user cleared on stream start so stale logs don't persist +- Fixed: last_sent reset on stream start so bubble refreshes immediately +- Fixed: raw FFmpeg logs via queue-drained thread (bitrate, fps, codec init, errors) +- Fixed: log tail display increased 8→15 lines +- New: GET /poll/{chat_id} endpoint — returns pending live-view edit for proactive refresh +- Live view: status+log bubble auto-edits every 2s (logg.py pattern) +- Pause/Resume/Refresh/Close live view controls +- Force Reboot button + /reboot command — last-resort recovery +- Fixed: error field no longer shows raw log lines +- Fixed: error cleared on clean stop/abort +- Fixed: stats (uptime/frames/bytes) update correctly in live view +- Fixed: abort works instantly during reconnect sleep +- Fixed: watchdog does not mark error after intentional stop +- Fixed: stale edits never pile up in outbound queue (cap=1 per message) +- SSE /events/{chat_id} for real-time web dashboard +- Scheduler-triggered streams queue outbound notifications """ import logging @@ -23,9 +32,10 @@ import asyncio from urllib.parse import urlparse from pathlib import Path import io -import requests # for Telegram file download from fastapi import FastAPI, Request, HTTPException +from fastapi.responses import StreamingResponse +import queue import av from PIL import Image, ImageEnhance, UnidentifiedImageError from apscheduler.schedulers.background import BackgroundScheduler @@ -33,13 +43,14 @@ from apscheduler.schedulers.background import BackgroundScheduler # ────────────────────────────────────────────── # VERSION & APP # ────────────────────────────────────────────── -APP_VERSION = "3.0.0" +APP_VERSION = "3.3.0" app = FastAPI(title="Advanced Stream Bot", version=APP_VERSION) scheduler = BackgroundScheduler(timezone="UTC") +# Per-job metadata store — APScheduler Job objects are frozen; +# we cannot attach arbitrary attrs to them. Store meta here instead. +_job_store: dict = {} # job_id -> {"output_url": ..., "input_url": ...} -# Telegram Bot Token — set via environment variable -TELEGRAM_BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN", "") # ────────────────────────────────────────────── # LOGGING @@ -79,6 +90,17 @@ LOGO_POSITIONS = ["top_left", "top_right", "bottom_left", "bottom_right", "cente FFMPEG_PRESETS = ["ultrafast", "superfast", "veryfast", "faster", "fast", "medium", "slow", "slower", "veryslow"] OUTPUT_FORMATS = ["flv", "mpegts", "mp4", "hls"] +# Common resolution presets for the picker +RESOLUTION_PRESETS = [ + ("360p", "640x360"), + ("480p", "854x480"), + ("720p", "1280x720"), + ("1080p", "1920x1080"), + ("1440p", "2560x1440"), + ("4K", "3840x2160"), + ("Source","source"), +] + QUALITY_PRESETS = { "low": {"video_bitrate": "800k", "audio_bitrate": "96k", "ffmpeg_preset": "superfast", "resolution": "854x480", "fps": 30}, "medium": {"video_bitrate": "1500k", "audio_bitrate": "128k", "ffmpeg_preset": "medium", "resolution": "1280x720", "fps": 30}, @@ -121,7 +143,7 @@ DEFAULT_USER_SETTINGS = { "stop_on_error_in_playlist": True, "reconnect_on_stream_error": True, "reconnect_delay_seconds": 5, - "max_reconnect_attempts": 10, + "max_reconnect_attempts": 3, # Connection / network "open_timeout_seconds": 15, @@ -137,11 +159,17 @@ DEFAULT_USER_SETTINGS = { "logo_opacity": 0.85, "logo_margin_px": 10, + # Logging + "verbose_ffmpeg_log": False, # show raw FFmpeg/libav lines in live log + # Conversation state "current_step": None, "current_step_index": 0, "conversation_fields_list": [], "settings_editing_field": None, + + # UX mode: "send" = always new message (default), "edit" = edit-in-place + "ux_mode": "send", } DEFAULT_SESSION_RUNTIME_STATE = { @@ -165,6 +193,9 @@ DEFAULT_SESSION_RUNTIME_STATE = { "reconnect_attempt": 0, "last_frame_time": None, # for watchdog "last_notified_state": None, # for state-change notifications + "status_message_id": None, # message to auto-update in real-time + "status_chat_id": None, + "active_output_url": None, # the URL actually being streamed to right now } # ───────────��────────────────────────────────── @@ -175,6 +206,67 @@ session_locks: dict = {} # Store the main event loop for cross-thread calls _main_event_loop: asyncio.AbstractEventLoop = None +# ────────────────────────────────────────────── +# REAL-TIME EVENT QUEUE SYSTEM +# Each chat_id has a list of asyncio.Queue objects (one per SSE subscriber). +# When any state/log change happens, we push an event to all subscribers. +# ────────────────────────────────────────────── +_sse_subscribers: dict = {} # chat_id -> list[asyncio.Queue] +_sse_lock = threading.Lock() + +# Outbound message queue: background threads can enqueue messages here; +# they will be returned on the next /webhook response for that chat. +_outbound_message_queue: dict = {} # chat_id -> list[dict] +_outbound_queue_lock = threading.Lock() + + +def _get_sse_subscribers(chat_id: int) -> list: + with _sse_lock: + return list(_sse_subscribers.get(chat_id, [])) + + +def _register_sse_subscriber(chat_id: int, q: asyncio.Queue): + with _sse_lock: + _sse_subscribers.setdefault(chat_id, []).append(q) + + +def _unregister_sse_subscriber(chat_id: int, q: asyncio.Queue): + with _sse_lock: + lst = _sse_subscribers.get(chat_id, []) + try: + lst.remove(q) + except ValueError: + pass + + +def push_sse_event(chat_id: int, event: dict): + """Push a JSON event to all SSE subscribers for this chat_id (thread-safe).""" + subscribers = _get_sse_subscribers(chat_id) + if not subscribers: + return + loop = _main_event_loop + if not loop or not loop.is_running(): + return + payload = json.dumps(event) + for q in subscribers: + try: + loop.call_soon_threadsafe(q.put_nowait, payload) + except Exception: + pass + + +def enqueue_outbound_message(chat_id: int, msg: dict): + """Queue a message dict to be returned on the next webhook response for chat_id.""" + with _outbound_queue_lock: + _outbound_message_queue.setdefault(chat_id, []).append(msg) + + +def pop_outbound_messages(chat_id: int) -> list: + """Pop and return all queued outbound messages for this chat_id.""" + with _outbound_queue_lock: + msgs = _outbound_message_queue.pop(chat_id, []) + return msgs + def get_user_session(chat_id: int) -> dict: if chat_id not in user_sessions: @@ -211,54 +303,23 @@ def append_user_live_log(chat_id: int, line: str): session['live_log_lines_user'].pop(0) else: session['live_log_lines_user'].append(entry) + # Push real-time log event to SSE subscribers + push_sse_event(chat_id, {"type": "log", "line": entry}) # ────────────────────────────────────────────── # TELEGRAM API HELPERS +# (Webhook-only mode: no outbound Telegram API calls) # ────────────────────────────────────────────── -def _tg_api_call(method: str, payload: dict) -> dict: - """Make a direct outgoing Telegram API call (for push notifications from threads).""" - if not TELEGRAM_BOT_TOKEN: - logger.warning("No TELEGRAM_BOT_TOKEN set — cannot make outgoing API calls.") - return {} - try: - url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/{method}" - resp = requests.post(url, json=payload, timeout=10) - return resp.json() - except Exception as e: - logger.error(f"Outgoing API call failed ({method}): {e}") - return {} - - def push_message(chat_id: int, text: str, reply_markup=None, parse_mode="HTML"): - """Send a message proactively from a background thread.""" - payload = {"chat_id": chat_id, "text": text, "parse_mode": parse_mode} - if reply_markup: - payload["reply_markup"] = json.dumps(reply_markup) - return _tg_api_call("sendMessage", payload) - - -def download_telegram_file(file_id: str) -> bytes: - """Download a file from Telegram by file_id. Returns bytes or None.""" - if not TELEGRAM_BOT_TOKEN: - return None - try: - resp = requests.get( - f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/getFile", - params={"file_id": file_id}, timeout=10 - ) - data = resp.json() - if not data.get("ok"): - return None - file_path = data["result"]["file_path"] - file_resp = requests.get( - f"https://api.telegram.org/file/bot{TELEGRAM_BOT_TOKEN}/{file_path}", - timeout=30 - ) - return file_resp.content - except Exception as e: - logger.error(f"Failed to download Telegram file {file_id}: {e}") - return None + """ + In webhook-only mode outbound calls are not possible. + Log the notification so it appears in /logs instead. + """ + import html + plain = html.unescape(re.sub(r'<[^>]+>', '', text)) + logger.info(f"[Chat {chat_id}] [push_message suppressed] {plain[:200]}") + append_user_live_log(chat_id, f"[notification] {plain[:200]}") # ────────────────────────────────────────────── @@ -289,6 +350,265 @@ def answer_callback_query(cq_id: str, text: str = None, show_alert: bool = False return resp +# ────────────────────────────────────────────── +# LIVE VIEW SYSTEM (inspired by logg.py) +# ────────────────────────────────────────────── +# Pattern taken from logg.py: +# • Each chat has a "live_view" entry: message_id + last_sent_text + show flag +# • A background thread every UPDATE_INTERVAL seconds builds the fresh combined +# status+log text and queues an editMessageText for that message_id. +# • On next webhook response the edit is piggybacked — the bubble updates in place. +# • Pause/Resume toggle show_live_view, suppressing/resuming edits (same as logg.py). +# • Refresh forces an immediate rebuild+enqueue regardless of interval. +# • Auto-registers when user taps 📊 Status or 📋 Logs. +# • Auto-unregisters when stream reaches a terminal state. +# • SSE events are pushed alongside every edit for web dashboard clients. +# +# UPDATE_INTERVAL: how often the background thread re-edits the live bubble (seconds) +LIVE_VIEW_UPDATE_INTERVAL = 2 + +_live_views: dict = {} # chat_id -> {message_id, last_sent, show, paused_at, mode} +_live_view_lock = threading.Lock() + + +def lock_for(chat_id: int): + return session_locks.get(chat_id, threading.Lock()) + + +def _build_live_view_text(chat_id: int, mode: str = "status") -> str: + """ + Build the combined live-view message text. + mode="status" → status block + last 8 log lines + mode="logs" → last 30 log lines only (like logg.py /logs) + """ + session = get_user_session(chat_id) + state = session.get('streaming_state', 'idle') + + with _live_view_lock: + info = _live_views.get(chat_id, {}) + is_paused = not info.get("show", True) + + if mode == "logs": + logs = session.get('live_log_lines_user', []) + log_tail = "\n".join(logs[-30:]) if logs else "No logs yet." + pause_note = "\n\n⏸ Updates paused" if is_paused else "" + return ( + f"📋 Live Logs {'(paused)' if is_paused else '(updating…)'}\n" + f"
{esc(log_tail)}
" + f"{pause_note}" + ) + else: + status = compose_status_message(chat_id, include_config=False, include_logs=False) + logs = session.get('live_log_lines_user', []) + log_tail = "\n".join(logs[-15:]) if logs else "No logs yet." + pause_note = "\n\n⏸ Auto-updates paused — tap ▶️ Resume to continue" if is_paused else "" + + # Only show log tail while actively streaming + if state in ("streaming", "starting", "paused", "reconnecting"): + return ( + f"{status}\n\n" + f"📋 Recent Logs:\n
{esc(log_tail)}
" + f"{pause_note}" + ) + else: + return status + pause_note + + +def _get_live_view_keyboard(chat_id: int) -> dict: + """ + Keyboard for the live-view bubble: Pause/Resume/Refresh + stream controls. + Mirrors logg.py's create_log_keyboard() pattern. + """ + session = get_user_session(chat_id) + state = session.get('streaming_state', 'idle') + streaming = state in ("streaming", "paused", "starting", "reconnecting") + + with _live_view_lock: + info = _live_views.get(chat_id, {}) + is_paused_view = not info.get("show", True) + + rows = [] + + # Row 1: Live-view controls (always present when registered) + view_row = [] + if is_paused_view: + view_row.append({"text": "▶️ Resume Updates", "callback_data": "lv_resume"}) + else: + view_row.append({"text": "⏸ Pause Updates", "callback_data": "lv_pause"}) + view_row.append({"text": "🔄 Refresh", "callback_data": "lv_refresh"}) + view_row.append({"text": "❌ Close", "callback_data": "lv_close"}) + rows.append(view_row) + + # Row 2: Stream controls (context-aware, same as get_main_keyboard) + if streaming: + ctrl = [] + if state == "streaming": + ctrl.append({"text": "⏸ Pause Stream", "callback_data": "stream_pause"}) + elif state == "paused": + ctrl.append({"text": "▶️ Resume Stream", "callback_data": "stream_resume"}) + ctrl.append({"text": "⏹ Stop Stream", "callback_data": "stream_abort"}) + rows.append(ctrl) + if session.get('loop_count', 0) == -1: + rows.append([{"text": "⏳ Finish After This Loop", "callback_data": "stream_stop_graceful"}]) + else: + playlist_count = len(session.get('input_url_playlist', [])) + if playlist_count > 0 and session.get('output_url'): + rows.append([{"text": "🚀 Start Stream", "callback_data": "stream_start"}]) + + return {"inline_keyboard": rows} + + +def register_live_view(chat_id: int, message_id: int, mode: str = "status"): + """ + Register a message as the live-view bubble for this chat. + Replaces any existing registration (one live bubble per chat). + """ + with _live_view_lock: + _live_views[chat_id] = { + "message_id": message_id, + "last_sent": "", + "show": True, # True = update edits are enabled + "paused_at": None, + "mode": mode, # "status" or "logs" + } + with lock_for(chat_id): + get_user_session(chat_id)['status_message_id'] = message_id + logger.info(f"[Chat {chat_id}] Live view registered → msg {message_id} mode={mode}") + + +def unregister_live_view(chat_id: int): + with _live_view_lock: + _live_views.pop(chat_id, None) + logger.info(f"[Chat {chat_id}] Live view unregistered.") + + +def set_live_view_show(chat_id: int, show: bool): + """Toggle live-view updates on/off (Pause/Resume).""" + with _live_view_lock: + if chat_id in _live_views: + _live_views[chat_id]["show"] = show + _live_views[chat_id]["paused_at"] = None if show else time.strftime('%H:%M:%S') + + +def _do_live_view_edit(chat_id: int, force: bool = False): + """ + Build fresh text and enqueue an editMessageText for the live-view bubble. + Skips if paused (unless force=True) or if text hasn't changed. + IMPORTANT: Replaces any existing queued edit for this message (cap=1) + so stale edits never pile up in the outbound queue. + Also pushes SSE event (for web dashboard clients). + """ + with _live_view_lock: + info = _live_views.get(chat_id) + if not info: + return + show = info["show"] + msg_id = info["message_id"] + last_sent = info["last_sent"] + mode = info.get("mode", "status") + + if not show and not force: + return # Paused — skip edit (same as logg.py) + + new_text = _build_live_view_text(chat_id, mode) + kb = _get_live_view_keyboard(chat_id) + + # Only enqueue if content actually changed (avoids "message not modified" errors) + if new_text == last_sent and not force: + # Still push SSE even if Telegram text unchanged + _push_state_sse(chat_id) + return + + with _live_view_lock: + if chat_id in _live_views: + _live_views[chat_id]["last_sent"] = new_text + + edit_payload = edit_message_text(chat_id, msg_id, new_text, reply_markup=kb) + + # Replace (not append) any existing live-view edit in the queue so we never + # pile up stale edits that Telegram would reject or deliver out of order. + with _outbound_queue_lock: + existing = _outbound_message_queue.get(chat_id, []) + # Remove any previous edit targeting the same message_id + filtered = [m for m in existing + if not (m.get("method") == "editMessageText" + and m.get("message_id") == msg_id)] + filtered.append(edit_payload) + _outbound_message_queue[chat_id] = filtered + + # Also push SSE snapshot for web clients + _push_state_sse(chat_id) + + +def _live_view_updater_loop(): + """ + Background thread — fires every LIVE_VIEW_UPDATE_INTERVAL seconds. + For each registered live-view, calls _do_live_view_edit(). + On terminal state: does one forced final update then unregisters. + Also pushes SSE for active sessions without a live-view registered. + Inspired directly by logg.py's stream_command() update-interval pattern. + """ + while True: + time.sleep(LIVE_VIEW_UPDATE_INTERVAL) + try: + with _live_view_lock: + entries = list(_live_views.items()) + + for chat_id, info in entries: + try: + session = get_user_session(chat_id) + state = session.get('streaming_state', 'idle') + + _do_live_view_edit(chat_id) + + # Auto-unregister on terminal state after one forced final update + if state in ('idle', 'stopped', 'completed', 'error'): + _do_live_view_edit(chat_id, force=True) + unregister_live_view(chat_id) + # Push final SSE + _push_state_sse(chat_id) + + except Exception as e_inner: + logger.warning(f"Live view updater error for {chat_id}: {e_inner}") + + # Push SSE for active sessions without a live-view (scheduler streams, etc.) + try: + with _live_view_lock: + registered = set(_live_views.keys()) + for chat_id, session in list(user_sessions.items()): + state = session.get('streaming_state', 'idle') + if state in ('streaming', 'starting', 'reconnecting', 'paused') \ + and chat_id not in registered: + _push_state_sse(chat_id) + except Exception: + pass + + except Exception as e: + logger.error(f"Live view updater loop error: {e}") + + +_live_view_updater_thread = threading.Thread( + target=_live_view_updater_loop, name="LiveViewUpdater", daemon=True) +_live_view_updater_thread.start() + + +# Compatibility shim — old code that called register_status_message still works +def register_status_message(chat_id: int, message_id: int): + register_live_view(chat_id, message_id, mode="status") + + +def unregister_status_message(chat_id: int): + unregister_live_view(chat_id) + + +def pop_pending_status_edit(chat_id: int): + """ + Legacy shim kept for webhook handler compatibility. + Now a no-op — edits are enqueued directly by _do_live_view_edit(). + """ + return None + + # ────────────────────────────────────────────── # CONTEXT-AWARE KEYBOARDS # ────────────────────────────────────────────── @@ -316,6 +636,7 @@ def get_main_keyboard(session: dict): if session.get('loop_count', 0) == -1: btns.append([{"text": "⏳ Finish After This Loop", "callback_data": "stream_stop_graceful"}]) btns.append([{"text": "❔ Help", "callback_data": "show_help"}]) + btns.append([{"text": "⚡ Force Reboot", "callback_data": "force_reboot"}]) else: # Idle mode — full navigation btns = [] @@ -343,18 +664,24 @@ def get_main_keyboard(session: dict): {"text": "🔄 Reset to Defaults", "callback_data": "confirm_reset"}, {"text": "❔ Help", "callback_data": "show_help"}, ]) + # Row 5: Emergency + btns.append([ + {"text": "⚡ Force Reboot", "callback_data": "force_reboot"}, + ]) return {"inline_keyboard": btns} -def get_settings_keyboard(): +def get_settings_keyboard(session: dict = None): """Inline keyboard for settings navigation.""" + ux = (session or {}).get("ux_mode", "send") + ux_label = "💬 UX: New Message ✅" if ux == "send" else "✏️ UX: Edit In-Place ✅" return {"inline_keyboard": [ [{"text": "📡 Output URL", "callback_data": "set_output_url"}, {"text": "🎬 Input Playlist", "callback_data": "view_playlist"}], [{"text": "🎥 Video Codec", "callback_data": "set_video_codec"}, {"text": "🔊 Audio Codec", "callback_data": "set_audio_codec"}], - [{"text": "📐 Resolution", "callback_data": "set_resolution"}, + [{"text": "📐 Resolution", "callback_data": "pick_resolution"}, {"text": "🎞 FPS", "callback_data": "set_fps"}], [{"text": "📶 Video Bitrate", "callback_data": "set_video_bitrate"}, {"text": "🔉 Audio Bitrate", "callback_data": "set_audio_bitrate"}], @@ -366,7 +693,9 @@ def get_settings_keyboard(): {"text": "🔄 Auto-Reconnect", "callback_data": "toggle_reconnect"}], [{"text": "🛑 Stop on Error", "callback_data": "toggle_stop_on_error"}, {"text": "⏰ Open Timeout", "callback_data": "set_open_timeout"}], + [{"text": "📋 Verbose FFmpeg Log: " + ("✅ On" if (session or {}).get("verbose_ffmpeg_log") else "❌ Off"), "callback_data": "toggle_verbose_log"}], [{"text": "🎨 Quality Preset", "callback_data": "pick_quality_preset"}], + [{"text": ux_label, "callback_data": "toggle_ux_mode"}], [{"text": "✅ Done", "callback_data": "settings_done"}], ]} @@ -408,6 +737,22 @@ def get_preset_keyboard(): return {"inline_keyboard": rows} +def get_resolution_keyboard(): + """Inline picker for common resolutions.""" + rows = [] + row = [] + for label, val in RESOLUTION_PRESETS: + row.append({"text": label, "callback_data": "set_res_" + val}) + if len(row) == 3: + rows.append(row) + row = [] + if row: + rows.append(row) + rows.append([{"text": "✏️ Custom…", "callback_data": "set_res_custom"}]) + rows.append([{"text": "↩ Back", "callback_data": "open_settings"}]) + return {"inline_keyboard": rows} + + def get_logo_pos_keyboard(): positions = ["top_left", "top_right", "bottom_left", "bottom_right", "center"] rows = [[{"text": p.replace("_", " ").title(), "callback_data": f"set_logo_pos_{p}"}] for p in positions] @@ -486,6 +831,7 @@ def format_settings_display(session: dict) -> str: auto_rc = "✅" if session.get('reconnect_on_stream_error') else "❌" stop_err = "✅" if session.get('stop_on_error_in_playlist') else "❌" + verbose_log = "✅ On" if session.get('verbose_ffmpeg_log') else "❌ Off" lines = [ f"📡 Output URL: {esc(session.get('output_url', '—'))}", @@ -500,6 +846,7 @@ def format_settings_display(session: dict) -> str: f"🔁 Loop: {loop_disp}", f"🔄 Auto-Reconnect: {auto_rc} (delay: {session.get('reconnect_delay_seconds')}s, max: {session.get('max_reconnect_attempts')})", f"🛑 Stop on Error: {stop_err}", + f"📋 Verbose FFmpeg Log: {verbose_log}", f"⏱ Timeouts: open {session.get('open_timeout_seconds')}s read {session.get('read_timeout_seconds')}s", f"🖼 Logo: {logo_status}", ] @@ -512,7 +859,7 @@ def format_settings_display(session: dict) -> str: return "\n".join(lines) -def compose_status_message(chat_id: int, include_config: bool = False) -> str: +def compose_status_message(chat_id: int, include_config: bool = False, include_logs: bool = True) -> str: session = get_user_session(chat_id) state = session.get('streaming_state', 'idle') icon = STATE_EMOJI.get(state, "❓") @@ -533,11 +880,21 @@ def compose_status_message(chat_id: int, include_config: bool = False) -> str: loop_total = "∞" if loop == -1 else str(max(loop, 1)) reconnect_att = session.get('reconnect_attempt', 0) + frames = session.get('frames_encoded', 0) + bytes_s = session.get('bytes_sent', 0) + stats_ready = frames > 0 or bytes_s > 0 + + stats_lines = [] + if stats_ready: + stats_lines += [ + f"⏱ Uptime: {get_uptime(session.get('stream_start_time'))}", + f"🎞 Frames: {frames:,}", + f"📤 Sent: {bytes_s / (1024*1024):.2f} MB", + ] + lines += [ "", - f"⏱ Uptime: {get_uptime(session.get('stream_start_time'))}", - f"🎞 Frames: {session.get('frames_encoded', 0):,}", - f"📤 Sent: {session.get('bytes_sent', 0) / (1024*1024):.2f} MB", + *stats_lines, f"🎬 Input: {esc(cur_url)} ({idx+1}/{len(playlist)})", f"🔁 Loop: {session.get('current_loop_iteration', 0)+1}/{loop_total}", ] @@ -547,7 +904,7 @@ def compose_status_message(chat_id: int, include_config: bool = False) -> str: if include_config or state in ("idle", "stopped", "completed", "error"): lines += ["", "⚙️ Configuration:", format_settings_display(session)] - if state in ("streaming", "paused", "starting", "reconnecting"): + if include_logs and state in ("streaming", "paused", "starting", "reconnecting"): user_logs = session.get('live_log_lines_user', []) last_logs = "\n".join(user_logs[-6:]) if user_logs else "No logs yet." lines += ["", "📋 Recent Logs:", f"
{esc(last_logs)}
"] @@ -566,8 +923,47 @@ def notify_state_change(chat_id: int, new_state: str, extra_msg: str = ""): if extra_msg: text += f"\n{extra_msg}" text += "\n\n" + compose_status_message(chat_id) - # Sent via direct API call (from a thread) - push_message(chat_id, text, reply_markup=get_main_keyboard(session)) + kb = get_main_keyboard(session) + # Webhook-only: queue outbound message to be returned on next webhook response + push_message(chat_id, text, reply_markup=kb) + enqueue_outbound_message(chat_id, send_message(chat_id, text, reply_markup=kb)) + # Push SSE state event with full status snapshot + _push_state_sse(chat_id) + + +def _push_state_sse(chat_id: int): + """Build and push a full real-time state snapshot to SSE subscribers.""" + try: + session = get_user_session(chat_id) + state = session.get('streaming_state', 'idle') + uptime = 0 + if session.get('stream_start_time'): + try: + uptime = int((datetime.datetime.now(datetime.timezone.utc) - session['stream_start_time']).total_seconds()) + except Exception: + pass + # Build keyboard state for client-side button rendering + kb = get_main_keyboard(session) + push_sse_event(chat_id, { + "type": "state", + "state": state, + "icon": STATE_EMOJI.get(state, "❓"), + "frames_encoded": session.get('frames_encoded', 0), + "bytes_sent": session.get('bytes_sent', 0), + "uptime_seconds": uptime, + "uptime_str": get_uptime(session.get('stream_start_time')), + "reconnect_attempt": session.get('reconnect_attempt', 0), + "error": session.get('error_notification_user', ''), + "active_output_url": session.get('active_output_url', ''), + "playlist_index": session.get('current_playlist_index', 0), + "playlist_count": len(session.get('input_url_playlist', [])), + "loop_iteration": session.get('current_loop_iteration', 0), + "status_text": compose_status_message(chat_id, include_config=False), + "keyboard": kb, + "ts": datetime.datetime.now(datetime.timezone.utc).isoformat(), + }) + except Exception as e: + logger.warning(f"_push_state_sse error for {chat_id}: {e}") def update_streaming_state(chat_id: int, new_state: str, lock, session: dict, extra_msg: str = ""): @@ -583,6 +979,9 @@ def update_streaming_state(chat_id: int, new_state: str, lock, session: dict, ex args=(chat_id, new_state, extra_msg), daemon=True ).start() + else: + # Still push SSE stats even if state didn't change + threading.Thread(target=_push_state_sse, args=(chat_id,), daemon=True).start() # ────────────────────────────────────────────── @@ -599,8 +998,9 @@ def watchdog_thread_target(chat_id: int): time.sleep(5) with lock: state = session.get('streaming_state') - if state not in ("streaming", "paused", "reconnecting"): - break # Stream ended, watchdog exits + # Exit watchdog if stream is in any terminal or stopping state + if state not in ("streaming", "paused", "reconnecting", "starting"): + break last_frame = session.get('last_frame_time') thread_ref = session.get('stream_thread_ref') @@ -609,7 +1009,8 @@ def watchdog_thread_target(chat_id: int): if thread_ref and not thread_ref.is_alive(): with lock: cur_state = session.get('streaming_state') - if cur_state in ("streaming", "paused", "reconnecting"): + # Only mark error if not already in a clean terminal/stopping state + if cur_state in ("streaming", "paused", "reconnecting", "starting"): logger.error(f"[Chat {chat_id}] Watchdog: stream thread died unexpectedly!") append_user_live_log(chat_id, "⚠️ Stream thread died unexpectedly. Marking as error.") with lock: @@ -623,12 +1024,12 @@ def watchdog_thread_target(chat_id: int): stall_seconds = (datetime.datetime.now() - last_frame).total_seconds() if stall_seconds > STALL_THRESHOLD: append_user_live_log(chat_id, f"⚠️ Watchdog: No frames for {stall_seconds:.0f}s — possible freeze!") - push_message( + enqueue_outbound_message(chat_id, send_message( chat_id, f"⚠️ Stream may be frozen!\nNo frames for {stall_seconds:.0f}s.\n" f"Use /abort to stop or wait for auto-reconnect.", reply_markup=get_main_keyboard(session) - ) + )) logger.info(f"[Chat {chat_id}] Watchdog exiting.") @@ -658,6 +1059,52 @@ def stream_engine_thread_target(chat_id: int): active_output_container = None + # ── Raw FFmpeg/libav log capture via queue ──────────────────────────────── + # PyAV's av.logging.set_log_callback installs a C-level callback. We route + # every libav message into a thread-safe queue and drain it in a background + # thread so we never call Python from a C signal context. + _fflog_queue: queue.Queue = queue.Queue(maxsize=500) + _fflog_stop = threading.Event() + + def _fflog_drain(): + _LEVELS = {0:"PANIC",8:"FATAL",16:"ERROR",24:"WARN",32:"INFO",40:"VERBOSE"} + _NOISE = ("Past duration", "deprecated pixel format", "DTS", "PTS", "non monotonous") + _last = "" + while not _fflog_stop.is_set() or not _fflog_queue.empty(): + try: + level, msg = _fflog_queue.get(timeout=0.3) + except queue.Empty: + continue + msg = msg.strip() + if not msg or msg == _last: + continue + # Verbose mode off → only pass warnings/errors (level <= 24) + verbose_on = session.get('verbose_ffmpeg_log', False) + if not verbose_on and level > 24: + continue + # Skip high-volume noise regardless of verbose setting + if any(s in msg for s in _NOISE): + continue + _last = msg + lvl = _LEVELS.get(level, f"L{level}") + prefix = f"[ffmpeg/{lvl}]" if level <= 24 else "[ffmpeg]" + append_user_live_log(chat_id, f"{prefix} {msg}") + + def _fflog_callback(level, message): + try: + _fflog_queue.put_nowait((level, message)) + except queue.Full: + pass + + _fflog_thread = threading.Thread(target=_fflog_drain, name=f"FFlogDrain-{chat_id}", daemon=True) + _fflog_thread.start() + try: + av.logging.set_log_level(av.logging.VERBOSE) + av.logging.set_log_callback(_fflog_callback) + except Exception as _e_avlog: + append_user_live_log(chat_id, f"[warn] av log callback unavailable: {_e_avlog}") + # ───────────────────────────────────────────────────────────────────────── + try: with lock: session['streaming_state'] = "starting" @@ -670,11 +1117,16 @@ def stream_engine_thread_target(chat_id: int): session['stop_gracefully_flag'] = False session['reconnect_attempt'] = 0 session['last_frame_time'] = None + session['live_log_lines_user'] = [] # clear stale logs from previous run session['pyav_objects'] = { "input_container": None, "output_container": None, "video_out_stream": None, "audio_out_stream": None, "logo_image_pil": None, } + # Also reset last_sent so the live bubble always refreshes immediately + with _live_view_lock: + if chat_id in _live_views: + _live_views[chat_id]["last_sent"] = "" append_user_live_log(chat_id, f"Stream engine started → {session['output_url']}") @@ -682,9 +1134,15 @@ def stream_engine_thread_target(chat_id: int): notify_state_change(chat_id, "starting") # --- Open output container --- - output_url = session['output_url'] - output_format = session.get('output_format', 'flv') - open_timeout = session.get('open_timeout_seconds', 15) + # Always read fresh from session so any /set output_url change takes effect. + with lock: + output_url = session['output_url'] + output_format = session.get('output_format', 'flv') + open_timeout = session.get('open_timeout_seconds', 15) + # Also store the active_output_url in session for status display + with lock: + session['active_output_url'] = output_url + append_user_live_log(chat_id, f"Opening output: {output_url} [{output_format}]") try: active_output_container = av.open( output_url, mode='w', format=output_format, @@ -731,6 +1189,9 @@ def stream_engine_thread_target(chat_id: int): if session.get('stop_gracefully_flag') or session['streaming_state'] == "stopping": break current_loop_iter = session['current_loop_iteration'] + # Re-read total_loops and playlist each iteration so live changes are picked up + total_loops = session.get('loop_count', 0) + playlist = list(session.get('input_url_playlist', [])) if total_loops != -1 and current_loop_iter >= max(total_loops, 1): append_user_live_log(chat_id, f"Completed all {max(total_loops, 1)} loop(s).") @@ -756,7 +1217,7 @@ def stream_engine_thread_target(chat_id: int): _input_container = None reconnect_on_error = session.get('reconnect_on_stream_error', True) reconnect_delay = session.get('reconnect_delay_seconds', 5) - max_reconnects = session.get('max_reconnect_attempts', 10) + max_reconnects = session.get('max_reconnect_attempts', 3) read_timeout = session.get('read_timeout_seconds', 30) try: @@ -882,12 +1343,22 @@ def stream_engine_thread_target(chat_id: int): except Exception as e_logo: pass + # Count decoded input frames immediately. + # The encoder buffers many frames before emitting + # the first output packet, so counting only encoded + # output kept stats stuck at 0 for a long time. + with lock: + session['frames_encoded'] += 1 + session['last_frame_time'] = datetime.datetime.now() + _fc = session['frames_encoded'] + # Push live stats every 30 decoded frames (~1s at 30fps) + if _fc % 30 == 0: + threading.Thread(target=_push_state_sse, args=(chat_id,), daemon=True).start() + for out_pkt in active_video_out_stream.encode(frame): active_output_container.mux(out_pkt) with lock: - session['frames_encoded'] += 1 session['bytes_sent'] += out_pkt.size - session['last_frame_time'] = datetime.datetime.now() elif active_audio_out_stream and packet.stream.type == 'audio' and in_a_streams and packet.stream.index == in_a_streams[0].index: for frame in packet.decode(): @@ -902,16 +1373,19 @@ def stream_engine_thread_target(chat_id: int): except Exception as e_input: tb = traceback.format_exc() - err_msg = f"{e_input}" + err_msg = str(e_input) + # Strip any log-format prefix that might have leaked in + if "] " in err_msg and "[" in err_msg[:40]: + err_msg = err_msg.split("] ", 1)[-1] if "] " in err_msg else err_msg append_user_live_log(chat_id, f"Error on input {current_input_url}: {err_msg}") with lock: - session['error_notification_user'] = err_msg + session['error_notification_user'] = err_msg[:300] with lock: should_stop_now = session['streaming_state'] == "stopping" reconnect_on = session.get('reconnect_on_stream_error', True) cur_attempt = session.get('reconnect_attempt', 0) - max_att = session.get('max_reconnect_attempts', 10) + max_att = session.get('max_reconnect_attempts', 3) stop_on_err = session.get('stop_on_error_in_playlist', True) if should_stop_now: @@ -921,7 +1395,8 @@ def stream_engine_thread_target(chat_id: int): with lock: session['reconnect_attempt'] = cur_attempt + 1 old_st = session['streaming_state'] - session['streaming_state'] = "reconnecting" + if session['streaming_state'] not in ('stopping',): + session['streaming_state'] = "reconnecting" append_user_live_log(chat_id, f"Reconnecting in {reconnect_delay}s (attempt {cur_attempt+1}/{max_att})…") if old_st != "reconnecting": notify_state_change(chat_id, "reconnecting", f"Attempt {cur_attempt+1}/{max_att}. Error: {esc(err_msg)}") @@ -929,7 +1404,15 @@ def stream_engine_thread_target(chat_id: int): if _input_container: try: _input_container.close() except Exception: pass - time.sleep(reconnect_delay) + # Sleep in small increments so abort flag is checked promptly + for _ in range(int(reconnect_delay * 5)): + with lock: + if session['streaming_state'] == 'stopping': + break + time.sleep(0.2) + with lock: + if session['streaming_state'] == 'stopping': + break # Retry same URL (re-enter inner loop iteration) continue elif stop_on_err: @@ -972,6 +1455,13 @@ def stream_engine_thread_target(chat_id: int): notify_state_change(chat_id, "error", f"Fatal error: {esc(str(e_fatal))}") finally: + # Shut down raw ffmpeg log capture + try: + av.logging.restore_default_callback() + except Exception: + pass + _fflog_stop.set() + append_user_live_log(chat_id, "Finalizing stream…") # Flush encoders @@ -1003,15 +1493,18 @@ def stream_engine_thread_target(chat_id: int): cur = session['streaming_state'] if cur == "stopping": final_state = "stopped" + session['error_notification_user'] = "" # clear error on clean stop elif cur == "error": final_state = "error" elif session.get('error_notification_user'): final_state = "error" else: final_state = "completed" + session['error_notification_user'] = "" session['streaming_state'] = final_state session['stream_thread_ref'] = None session['watchdog_thread_ref'] = None + session['stop_gracefully_flag'] = False append_user_live_log(chat_id, f"Stream ended. Final state: {final_state}.") notify_state_change(chat_id, final_state, @@ -1023,27 +1516,32 @@ def stream_engine_thread_target(chat_id: int): # ────────────────────────────────────────────── # STREAM CONTROL HANDLERS # ────────────────────────────────────────────── -async def start_stream_handler(chat_id: int): +async def start_stream_handler(chat_id: int, message_id_to_edit: int = None): session = get_user_session(chat_id) lock = session_locks[chat_id] + def _reply(text, kb): + if message_id_to_edit: + return edit_message_text(chat_id, message_id_to_edit, text, reply_markup=kb) + return send_message(chat_id, text, reply_markup=kb) + with lock: state = session['streaming_state'] if state in ("streaming", "paused", "starting", "reconnecting"): - return send_message(chat_id, + return _reply( f"⚠️ Stream already active (state: {state}).\n" f"Use /abort to stop it first.", - reply_markup=get_main_keyboard(session)) + get_main_keyboard(session)) if not session.get('input_url_playlist'): - return send_message(chat_id, + return _reply( "📜 Playlist is empty.\nAdd at least one URL:\n" "/playlist add <url>", - reply_markup=get_main_keyboard(session)) + get_main_keyboard(session)) if not session.get('output_url') or session['output_url'] == DEFAULT_USER_SETTINGS['output_url']: - return send_message(chat_id, + return _reply( "📡 Output URL not configured.\n" "Set it with: /set output_url rtmp://your-url/key", - reply_markup=get_main_keyboard(session)) + get_main_keyboard(session)) session['streaming_state'] = "starting" session['error_notification_user'] = "" @@ -1059,7 +1557,7 @@ async def start_stream_handler(chat_id: int): await asyncio.sleep(0.3) t_watch.start() - return send_message(chat_id, + return _reply( "🔄 Stream starting…\n\n" "I'll notify you when it's live.\n\n" "Commands while streaming:\n" @@ -1067,10 +1565,10 @@ async def start_stream_handler(chat_id: int): " /resume — resume\n" " /abort — stop the stream\n" " /status — current status", - reply_markup=get_main_keyboard(session)) + get_main_keyboard(session)) -async def pause_stream_handler(chat_id: int): +async def pause_stream_handler(chat_id: int, message_id: int = None): session = get_user_session(chat_id) lock = session_locks[chat_id] with lock: @@ -1081,17 +1579,21 @@ async def pause_stream_handler(chat_id: int): ok = False st = session['streaming_state'] + def _reply(text, kb): + if message_id: + return edit_message_text(chat_id, message_id, text, reply_markup=kb) + return send_message(chat_id, text, reply_markup=kb) + if ok: append_user_live_log(chat_id, "Paused by user.") threading.Thread(target=notify_state_change, args=(chat_id, "paused"), daemon=True).start() - return send_message(chat_id, "⏸ Stream paused.\nUse /resume to continue or /abort to stop.", - reply_markup=get_main_keyboard(session)) + return _reply("⏸ Stream paused.\nUse /resume to continue or /abort to stop.", + get_main_keyboard(session)) else: - return send_message(chat_id, f"ℹ️ Cannot pause — state is {st}.", - reply_markup=get_main_keyboard(session)) + return _reply(f"ℹ️ Cannot pause — state is {st}.", get_main_keyboard(session)) -async def resume_stream_handler(chat_id: int): +async def resume_stream_handler(chat_id: int, message_id: int = None): session = get_user_session(chat_id) lock = session_locks[chat_id] with lock: @@ -1102,47 +1604,127 @@ async def resume_stream_handler(chat_id: int): ok = False st = session['streaming_state'] + def _reply(text, kb): + if message_id: + return edit_message_text(chat_id, message_id, text, reply_markup=kb) + return send_message(chat_id, text, reply_markup=kb) + if ok: append_user_live_log(chat_id, "Resumed by user.") threading.Thread(target=notify_state_change, args=(chat_id, "streaming"), daemon=True).start() - return send_message(chat_id, "▶️ Stream resumed.", reply_markup=get_main_keyboard(session)) + return _reply("▶️ Stream resumed.", get_main_keyboard(session)) else: - return send_message(chat_id, f"ℹ️ Cannot resume — state is {st}.", - reply_markup=get_main_keyboard(session)) + return _reply(f"ℹ️ Cannot resume — state is {st}.", get_main_keyboard(session)) + + +async def force_reboot_handler(chat_id: int, message_id: int = None): + """ + Force Reboot — last-resort recovery tool. + Kills any running stream thread, wipes ALL runtime state back to defaults + while preserving user settings (URLs, codec, etc.), clears all queues, + unregisters the live view, and returns the bot to a clean idle state. + """ + session = get_user_session(chat_id) + lock = session_locks[chat_id] + + append_user_live_log(chat_id, "⚡ Force Reboot initiated by user.") + + # 1. Signal any running thread to stop + with lock: + session['streaming_state'] = "stopping" + session['stop_gracefully_flag'] = True + thread_ref = session.get('stream_thread_ref') + + # 2. Wait briefly for thread to die + if thread_ref and thread_ref.is_alive(): + thread_ref.join(timeout=5.0) + + # 3. Hard-close any open PyAV containers + try: + with lock: + objs = session.get('pyav_objects', {}) + for key in ('output_container', 'input_container'): + c = objs.get(key) + if c: + try: c.close() + except Exception: pass + objs[key] = None + objs['video_out_stream'] = None + objs['audio_out_stream'] = None + objs['logo_image_pil'] = None + except Exception as e: + logger.warning(f"[Reboot] PyAV close error: {e}") + + # 4. Reset ALL runtime state — keep user settings intact + with lock: + for k, v in DEFAULT_SESSION_RUNTIME_STATE.items(): + session[k] = dict(v) if isinstance(v, dict) else (list(v) if isinstance(v, list) else v) + + # 5. Clear queues and live view + unregister_live_view(chat_id) + with _outbound_queue_lock: + _outbound_message_queue.pop(chat_id, None) + + _push_state_sse(chat_id) + append_user_live_log(chat_id, "Force Reboot complete — bot is in clean idle state.") + + reply_text = ( + "⚡ Force Reboot Complete!\n\n" + "All stream threads killed, state wiped to clean idle.\n" + "Your settings (URLs, codec, etc.) are preserved.\n\n" + + compose_status_message(chat_id, include_config=False) + ) + + def _reply(text, kb): + if message_id: + return edit_message_text(chat_id, message_id, text, reply_markup=kb) + return send_message(chat_id, text, reply_markup=kb) + return _reply(reply_text, get_main_keyboard(session)) -async def abort_stream_handler(chat_id: int): + +async def abort_stream_handler(chat_id: int, message_id: int = None): session = get_user_session(chat_id) lock = session_locks[chat_id] thread_ref = None with lock: - if session['streaming_state'] in ("streaming", "paused", "starting", "reconnecting"): + state = session['streaming_state'] + if state in ("streaming", "paused", "starting", "reconnecting", "stopping"): session['streaming_state'] = "stopping" + session['stop_gracefully_flag'] = True thread_ref = session.get('stream_thread_ref') aborted = True else: aborted = False - st = session['streaming_state'] + st = state + + def _reply(text, kb): + if message_id: + return edit_message_text(chat_id, message_id, text, reply_markup=kb) + return send_message(chat_id, text, reply_markup=kb) if not aborted: - return send_message(chat_id, f"ℹ️ No active stream (state: {st}).", - reply_markup=get_main_keyboard(session)) + return _reply(f"ℹ️ No active stream (state: {st}).", get_main_keyboard(session)) append_user_live_log(chat_id, "Abort requested by user.") if thread_ref and thread_ref.is_alive(): - thread_ref.join(timeout=8.0) + thread_ref.join(timeout=10.0) if thread_ref.is_alive(): - append_user_live_log(chat_id, "Warning: thread did not stop cleanly.") + append_user_live_log(chat_id, "Warning: thread did not stop cleanly in 10s — forcing.") with lock: - if session['streaming_state'] == "stopping": + # Force stopped state regardless of what thread set it to + if session['streaming_state'] in ("stopping", "reconnecting", "starting", "streaming", "paused"): session['streaming_state'] = "stopped" session['stream_thread_ref'] = None + session['watchdog_thread_ref'] = None + session['stop_gracefully_flag'] = False + session['error_notification_user'] = "" # clear error on clean stop - return send_message(chat_id, - "⏹ Stream aborted.\n\n" + compose_status_message(chat_id), - reply_markup=get_main_keyboard(session)) + _push_state_sse(chat_id) + return _reply("⏹ Stream stopped.\n\n" + compose_status_message(chat_id), + get_main_keyboard(session)) # ────────────────────────────────────────────── @@ -1341,122 +1923,287 @@ async def handle_logo_upload(chat_id: int, message: dict): if not file_id: return send_message(chat_id, "❌ Could not get file from message.", reply_markup=get_main_keyboard(session)) - if not TELEGRAM_BOT_TOKEN: - return send_message(chat_id, - "⚠️ TELEGRAM_BOT_TOKEN not set.\nCannot download files. Set the env variable and restart.", - reply_markup=get_main_keyboard(session)) - - logo_bytes = download_telegram_file(file_id) - if not logo_bytes: - return send_message(chat_id, "❌ Failed to download the file from Telegram. Try again.", - reply_markup=get_main_keyboard(session)) - - # Validate it's an image - try: - img = Image.open(io.BytesIO(logo_bytes)) - img.verify() - except Exception as e: - return send_message(chat_id, f"❌ Invalid image file: {esc(str(e))}", - reply_markup=get_main_keyboard(session)) - - MAX_SIZE = 5 * 1024 * 1024 # 5 MB - if len(logo_bytes) > MAX_SIZE: - return send_message(chat_id, - f"❌ File too large ({len(logo_bytes)/1024/1024:.1f} MB). Max 5 MB.", - reply_markup=get_main_keyboard(session)) - + # Webhook-only mode: Telegram sends only file_id, not the actual bytes. + # Downloading requires an outbound getFile API call which is blocked on HuggingFace Spaces. with lock: - session['logo_data_bytes'] = logo_bytes - session['logo_mime_type'] = mime_type - session['logo_original_filename'] = filename - session['logo_enabled'] = True session['current_step'] = None - - append_user_live_log(chat_id, f"Logo uploaded: {filename} ({len(logo_bytes)/1024:.1f} KB)") return send_message(chat_id, - f"✅ Logo uploaded and enabled!\n" - f"File: {esc(filename)} ({len(logo_bytes)/1024:.1f} KB)\n\n" - f"Adjust logo settings:\n" - f" /set logo_position top_right\n" - f" /set logo_scale 0.1\n" - f" /set logo_opacity 0.8\n" - f" /set logo_enabled off — to disable", + "⚠️ Logo upload is unavailable in webhook-only mode.\n\n" + "This deployment blocks outbound requests, so the bot cannot download " + "files from Telegram's servers.\n\n" + "To use a logo, host the image publicly and set it via a URL workaround, " + "or run the bot in an environment that allows outbound connections.", reply_markup=get_main_keyboard(session)) # ────────────────────────────────────────────── # SCHEDULE HANDLER # ────────────────────────────────────────────── +# Scheduling flow (conversation steps): +# sched_when → user picks "timer" or "datetime" +# sched_timer → user types "in X minutes/hours" +# sched_datetime → user types YYYY-MM-DD HH:MM:SS +# sched_name → user types a friendly job name +# sched_rtmp → user types the RTMP output URL for this job +# sched_input → user types the input URL for this job +# Then job is registered with its own output+input URLs (independent of session). + +def _list_schedules(chat_id: int) -> str: + """Build schedule list text.""" + jobs = scheduler.get_jobs() + user_jobs = [j for j in jobs if j.id.startswith(f"stream_{chat_id}_")] + lines = ["🕒 Scheduled Streams (in-memory, lost on restart)\n"] + if user_jobs: + for j in user_jobs: + rt = j.next_run_time.strftime("%Y-%m-%d %H:%M:%S UTC") if j.next_run_time else "N/A" + meta = _job_store.get(j.id, {}) + rtmp = meta.get("output_url", "?") + inp = meta.get("input_url", "?") + lines.append( + f" • {esc(j.name)}\n" + f" ⏰ {rt}\n" + f" 📡 {esc(rtmp[:60])}\n" + f" 🎬 {esc(inp[:60])}\n" + f" Cancel: /schedule cancel {esc(j.id)}" + ) + else: + lines.append(" No scheduled streams.") + lines += [ + "", + "➕ Add a schedule: use 🕒 Schedule button or /schedule new", + "❌ Cancel: /schedule cancel <job_id>", + ] + return "\n".join(lines) + + +def get_schedule_when_keyboard(): + return {"inline_keyboard": [ + [{"text": "⏱ Timer (in X minutes)", "callback_data": "sched_pick_timer"}, + {"text": "📅 Date & Time (UTC)", "callback_data": "sched_pick_datetime"}], + [{"text": "❌ Cancel", "callback_data": "sched_cancel_setup"}], + ]} + + +def get_schedule_menu_keyboard(session: dict = None): + """Keyboard shown on the schedule list view.""" + return {"inline_keyboard": [ + [{"text": "➕ New Schedule", "callback_data": "sched_new"}], + [{"text": "🔙 Back", "callback_data": "settings_done"}], + ]} + + async def handle_schedule_command(chat_id: int, text: str): session = get_user_session(chat_id) - parts = text.split(maxsplit=3) + lock = session_locks[chat_id] + parts = text.split(maxsplit=2) + sub = parts[1].lower() if len(parts) > 1 else "" - if len(parts) < 3: - jobs = scheduler.get_jobs() - user_jobs = [j for j in jobs if j.id.startswith(f"stream_{chat_id}_")] - lines = ["🕒 Scheduled Streams (in-memory, lost on restart)\n"] - if user_jobs: - for j in user_jobs: - rt = j.next_run_time.strftime('%Y-%m-%d %H:%M:%S UTC') if j.next_run_time else "N/A" - lines.append(f" • {esc(j.name)} at {rt}") - lines.append(f" Cancel: /schedule cancel {esc(j.id)}") - else: - lines.append(" No scheduled streams.") - lines += [ - "", - "Schedule a stream:", - " /schedule YYYY-MM-DD HH:MM:SS [Name]", - " Example: /schedule 2025-12-31 23:55:00 NYE", - "", - "Cancel a schedule:", - " /schedule cancel <job_id>", - ] - return send_message(chat_id, "\n".join(lines), reply_markup=get_main_keyboard(session)) + # List / entry point + if not sub or sub == "list": + return send_message(chat_id, _list_schedules(chat_id), + reply_markup=get_main_keyboard(session)) - if parts[1].lower() == "cancel": - job_id = parts[2] if len(parts) > 2 else None - if job_id: - try: - scheduler.remove_job(job_id) - return send_message(chat_id, f"✅ Schedule {esc(job_id)} cancelled.", - reply_markup=get_main_keyboard(session)) - except Exception: - return send_message(chat_id, f"❌ Job not found: {esc(job_id)}", - reply_markup=get_main_keyboard(session)) + # Start new schedule conversation + if sub == "new": + with lock: + session["current_step"] = "sched_when" + session["_sched_draft"] = {} + return send_message(chat_id, + "🕒 New Scheduled Stream\n\nWhen should it start?", + reply_markup=get_schedule_when_keyboard()) + + # Cancel a job + if sub == "cancel": + job_id = parts[2].strip() if len(parts) > 2 else None + if not job_id: + return send_message(chat_id, + "Usage: /schedule cancel <job_id>", + reply_markup=get_main_keyboard(session)) + try: + scheduler.remove_job(job_id) + _job_store.pop(job_id, None) + return send_message(chat_id, + f"✅ Cancelled: {esc(job_id)}", + reply_markup=get_main_keyboard(session)) + except Exception: + return send_message(chat_id, + f"❌ Job not found: {esc(job_id)}", + reply_markup=get_main_keyboard(session)) + + return send_message(chat_id, + "⚠️ Unknown sub-command.\nTry /schedule, /schedule new, or /schedule cancel <id>", + reply_markup=get_main_keyboard(session)) - try: - date_str, time_str = parts[1], parts[2] - job_name = parts[3] if len(parts) > 3 else f"Stream for {chat_id}" - trigger_time = datetime.datetime.strptime(f"{date_str} {time_str}", "%Y-%m-%d %H:%M:%S") - trigger_time = trigger_time.replace(tzinfo=datetime.timezone.utc) +async def handle_schedule_conversation(chat_id: int, text: str): + """Handle text input during schedule setup conversation steps.""" + session = get_user_session(chat_id) + lock = session_locks[chat_id] + step = session.get("current_step", "") + draft = session.setdefault("_sched_draft", {}) + + if step == "sched_timer": + # Parse "30", "30m", "2h", "1h30m", "90 minutes" etc. + raw = text.strip().lower().replace(" ", "") + minutes = 0 + import re as _re + hm = _re.match(r"^(?:(\d+)h)?(?:(\d+)m?)?$", raw) + plain = _re.match(r"^(\d+)$", raw) + if hm and (hm.group(1) or hm.group(2)): + minutes = int(hm.group(1) or 0) * 60 + int(hm.group(2) or 0) + elif plain: + minutes = int(plain.group(1)) + if minutes <= 0: + return send_message(chat_id, + "❌ Could not parse duration. Try 30, 2h, 1h30m\nOr /cancel.") + trigger_time = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(minutes=minutes) + draft["trigger_time"] = trigger_time + with lock: + session["current_step"] = "sched_name" + return send_message(chat_id, + f"✅ Timer set: {minutes} minute(s) from now\n" + f" → {trigger_time.strftime('%Y-%m-%d %H:%M:%S UTC')}\n\n" + f"Give this schedule a name (or type skip):") + + elif step == "sched_datetime": + raw = text.strip() + try: + # Accept both "YYYY-MM-DD HH:MM:SS" and "YYYY-MM-DD HH:MM" + fmt = "%Y-%m-%d %H:%M:%S" if len(raw) > 16 else "%Y-%m-%d %H:%M" + trigger_time = datetime.datetime.strptime(raw, fmt).replace(tzinfo=datetime.timezone.utc) + except ValueError: + return send_message(chat_id, + "❌ Invalid format. Use YYYY-MM-DD HH:MM:SS (UTC)\nOr /cancel.") if trigger_time <= datetime.datetime.now(datetime.timezone.utc): - return send_message(chat_id, "❌ Scheduled time must be in the future (UTC).") + return send_message(chat_id, + "❌ That time is in the past. Enter a future time (UTC):\nOr /cancel.") + draft["trigger_time"] = trigger_time + with lock: + session["current_step"] = "sched_name" + return send_message(chat_id, + f"✅ Time set: {trigger_time.strftime('%Y-%m-%d %H:%M:%S UTC')}\n\n" + f"Give this schedule a name (or type skip):") + + elif step == "sched_name": + name = text.strip() + draft["name"] = f"Scheduled Stream {chat_id}" if name.lower() == "skip" or not name else name + with lock: + session["current_step"] = "sched_rtmp" + return send_message(chat_id, + f"📡 Step: RTMP Output URL\n\n" + f"Enter the RTMP URL to stream to:\n" + f"rtmp://a.rtmp.youtube.com/live2/YOUR_KEY\n\n" + f"Or type use to use your current output URL ({esc(session.get('output_url', 'not set'))})\n" + f"Or /cancel.") + + elif step == "sched_rtmp": + raw = text.strip() + if raw.lower() == "use": + url = session.get("output_url", "") + else: + url = raw + if not validate_url(url): + return send_message(chat_id, + "❌ Invalid URL. Must start with rtmp/http/https/rtsp/...\nTry again or /cancel.") + draft["output_url"] = url + with lock: + session["current_step"] = "sched_input" + return send_message(chat_id, + f"🎬 Step: Input URL\n\n" + f"Enter the stream/video input URL:\n" + f"https://example.com/video.mp4\n\n" + f"Or type use to use your current playlist first item ({esc((session.get('input_url_playlist') or ['not set'])[0])})\n" + f"Or /cancel.") + + elif step == "sched_input": + raw = text.strip() + if raw.lower() == "use": + pl = session.get("input_url_playlist", []) + url = pl[0] if pl else "" + else: + url = raw + if not validate_url(url): + return send_message(chat_id, + "❌ Invalid URL. Must start with http/https/rtmp/rtsp/...\nTry again or /cancel.") + draft["input_url"] = url + + # All info collected — register the job + trigger_time = draft["trigger_time"] + job_name = draft["name"] + output_url = draft["output_url"] + input_url = draft["input_url"] job_id = f"stream_{chat_id}_{int(trigger_time.timestamp())}" - scheduler.add_job( - _scheduled_stream_runner, - 'date', - run_date=trigger_time, - args=[chat_id], - id=job_id, - name=job_name, - replace_existing=True - ) + try: + scheduler.add_job( + _scheduled_stream_runner, + "date", + run_date=trigger_time, + args=[chat_id, output_url, input_url], + id=job_id, + name=job_name, + replace_existing=True, + ) + # Store metadata in our own dict (APScheduler Jobs are frozen) + _job_store[job_id] = {"output_url": output_url, "input_url": input_url} + except Exception as e: + return send_message(chat_id, f"❌ Failed to register job: {esc(str(e))}", + reply_markup=get_main_keyboard(session)) + + mins_away = int((trigger_time - datetime.datetime.now(datetime.timezone.utc)).total_seconds() / 60) + with lock: + session["current_step"] = None + session["_sched_draft"] = {} + return send_message(chat_id, - f"✅ Stream scheduled!\n" - f" Name: {esc(job_name)}\n" - f" Time: {trigger_time.strftime('%Y-%m-%d %H:%M:%S UTC')}\n" - f" ID: {job_id}\n\n" - f"View/cancel: /schedule", + f"✅ Stream Scheduled!\n\n" + f" 📛 Name: {esc(job_name)}\n" + f" ⏰ Time: {trigger_time.strftime('%Y-%m-%d %H:%M:%S UTC')} (~{mins_away}m from now)\n" + f" 📡 Output: {esc(output_url[:60])}\n" + f" 🎬 Input: {esc(input_url[:60])}\n" + f" 🔑 ID: {job_id}\n\n" + f"View all: /schedule\nCancel: /schedule cancel {job_id}", reply_markup=get_main_keyboard(session)) - except ValueError as e: - return send_message(chat_id, - f"❌ Invalid date/time format.\nExpected: YYYY-MM-DD HH:MM:SS\nError: {esc(str(e))}") + # Fallback — unknown step + with lock: + session["current_step"] = None + return send_message(chat_id, + "⚠️ Schedule setup cancelled (unknown step).", + reply_markup=get_main_keyboard(session)) -def _scheduled_stream_runner(chat_id: int): + +def _scheduled_stream_runner(chat_id: int, output_url: str = None, input_url: str = None): + """Run by APScheduler at the scheduled time. Overrides session URLs if provided.""" logger.info(f"[Scheduler] Firing scheduled stream for chat {chat_id}") + session = get_user_session(chat_id) + lock = session_locks[chat_id] + + # Override session URLs with the per-job ones + if output_url: + with lock: + session["output_url"] = output_url + session["active_output_url"] = output_url + if input_url: + with lock: + session["input_url_playlist"] = [input_url] + + # Notify user via queued outbound message (returned on next webhook) + msg_text = ( + f"⏰ Scheduled stream is starting!\n\n" + f"📡 Output: {output_url or session.get('output_url', '—')}\n" + f"🎬 Input: {input_url or (session.get('input_url_playlist') or ['—'])[0]}\n\n" + f"Use /abort to stop or /status for live status." + ) + enqueue_outbound_message(chat_id, send_message(chat_id, msg_text, reply_markup=get_main_keyboard(session))) + # Push immediate SSE notification + push_sse_event(chat_id, { + "type": "notification", + "subtype": "scheduled_start", + "text": msg_text, + "ts": datetime.datetime.now(datetime.timezone.utc).isoformat(), + }) + loop = _main_event_loop if loop and loop.is_running(): asyncio.run_coroutine_threadsafe(start_stream_handler(chat_id), loop) @@ -1477,7 +2224,8 @@ def get_help_text() -> str: " /pause — Pause stream\n" " /resume — Resume stream\n" " /abort — Stop stream\n" - " /status — Detailed status\n\n" + " /status — Detailed status\n" + " /reboot — ⚡ Force Reboot (last resort if bot stops responding)\n\n" "━━ Configuration ━━\n" " /set — List all settings\n" " /set <field> <value> — Set a value\n" @@ -1495,12 +2243,20 @@ def get_help_text() -> str: " /set logo_scale 0.1\n" " /set logo_opacity 0.8\n\n" "━━ Schedule ━━\n" - " /schedule — List schedules\n" - " /schedule YYYY-MM-DD HH:MM:SS [Name]\n" - " /schedule cancel <id>\n\n" + " /schedule — List & manage schedules\n" + " /schedule new — Create a new schedule (conversation)\n" + " /schedule cancel <id> — Cancel a scheduled stream\n\n" + "📅 Scheduling supports:\n" + " ⏱ Timer: in X minutes/hours\n" + " 📅 Date & time: YYYY-MM-DD HH:MM:SS (UTC)\n" + " Each schedule has its own RTMP output + input URL\n\n" "━━ Diagnostics ━━\n" " /logs — Recent user logs\n" " /globallogs — System logs\n\n" + "━━ Emergency ━━\n" + " /reboot — Kills all threads, resets state to idle.\n" + " Use if the bot stops responding or gets stuck.\n" + " Your URL/settings are preserved.\n\n" "⚠️ Data is in-memory only — lost on restart." ) @@ -1578,6 +2334,9 @@ async def _handle_update_inner(update: dict): elif command == "/abort": return await abort_stream_handler(chat_id) + elif command in ("/reboot", "/forcereboot"): + return await force_reboot_handler(chat_id) + elif command == "/status": return send_message(chat_id, compose_status_message(chat_id, include_config=True), reply_markup=get_main_keyboard(session)) @@ -1589,7 +2348,7 @@ async def _handle_update_inner(update: dict): "To change a setting:\n" "/set <field> <value>\n\n" "Or use the buttons below to navigate interactively.", - reply_markup=get_settings_keyboard()) + reply_markup=get_settings_keyboard(session)) elif command == "/set": return await handle_set_command(chat_id, text) @@ -1615,10 +2374,12 @@ async def _handle_update_inner(update: dict): return await handle_schedule_command(chat_id, text) elif command == "/logs": + # Send a snapshot. User can tap 📋 Logs button for the live-updating bubble. logs = session.get('live_log_lines_user', []) - last = "\n".join(logs[-20:]) if logs else "No logs yet." + log_tail = "\n".join(logs[-30:]) if logs else "No logs yet." return send_message(chat_id, - "📋 Recent Stream Logs (last 20):\n
" + esc(last) + "
", + f"📋 Stream Logs (last 30):\n
{esc(log_tail)}
\n\n" + f"💡 Tap the 📋 Logs button below for a live-updating view.", reply_markup=get_main_keyboard(session)) elif command == "/globallogs": @@ -1655,40 +2416,119 @@ async def _handle_update_inner(update: dict): logger.info(f"[Chat {chat_id}] Callback: {data}") ack = answer_callback_query(cq["id"]) + # ── UX mode helpers ────────────────────────────────────────────────── + # ux_mode "send" → always sendMessage (new bubble, default) + # ux_mode "edit" → editMessageText in-place (replaces the button msg) + ux = session.get("ux_mode", "send") + + def reply(text, kb=None): + """Send or edit depending on ux_mode.""" + kb = kb or get_main_keyboard(session) + if ux == "edit": + return edit_message_text(chat_id, message_id, text, reply_markup=kb) + return send_message(chat_id, text, reply_markup=kb) + + def edit(text, kb=None): + """Always edit in-place (used for sub-menus that must stay in same msg).""" + return edit_message_text(chat_id, message_id, text, + reply_markup=kb or get_main_keyboard(session)) + + # ───────────────────────────────────────────────────────────────────── # Stream controls if data == "stream_start": - return [ack, await start_stream_handler(chat_id)] + mid = message_id if ux == "edit" else None + return [ack, await start_stream_handler(chat_id, message_id_to_edit=mid)] elif data == "stream_pause": - return [ack, await pause_stream_handler(chat_id)] + mid = message_id if ux == "edit" else None + return [ack, await pause_stream_handler(chat_id, message_id=mid)] elif data == "stream_resume": - return [ack, await resume_stream_handler(chat_id)] + mid = message_id if ux == "edit" else None + return [ack, await resume_stream_handler(chat_id, message_id=mid)] elif data == "stream_abort": - return [ack, await abort_stream_handler(chat_id)] + mid = message_id if ux == "edit" else None + return [ack, await abort_stream_handler(chat_id, message_id=mid)] + + elif data == "force_reboot": + return [answer_callback_query(cq["id"], "⚡ Force rebooting…"), + await force_reboot_handler(chat_id, message_id=message_id)] + elif data == "stream_status": - return [ack, send_message(chat_id, compose_status_message(chat_id, include_config=True), - reply_markup=get_main_keyboard(session))] + # Register this message as the live-view bubble (status+log mode) + # Background thread will edit it every 2s — exactly like logg.py + register_live_view(chat_id, message_id, mode="status") + live_text = _build_live_view_text(chat_id, "status") + live_kb = _get_live_view_keyboard(chat_id) + return [ack, edit(live_text, live_kb)] + + # ── Live-view controls (Pause / Resume / Refresh / Close) ──────────── + # Directly mirrors logg.py's handle_callback() for stop_logs/resume_logs/refresh_logs + elif data == "lv_pause": + set_live_view_show(chat_id, False) + # Force one immediate edit to show the "paused" state + live_text = _build_live_view_text(chat_id, _live_views.get(chat_id, {}).get("mode", "status")) + live_kb = _get_live_view_keyboard(chat_id) + with _live_view_lock: + if chat_id in _live_views: + _live_views[chat_id]["last_sent"] = live_text + return [answer_callback_query(cq["id"], "⏸ Updates paused"), + edit(live_text, live_kb)] + + elif data == "lv_resume": + set_live_view_show(chat_id, True) + _do_live_view_edit(chat_id, force=True) + live_text = _build_live_view_text(chat_id, _live_views.get(chat_id, {}).get("mode", "status")) + live_kb = _get_live_view_keyboard(chat_id) + return [answer_callback_query(cq["id"], "▶️ Updates resumed"), + edit(live_text, live_kb)] + + elif data == "lv_refresh": + # Force immediate rebuild — same as logg.py's refresh_logs + _do_live_view_edit(chat_id, force=True) + live_text = _build_live_view_text(chat_id, _live_views.get(chat_id, {}).get("mode", "status")) + live_kb = _get_live_view_keyboard(chat_id) + with _live_view_lock: + if chat_id in _live_views: + _live_views[chat_id]["last_sent"] = live_text + return [answer_callback_query(cq["id"], "🔄 Refreshed"), + edit(live_text, live_kb)] + + elif data == "lv_close": + unregister_live_view(chat_id) + return [answer_callback_query(cq["id"], "Live view closed"), + edit(compose_status_message(chat_id, include_config=False), + get_main_keyboard(session))] + elif data == "stream_stop_graceful": with lock: session['stop_gracefully_flag'] = True append_user_live_log(chat_id, "Graceful stop requested.") - return [ack, send_message(chat_id, - "⏳ Will stop after current loop finishes.", - reply_markup=get_main_keyboard(session))] + return [ack, edit("⏳ Will stop after current loop finishes.\n\n" + + compose_status_message(chat_id))] # Settings navigation elif data == "open_settings": - return [ack, send_message(chat_id, + return [ack, edit( "⚙️ Settings\n\n" + format_settings_display(session) + "\n\n" "Tap a parameter to change it, or use /set <field> <value>", - reply_markup=get_settings_keyboard())] + get_settings_keyboard(session))] + + elif data == "toggle_ux_mode": + with lock: + current_ux = session.get("ux_mode", "send") + session["ux_mode"] = "edit" if current_ux == "send" else "send" + new_ux = session["ux_mode"] + label = "New Message (sendMessage)" if new_ux == "send" else "Edit In-Place (editMessageText)" + return [ack, edit( + f"✅ UX Mode switched to: {label}\n\n" + f"{'💬 Each button press sends a new message.' if new_ux == 'send' else '✏️ Button presses edit the existing message in-place.'}\n\n" + + format_settings_display(session), + get_settings_keyboard(session))] elif data == "settings_done": - return [ack, send_message(chat_id, compose_status_message(chat_id, True), - reply_markup=get_main_keyboard(session))] + return [ack, reply(compose_status_message(chat_id, True))] elif data == "pick_quality_preset": - return [ack, send_message(chat_id, "🎨 Choose Quality Preset:", - reply_markup=get_quality_keyboard())] + return [ack, edit("🎨 Choose Quality Preset:", get_quality_keyboard())] elif data.startswith("apply_quality_"): q = data.replace("apply_quality_", "") @@ -1697,47 +2537,63 @@ async def _handle_update_inner(update: dict): session['quality_preset'] = q for k, v in QUALITY_PRESETS[q].items(): session[k] = v - return [ack, send_message(chat_id, + return [ack, edit( f"✅ Quality preset {q} applied.\n\n" + format_settings_display(session), - reply_markup=get_settings_keyboard())] + get_settings_keyboard(session))] elif data == "set_video_codec": - return [ack, send_message(chat_id, "🎥 Choose Video Codec:", - reply_markup=get_codec_keyboard("video"))] + return [ack, edit("🎥 Choose Video Codec:", get_codec_keyboard("video"))] elif data.startswith("set_vcodec_"): codec = data.replace("set_vcodec_", "") with lock: session['video_codec'] = codec - return [ack, send_message(chat_id, f"✅ Video codec: {codec}", - reply_markup=get_settings_keyboard())] + return [ack, edit(f"✅ Video codec: {codec}", get_settings_keyboard(session))] elif data == "set_audio_codec": - return [ack, send_message(chat_id, "🔊 Choose Audio Codec:", - reply_markup=get_codec_keyboard("audio"))] + return [ack, edit("🔊 Choose Audio Codec:", get_codec_keyboard("audio"))] elif data.startswith("set_acodec_"): codec = data.replace("set_acodec_", "") with lock: session['audio_codec'] = codec - return [ack, send_message(chat_id, f"✅ Audio codec: {codec}", - reply_markup=get_settings_keyboard())] + return [ack, edit(f"✅ Audio codec: {codec}", get_settings_keyboard(session))] elif data == "set_ffmpeg_preset": - return [ack, send_message(chat_id, "⚡ Choose FFmpeg Preset:", - reply_markup=get_preset_keyboard())] + return [ack, edit("⚡ Choose FFmpeg Preset:", get_preset_keyboard())] elif data.startswith("set_preset_"): preset = data.replace("set_preset_", "") with lock: session['ffmpeg_preset'] = preset - return [ack, send_message(chat_id, f"✅ Preset: {preset}", - reply_markup=get_settings_keyboard())] + return [ack, edit(f"✅ Preset: {preset}", get_settings_keyboard(session))] + + # Resolution picker + elif data == "pick_resolution": + return [ack, edit("📐 Choose Resolution:", get_resolution_keyboard())] + + elif data.startswith("set_res_"): + val = data.replace("set_res_", "") + if val == "custom": + with lock: + session["current_step"] = "editing_field" + session["settings_editing_field"] = "resolution" + return [ack, reply( + "📐 Custom Resolution\n" + "Current: " + esc(str(session.get("resolution"))) + "\n" + "Format: WIDTHxHEIGHT e.g. 1280x720 or source\n\n" + "Type the value now, or /cancel.")] + else: + with lock: + session["resolution"] = val + label = next((l for l, v in RESOLUTION_PRESETS if v == val), val) + return [ack, edit( + f"✅ Resolution set to {label} ({val})", + get_settings_keyboard(session))] # Inline-triggered field edits (ask user to type) - elif data in ("set_output_url", "set_resolution", "set_fps", "set_video_bitrate", + elif data in ("set_output_url", "set_fps", "set_video_bitrate", "set_audio_bitrate", "set_loop_count", "set_gop_size", "set_output_format", "set_reconnect_delay", "set_open_timeout"): field_map = { "set_output_url": "output_url", - "set_resolution": "resolution", "set_fps": "fps", "set_video_bitrate": "video_bitrate", "set_audio_bitrate": "audio_bitrate", @@ -1753,25 +2609,35 @@ async def _handle_update_inner(update: dict): with lock: session['current_step'] = "editing_field" session['settings_editing_field'] = field - return [ack, send_message(chat_id, + return [ack, reply( f"📝 Set {field}\n" f"Current: {esc(str(cur))}\n" f"Expected: {esc(desc)}\n\n" - f"Type the new value now, or /cancel to abort.", - reply_markup=get_main_keyboard(session))] + f"Type the new value now, or /cancel to abort.")] - elif data in ("toggle_reconnect", "toggle_stop_on_error"): + elif data in ("toggle_reconnect", "toggle_stop_on_error", "toggle_verbose_log"): field_map = { - "toggle_reconnect": "reconnect_on_stream_error", + "toggle_reconnect": "reconnect_on_stream_error", "toggle_stop_on_error": "stop_on_error_in_playlist", + "toggle_verbose_log": "verbose_ffmpeg_log", + } + default_map = { + "reconnect_on_stream_error": True, + "stop_on_error_in_playlist": True, + "verbose_ffmpeg_log": False, } field = field_map[data] with lock: - session[field] = not session.get(field, True) + session[field] = not session.get(field, default_map[field]) new_val = session[field] - return [ack, send_message(chat_id, - f"✅ {field} → {'on' if new_val else 'off'}", - reply_markup=get_settings_keyboard())] + labels = { + "reconnect_on_stream_error": "Auto-Reconnect", + "stop_on_error_in_playlist": "Stop on Error", + "verbose_ffmpeg_log": "Verbose FFmpeg Log", + } + return [ack, edit( + f"✅ {labels[field]} → {'on' if new_val else 'off'}", + get_settings_keyboard(session))] # Playlist view elif data == "view_playlist": @@ -1786,7 +2652,7 @@ async def _handle_update_inner(update: dict): else: msg = ("📜 Playlist is empty.\n" "/playlist add <url>") - return [ack, send_message(chat_id, msg, reply_markup=get_main_keyboard(session))] + return [ack, reply(msg)] # Logo elif data == "cfg_logo": @@ -1813,69 +2679,99 @@ async def _handle_update_inner(update: dict): lines.append("Change scale/opacity: /set logo_scale 0.15") else: lines.append("No logo uploaded.") - return [ack, send_message(chat_id, "\n".join(lines), - reply_markup={"inline_keyboard": logo_btns})] + return [ack, edit("\n".join(lines), {"inline_keyboard": logo_btns})] elif data == "toggle_logo": with lock: session['logo_enabled'] = not session.get('logo_enabled', False) v = session['logo_enabled'] - return [ack, send_message(chat_id, f"🖼 Logo {'enabled ✅' if v else 'disabled ❌'}.", - reply_markup=get_main_keyboard(session))] + return [ack, edit(f"🖼 Logo {'enabled ✅' if v else 'disabled ❌'}.")] elif data == "change_logo_pos": - return [ack, send_message(chat_id, "📍 Choose logo position:", - reply_markup=get_logo_pos_keyboard())] + return [ack, edit("📍 Choose logo position:", get_logo_pos_keyboard())] elif data.startswith("set_logo_pos_"): pos = data.replace("set_logo_pos_", "") with lock: session['logo_position'] = pos - return [ack, send_message(chat_id, f"✅ Logo position: {pos}", - reply_markup=get_main_keyboard(session))] + return [ack, edit(f"✅ Logo position: {pos}")] elif data == "upload_new_logo": with lock: session['current_step'] = "awaiting_logo" - return [ack, send_message(chat_id, - "🖼 Send a PNG or JPG image now. /cancel to abort.", - reply_markup=get_main_keyboard(session))] + return [ack, reply("🖼 Send a PNG or JPG image now. /cancel to abort.")] # Schedule elif data == "cfg_schedule": - return [ack, await handle_schedule_command(chat_id, "/schedule")] + return [ack, reply(_list_schedules(chat_id), + get_schedule_menu_keyboard(session))] + + elif data == "sched_new": + with lock: + session["current_step"] = "sched_when" + session["_sched_draft"] = {} + return [ack, reply( + "🕒 New Scheduled Stream\n\nWhen should it start?", + get_schedule_when_keyboard())] + + elif data == "sched_pick_timer": + with lock: + session["current_step"] = "sched_timer" + return [ack, reply( + "\u23f1 Timer Setup\n\n" + "How long from now? Type a duration:\n" + " 30 \u2192 30 minutes\n" + " 2h \u2192 2 hours\n" + " 1h30m \u2192 1h 30min\n\n" + "Or /cancel.")] + + elif data == "sched_pick_datetime": + with lock: + session["current_step"] = "sched_datetime" + return [ack, reply( + "\U0001f4c5 Date & Time (UTC)\n\n" + "Enter the start time:\n" + "YYYY-MM-DD HH:MM:SS\n" + "Example: 2025-12-31 23:55:00\n\n" + "Or /cancel.")] + + elif data == "sched_cancel_setup": + with lock: + session["current_step"] = None + session["_sched_draft"] = {} + return [ack, reply("❌ Schedule setup cancelled.", + get_main_keyboard(session))] # Quick setup elif data == "quick_setup": with lock: session['current_step'] = "quick_output_url" - return [ack, send_message(chat_id, + return [ack, reply( "⚙️ Quick Setup\n\n" "Step 1/2: Enter your RTMP Output URL:\n" "rtmp://a.rtmp.youtube.com/live2/YOUR_STREAM_KEY\n\n" - "Or /cancel to abort.", - reply_markup=get_main_keyboard(session))] + "Or /cancel to abort.")] # Reset elif data == "confirm_reset": - return [ack, send_message(chat_id, - "⚠️ Reset all settings to defaults?", - reply_markup=get_reset_confirm_keyboard())] + return [ack, edit("⚠️ Reset all settings to defaults?", + get_reset_confirm_keyboard())] elif data == "do_reset": reset_session_settings(chat_id) - return [ack, send_message(chat_id, - "🔄 Settings restored to defaults.\n\n" + compose_status_message(chat_id, True), - reply_markup=get_main_keyboard(session))] + return [ack, reply("🔄 Settings restored to defaults.\n\n" + + compose_status_message(chat_id, True))] - # Logs + # Logs — open as live-view bubble (logs mode), same concept as logg.py /logs elif data == "show_user_logs": - logs = session.get('live_log_lines_user', []) - last = "\n".join(logs[-20:]) if logs else "No logs yet." - return [ack, send_message(chat_id, - "📋 Stream Logs (last 20):\n
" + esc(last) + "
", - reply_markup=get_main_keyboard(session))] + register_live_view(chat_id, message_id, mode="logs") + live_text = _build_live_view_text(chat_id, "logs") + live_kb = _get_live_view_keyboard(chat_id) + with _live_view_lock: + if chat_id in _live_views: + _live_views[chat_id]["last_sent"] = live_text + return [ack, edit(live_text, live_kb)] elif data == "show_help": - return [ack, send_message(chat_id, get_help_text(), reply_markup=get_main_keyboard(session))] + return [ack, reply(get_help_text())] return [ack, answer_callback_query(cq["id"], "Unknown action", show_alert=True)] @@ -1914,7 +2810,7 @@ async def handle_conversation_input(chat_id: int, text: str): append_user_live_log(chat_id, f"Set {field} = {raw}") return send_message(chat_id, f"✅ {field} set to {esc(raw)}\n\n" + format_settings_display(session), - reply_markup=get_settings_keyboard()) + reply_markup=get_settings_keyboard(session)) elif step == "quick_output_url": url = text.strip() @@ -1948,6 +2844,10 @@ async def handle_conversation_input(chat_id: int, text: str): return send_message(chat_id, "🖼 Please send an image file (PNG/JPG), not text. /cancel to abort.") + elif step in ("sched_when", "sched_timer", "sched_datetime", + "sched_name", "sched_rtmp", "sched_input"): + return await handle_schedule_conversation(chat_id, text) + else: with lock: session['current_step'] = None @@ -1981,17 +2881,55 @@ async def shutdown_event(): async def telegram_webhook_endpoint(request: Request): try: update = await request.json() + + # Identify chat_id early to drain any pending real-time status edit + chat_id = None + try: + if "message" in update: + chat_id = update["message"]["chat"]["id"] + elif "callback_query" in update: + chat_id = update["callback_query"]["message"]["chat"]["id"] + except Exception: + pass + response_data = await handle_telegram_update(update) + # Collect the primary response + primary = None if isinstance(response_data, list): - for item in response_data: - if item and isinstance(item, dict) and item.get("method") in ( - "sendMessage", "editMessageText", "answerCallbackQuery" - ): - return item - return response_data[0] if response_data else {"status": "ok"} + items = [i for i in response_data if i and isinstance(i, dict) and i.get("method")] + for priority in ("sendMessage", "editMessageText"): + for item in items: + if item.get("method") == priority: + primary = item + break + if primary: + break + if not primary: + for item in items: + if item.get("method") == "answerCallbackQuery": + primary = item + break + elif isinstance(response_data, dict): + primary = response_data + + # Drain any background-queued outbound messages (live-view edits, scheduler + # notifications, state-change alerts) and return the first one as the response. + # Telegram only accepts one response per webhook call, so extras are re-queued. + if chat_id: + queued_msgs = pop_outbound_messages(chat_id) + if queued_msgs: + if primary is None or primary.get("method") == "answerCallbackQuery": + primary = queued_msgs[0] + # Re-queue remaining for future webhook calls + for m in queued_msgs[1:]: + enqueue_outbound_message(chat_id, m) + else: + # Already have a real primary — re-queue all queued for next call + for m in queued_msgs: + enqueue_outbound_message(chat_id, m) - return response_data + return primary or {"status": "ok"} except json.JSONDecodeError: raise HTTPException(status_code=400, detail="Invalid JSON") @@ -2000,13 +2938,167 @@ async def telegram_webhook_endpoint(request: Request): return {"status": "ok"} +@app.get("/events/{chat_id}") +async def sse_events_endpoint(chat_id: int, request: Request): + """ + Server-Sent Events endpoint for real-time stream status. + Connect from any HTTP client: + GET /events/{chat_id} + Accept: text/event-stream + Events are JSON objects with a "type" field: + type=state — full status snapshot (state, frames, bytes, uptime, keyboard, etc.) + type=log — single log line + type=notification — background event (e.g. scheduled start) + """ + if chat_id not in user_sessions: + # Auto-create session so clients can connect before first message + get_user_session(chat_id) + + q: asyncio.Queue = asyncio.Queue(maxsize=200) + _register_sse_subscriber(chat_id, q) + + async def event_generator(): + try: + # Send immediate snapshot on connect + _push_state_sse(chat_id) + # Send recent logs as a batch + session = get_user_session(chat_id) + recent_logs = session.get('live_log_lines_user', [])[-20:] + if recent_logs: + payload = json.dumps({"type": "log_batch", "lines": recent_logs}) + yield f"data: {payload}\n\n" + + while True: + # Check client disconnected + if await request.is_disconnected(): + break + try: + # Wait up to 3s for an event, then send a heartbeat + payload = await asyncio.wait_for(q.get(), timeout=3.0) + yield f"data: {payload}\n\n" + except asyncio.TimeoutError: + # Heartbeat to keep connection alive + yield f": heartbeat\n\n" + except asyncio.CancelledError: + pass + finally: + _unregister_sse_subscriber(chat_id, q) + + return StreamingResponse( + event_generator(), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Accel-Buffering": "no", + } + ) + + +@app.get("/stream-log/{chat_id}") +async def sse_log_endpoint(chat_id: int, request: Request): + """ + Dedicated SSE endpoint for live log streaming only. + Lighter than /events — only delivers log lines. + """ + if chat_id not in user_sessions: + get_user_session(chat_id) + + q: asyncio.Queue = asyncio.Queue(maxsize=200) + _register_sse_subscriber(chat_id, q) + + async def log_generator(): + try: + # Send recent log history first + session = get_user_session(chat_id) + recent_logs = session.get('live_log_lines_user', [])[-50:] + for line in recent_logs: + payload = json.dumps({"type": "log", "line": line}) + yield f"data: {payload}\n\n" + + while True: + if await request.is_disconnected(): + break + try: + payload = await asyncio.wait_for(q.get(), timeout=5.0) + # Only forward log events + try: + ev = json.loads(payload) + if ev.get("type") in ("log", "log_batch"): + yield f"data: {payload}\n\n" + except Exception: + pass + except asyncio.TimeoutError: + yield f": heartbeat\n\n" + except asyncio.CancelledError: + pass + finally: + _unregister_sse_subscriber(chat_id, q) + + return StreamingResponse( + log_generator(), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Accel-Buffering": "no", + } + ) + + +@app.post("/notify/{chat_id}") +async def internal_notify_endpoint(chat_id: int, request: Request): + """ + Internal endpoint: POST a JSON event to push to SSE subscribers. + Body: {"type": "...", ...} (any JSON) + Also queues a sendMessage to be returned on the next webhook response. + """ + try: + body = await request.json() + except Exception: + body = {} + body.setdefault("ts", datetime.datetime.now(datetime.timezone.utc).isoformat()) + push_sse_event(chat_id, body) + # If there's a "text" field, queue it as an outbound message too + if "text" in body and chat_id in user_sessions: + session = get_user_session(chat_id) + enqueue_outbound_message(chat_id, send_message(chat_id, body["text"], + reply_markup=get_main_keyboard(session))) + return {"queued": True} + + +@app.get("/sessions") +async def list_sessions(): + """List all active sessions and their current state.""" + result = {} + for chat_id, session in user_sessions.items(): + result[str(chat_id)] = { + "state": session.get("streaming_state", "idle"), + "frames_encoded": session.get("frames_encoded", 0), + "bytes_sent": session.get("bytes_sent", 0), + "sse_subscribers": len(_sse_subscribers.get(chat_id, [])), + "queued_messages": len(_outbound_message_queue.get(chat_id, [])), + } + return result + + @app.get("/") async def root(): return { "bot": f"Advanced Stream Bot v{APP_VERSION}", "status": "running", - "webhook": "/webhook", + "endpoints": { + "webhook": "POST /webhook", + "status": "GET /status/{chat_id}", + "sessions": "GET /sessions", + "sse_events": "GET /events/{chat_id} (text/event-stream)", + "sse_logs": "GET /stream-log/{chat_id} (text/event-stream)", + "notify": "POST /notify/{chat_id}", + "health": "GET /health", + }, "sessions": len(user_sessions), + "sse_connections": sum(len(v) for v in _sse_subscribers.values()), + "scheduler_jobs": len(scheduler.get_jobs()), } @@ -2017,6 +3109,89 @@ async def health(): "version": APP_VERSION, "sessions": len(user_sessions), "scheduler_jobs": len(scheduler.get_jobs()), + "sse_connections": sum(len(v) for v in _sse_subscribers.values()), + "queued_messages": sum(len(v) for v in _outbound_message_queue.values()), + } + + +@app.get("/status/{chat_id}") +async def get_status_endpoint(chat_id: int): + """HTTP polling — returns current stream state as JSON for external dashboards.""" + if chat_id not in user_sessions: + raise HTTPException(status_code=404, detail="No session for this chat_id") + session = get_user_session(chat_id) + uptime = 0 + if session.get("stream_start_time"): + try: + uptime = int((datetime.datetime.now(datetime.timezone.utc) - session["stream_start_time"]).total_seconds()) + except Exception: + pass + return { + "chat_id": chat_id, + "state": session.get("streaming_state", "idle"), + "state_icon": STATE_EMOJI.get(session.get("streaming_state", "idle"), "❓"), + "frames_encoded": session.get("frames_encoded", 0), + "bytes_sent": session.get("bytes_sent", 0), + "uptime_seconds": uptime, + "uptime_str": get_uptime(session.get("stream_start_time")), + "reconnect_attempt": session.get("reconnect_attempt", 0), + "error": session.get("error_notification_user", ""), + "active_output_url": session.get("active_output_url", ""), + "playlist_index": session.get("current_playlist_index", 0), + "playlist_count": len(session.get("input_url_playlist", [])), + "sse_subscribers": len(_sse_subscribers.get(chat_id, [])), + "queued_messages": len(_outbound_message_queue.get(chat_id, [])), + "keyboard": get_main_keyboard(session), + "status_text": compose_status_message(chat_id, include_config=False), + "ts": datetime.datetime.now(datetime.timezone.utc).isoformat(), + } + + +@app.get("/poll/{chat_id}") +async def poll_pending_edit(chat_id: int): + """ + Lightweight polling endpoint — returns the pending live-view edit (if any) + as a Telegram Bot API method dict, then clears it. + + The Telegram Mini App / web dashboard can call this every 2s and POST the + returned payload directly to api.telegram.org/bot/editMessageText to + achieve real-time live-view updates without waiting for an incoming webhook. + + If no edit is pending, returns {"pending": false}. + Also always includes a fresh status snapshot so callers can update their UI. + """ + if chat_id not in user_sessions: + raise HTTPException(status_code=404, detail="No session for this chat_id") + + # Force-build a fresh live-view edit right now + _do_live_view_edit(chat_id, force=True) + + with _outbound_queue_lock: + msgs = _outbound_message_queue.pop(chat_id, []) + + # Pick the most recent editMessageText (last one wins — it's the freshest) + edit_msg = None + for m in reversed(msgs): + if m.get("method") == "editMessageText": + edit_msg = m + break + # Re-queue any non-edit messages (sendMessage notifications etc.) + for m in msgs: + if m.get("method") != "editMessageText": + enqueue_outbound_message(chat_id, m) + + session = get_user_session(chat_id) + return { + "pending": edit_msg is not None, + "edit": edit_msg, + "status": { + "state": session.get("streaming_state", "idle"), + "frames_encoded": session.get("frames_encoded", 0), + "bytes_sent": session.get("bytes_sent", 0), + "uptime_str": get_uptime(session.get("stream_start_time")), + "logs": session.get("live_log_lines_user", [])[-15:], + }, + "ts": datetime.datetime.now(datetime.timezone.utc).isoformat(), } @@ -2026,5 +3201,4 @@ async def health(): if __name__ == "__main__": import uvicorn logger.info(f"Starting Advanced Stream Bot v{APP_VERSION}...") - logger.warning("⚠️ Set TELEGRAM_BOT_TOKEN env variable for full functionality.") uvicorn.run("stream_bot:app", host="0.0.0.0", port=8000, reload=False) \ No newline at end of file