| """ |
| 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 |
| import threading |
| import time |
| import datetime |
| import traceback |
| import fractions |
| import json |
| import os |
| import re |
| import asyncio |
| from urllib.parse import urlparse |
| from pathlib import Path |
| import io |
|
|
| 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 |
|
|
| |
| |
| |
| APP_VERSION = "3.3.0" |
|
|
| app = FastAPI(title="Advanced Stream Bot", version=APP_VERSION) |
| scheduler = BackgroundScheduler(timezone="UTC") |
| |
| |
| _job_store: dict = {} |
|
|
|
|
| |
| |
| |
| logging.basicConfig( |
| level=logging.INFO, |
| format="%(asctime)s [%(levelname)s] %(threadName)s %(module)s:%(lineno)d %(message)s", |
| datefmt="%Y-%m-%d %H:%M:%S", |
| ) |
| logger = logging.getLogger("stream_bot") |
|
|
| live_log_lines_global = [] |
| MAX_GLOBAL_LOG_LINES = 200 |
|
|
|
|
| def append_global_live_log(line: str): |
| live_log_lines_global.append(line) |
| if len(live_log_lines_global) > MAX_GLOBAL_LOG_LINES: |
| live_log_lines_global.pop(0) |
|
|
|
|
| class GlobalListHandler(logging.Handler): |
| def emit(self, record): |
| append_global_live_log(self.format(record)) |
|
|
|
|
| _gh = GlobalListHandler() |
| _gh.setFormatter(logging.Formatter("[%(levelname)s] %(message)s")) |
| logger.addHandler(_gh) |
|
|
| |
| |
| |
| SUPPORTED_VIDEO_CODECS = ["libx264", "h264_nvenc", "h264_qsv", "libx265", "hevc_nvenc", "hevc_qsv", "copy"] |
| SUPPORTED_AUDIO_CODECS = ["aac", "opus", "libmp3lame", "copy"] |
| LOGO_POSITIONS = ["top_left", "top_right", "bottom_left", "bottom_right", "center"] |
| FFMPEG_PRESETS = ["ultrafast", "superfast", "veryfast", "faster", "fast", "medium", "slow", "slower", "veryslow"] |
| OUTPUT_FORMATS = ["flv", "mpegts", "mp4", "hls"] |
|
|
| |
| 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}, |
| "high": {"video_bitrate": "3000k", "audio_bitrate": "192k", "ffmpeg_preset": "fast", "resolution": "1920x1080","fps": 30}, |
| "ultra": {"video_bitrate": "6000k", "audio_bitrate": "256k", "ffmpeg_preset": "slow", "resolution": "1920x1080","fps": 60}, |
| "source": {}, |
| } |
|
|
| |
| |
| |
| DEFAULT_USER_SETTINGS = { |
| |
| "input_url_playlist": [], |
| "current_playlist_index": 0, |
| "output_url": "rtmp://a.rtmp.youtube.com/live2/YOUR_KEY", |
| "output_format": "flv", |
|
|
| |
| "quality_preset": "medium", |
|
|
| |
| "video_codec": "libx264", |
| "resolution": "1280x720", |
| "fps": 30, |
| "gop_size": 60, |
| "video_bitrate": "1500k", |
| "ffmpeg_preset": "medium", |
| "video_pix_fmt": "yuv420p", |
| "video_thread_count": 0, |
|
|
| |
| "audio_codec": "aac", |
| "audio_bitrate": "128k", |
| "audio_sample_rate": 44100, |
| "audio_channels": 2, |
|
|
| |
| "loop_count": 0, |
| "stop_on_error_in_playlist": True, |
| "reconnect_on_stream_error": True, |
| "reconnect_delay_seconds": 5, |
| "max_reconnect_attempts": 3, |
|
|
| |
| "open_timeout_seconds": 15, |
| "read_timeout_seconds": 30, |
|
|
| |
| "logo_enabled": False, |
| "logo_data_bytes": None, |
| "logo_mime_type": None, |
| "logo_original_filename": None, |
| "logo_position": "top_right", |
| "logo_scale": 0.10, |
| "logo_opacity": 0.85, |
| "logo_margin_px": 10, |
|
|
| |
| "verbose_ffmpeg_log": False, |
|
|
| |
| "current_step": None, |
| "current_step_index": 0, |
| "conversation_fields_list": [], |
| "settings_editing_field": None, |
|
|
| |
| "ux_mode": "send", |
| } |
|
|
| DEFAULT_SESSION_RUNTIME_STATE = { |
| "streaming_state": "idle", |
| "stream_start_time": None, |
| "frames_encoded": 0, |
| "bytes_sent": 0, |
| "stream_thread_ref": None, |
| "watchdog_thread_ref": None, |
| "pyav_objects": { |
| "input_container": None, |
| "output_container": None, |
| "video_out_stream": None, |
| "audio_out_stream": None, |
| "logo_image_pil": None, |
| }, |
| "live_log_lines_user": [], |
| "error_notification_user": "", |
| "stop_gracefully_flag": False, |
| "current_loop_iteration": 0, |
| "reconnect_attempt": 0, |
| "last_frame_time": None, |
| "last_notified_state": None, |
| "status_message_id": None, |
| "status_chat_id": None, |
| "active_output_url": None, |
| } |
|
|
| |
| |
| |
| user_sessions: dict = {} |
| session_locks: dict = {} |
| |
| _main_event_loop: asyncio.AbstractEventLoop = None |
|
|
| |
| |
| |
| |
| |
| _sse_subscribers: dict = {} |
| _sse_lock = threading.Lock() |
|
|
| |
| |
| _outbound_message_queue: 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: |
| session_locks[chat_id] = threading.Lock() |
| with session_locks[chat_id]: |
| session = {} |
| for k, v in DEFAULT_USER_SETTINGS.items(): |
| session[k] = list(v) if isinstance(v, list) else (dict(v) if isinstance(v, dict) else v) |
| 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) |
| user_sessions[chat_id] = session |
| logger.info(f"[Chat {chat_id}] New session created (in-memory).") |
| return user_sessions[chat_id] |
|
|
|
|
| def reset_session_settings(chat_id: int): |
| """Restore all settings to defaults, keep runtime state.""" |
| session = get_user_session(chat_id) |
| lock = session_locks[chat_id] |
| with lock: |
| for k, v in DEFAULT_USER_SETTINGS.items(): |
| session[k] = list(v) if isinstance(v, list) else (dict(v) if isinstance(v, dict) else v) |
| append_user_live_log(chat_id, "Settings restored to defaults.") |
|
|
|
|
| def append_user_live_log(chat_id: int, line: str): |
| session = get_user_session(chat_id) |
| lock = session_locks.get(chat_id) |
| entry = f"{datetime.datetime.now().strftime('%H:%M:%S')} {line}" |
| if lock: |
| with lock: |
| session['live_log_lines_user'].append(entry) |
| if len(session['live_log_lines_user']) > 100: |
| session['live_log_lines_user'].pop(0) |
| else: |
| session['live_log_lines_user'].append(entry) |
| |
| push_sse_event(chat_id, {"type": "log", "line": entry}) |
|
|
|
|
| |
| |
| |
| |
| def push_message(chat_id: int, text: str, reply_markup=None, parse_mode="HTML"): |
| """ |
| 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]}") |
|
|
|
|
| |
| |
| |
| def send_message(chat_id: int, text: str, parse_mode="HTML", reply_markup=None): |
| """Build a sendMessage response dict (for webhook response).""" |
| resp = {"method": "sendMessage", "chat_id": chat_id, "text": text, "parse_mode": parse_mode} |
| if reply_markup: |
| resp["reply_markup"] = reply_markup |
| return resp |
|
|
|
|
| def edit_message_text(chat_id: int, message_id: int, text: str, parse_mode="HTML", reply_markup=None): |
| resp = {"method": "editMessageText", "chat_id": chat_id, "message_id": message_id, |
| "text": text, "parse_mode": parse_mode} |
| if reply_markup: |
| resp["reply_markup"] = reply_markup |
| return resp |
|
|
|
|
| def answer_callback_query(cq_id: str, text: str = None, show_alert: bool = False): |
| resp = {"method": "answerCallbackQuery", "callback_query_id": cq_id} |
| if text: |
| resp["text"] = text |
| if show_alert: |
| resp["show_alert"] = True |
| return resp |
|
|
|
|
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| LIVE_VIEW_UPDATE_INTERVAL = 2 |
|
|
| _live_views: dict = {} |
| _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<i>⏸ Updates paused</i>" if is_paused else "" |
| return ( |
| f"📋 <b>Live Logs</b> {'(paused)' if is_paused else '(updating…)'}\n" |
| f"<pre>{esc(log_tail)}</pre>" |
| 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[-4:]) if logs else "No logs yet." |
| pause_note = " <i>⏸ paused</i>" if is_paused else "" |
|
|
| |
| if state in ("streaming", "starting", "paused", "reconnecting"): |
| return ( |
| f"{status}" |
| f"\n📋 <b>Recent Logs:</b>{pause_note}\n<pre>{esc(log_tail)}</pre>" |
| ) |
| 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 = [] |
|
|
| |
| 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) |
|
|
| |
| 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, |
| "paused_at": None, |
| "mode": mode, |
| } |
| 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 |
|
|
| new_text = _build_live_view_text(chat_id, mode) |
| kb = _get_live_view_keyboard(chat_id) |
|
|
| |
| if new_text == last_sent and not force: |
| |
| _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) |
|
|
| |
| |
| with _outbound_queue_lock: |
| existing = _outbound_message_queue.get(chat_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 |
|
|
| |
| _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) |
|
|
| |
| if state in ('idle', 'stopped', 'completed', 'error'): |
| _do_live_view_edit(chat_id, force=True) |
| unregister_live_view(chat_id) |
| |
| _push_state_sse(chat_id) |
|
|
| except Exception as e_inner: |
| logger.warning(f"Live view updater error for {chat_id}: {e_inner}") |
|
|
| |
| 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() |
|
|
|
|
| |
| 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 |
|
|
|
|
| |
| |
| |
| def get_main_keyboard(session: dict): |
| """Returns inline keyboard appropriate for current session state.""" |
| state = session.get('streaming_state', 'idle') |
| streaming = state in ("streaming", "paused", "starting", "reconnecting") |
| playlist_count = len(session.get('input_url_playlist', [])) |
|
|
| if streaming: |
| |
| btns = [] |
| row1 = [] |
| if state == "streaming": |
| row1.append({"text": "⏸ Pause", "callback_data": "stream_pause"}) |
| elif state == "paused": |
| row1.append({"text": "▶️ Resume", "callback_data": "stream_resume"}) |
| row1.append({"text": "⏹ Stop", "callback_data": "stream_abort"}) |
| btns.append(row1) |
|
|
| btns.append([ |
| {"text": "📊 Status", "callback_data": "stream_status"}, |
| {"text": "📋 Logs", "callback_data": "show_user_logs"}, |
| ]) |
| 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: |
| |
| btns = [] |
| |
| if playlist_count > 0 and session.get('output_url'): |
| btns.append([{"text": "🚀 Start Stream", "callback_data": "stream_start"}]) |
| else: |
| btns.append([{"text": "⚙️ Quick Setup (required)", "callback_data": "quick_setup"}]) |
|
|
| |
| btns.append([ |
| {"text": f"📜 Playlist ({playlist_count})", "callback_data": "view_playlist"}, |
| {"text": "⚙️ Settings", "callback_data": "open_settings"}, |
| ]) |
|
|
| |
| logo_icon = "🖼✅" if session.get('logo_enabled') else "🖼" |
| btns.append([ |
| {"text": logo_icon + " Logo", "callback_data": "cfg_logo"}, |
| {"text": "🕒 Schedule", "callback_data": "cfg_schedule"}, |
| ]) |
|
|
| |
| btns.append([ |
| {"text": "🔄 Reset to Defaults", "callback_data": "confirm_reset"}, |
| {"text": "❔ Help", "callback_data": "show_help"}, |
| ]) |
| |
| btns.append([ |
| {"text": "⚡ Force Reboot", "callback_data": "force_reboot"}, |
| ]) |
|
|
| return {"inline_keyboard": btns} |
|
|
|
|
| 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": "pick_resolution"}, |
| {"text": "🎞 FPS", "callback_data": "set_fps"}], |
| [{"text": "📶 Video Bitrate", "callback_data": "set_video_bitrate"}, |
| {"text": "🔉 Audio Bitrate", "callback_data": "set_audio_bitrate"}], |
| [{"text": "⚡ Preset", "callback_data": "set_ffmpeg_preset"}, |
| {"text": "🔁 Loop Count", "callback_data": "set_loop_count"}], |
| [{"text": "🔀 GOP Size", "callback_data": "set_gop_size"}, |
| {"text": "📦 Output Format", "callback_data": "set_output_format"}], |
| [{"text": "⏱ Reconnect Delay", "callback_data": "set_reconnect_delay"}, |
| {"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"}], |
| ]} |
|
|
|
|
| def get_quality_keyboard(): |
| rows = [[{"text": q.title(), "callback_data": f"apply_quality_{q}"} for q in list(QUALITY_PRESETS.keys())[:3]]] |
| rows.append([{"text": q.title(), "callback_data": f"apply_quality_{q}"} for q in list(QUALITY_PRESETS.keys())[3:]]) |
| rows.append([{"text": "↩ Back", "callback_data": "open_settings"}]) |
| return {"inline_keyboard": rows} |
|
|
|
|
| def get_codec_keyboard(codec_type: str): |
| codecs = SUPPORTED_VIDEO_CODECS if codec_type == "video" else SUPPORTED_AUDIO_CODECS |
| cb_prefix = "set_vcodec_" if codec_type == "video" else "set_acodec_" |
| rows = [] |
| row = [] |
| for c in codecs: |
| row.append({"text": c, "callback_data": cb_prefix + c}) |
| if len(row) == 2: |
| rows.append(row) |
| row = [] |
| if row: |
| rows.append(row) |
| rows.append([{"text": "↩ Back to Settings", "callback_data": "open_settings"}]) |
| return {"inline_keyboard": rows} |
|
|
|
|
| def get_preset_keyboard(): |
| rows = [] |
| row = [] |
| for p in FFMPEG_PRESETS: |
| row.append({"text": p, "callback_data": "set_preset_" + p}) |
| if len(row) == 3: |
| rows.append(row) |
| row = [] |
| if row: |
| rows.append(row) |
| rows.append([{"text": "↩ Back", "callback_data": "open_settings"}]) |
| 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] |
| rows.append([{"text": "↩ Back", "callback_data": "cfg_logo"}]) |
| return {"inline_keyboard": rows} |
|
|
|
|
| def get_reset_confirm_keyboard(): |
| return {"inline_keyboard": [ |
| [{"text": "✅ Yes, Reset Everything", "callback_data": "do_reset"}, |
| {"text": "❌ Cancel", "callback_data": "show_help"}], |
| ]} |
|
|
|
|
| |
| |
| |
| def validate_url(url: str) -> bool: |
| try: |
| r = urlparse(url) |
| return all([r.scheme, r.netloc]) and r.scheme in ['http', 'https', 'rtmp', 'rtsp', 'udp', 'srt', 'file'] |
| except Exception: |
| return False |
|
|
|
|
| def validate_resolution(res: str) -> bool: |
| if res.lower() == "source": |
| return True |
| return bool(re.match(r"^\d{3,4}x\d{3,4}$", res)) |
|
|
|
|
| def parse_bitrate(bitrate_str: str) -> int: |
| s = str(bitrate_str).lower() |
| try: |
| if s.endswith('m'): |
| return int(float(s[:-1]) * 1_000_000) |
| if s.endswith('k'): |
| return int(float(s[:-1]) * 1_000) |
| return int(s) |
| except ValueError: |
| return 1_500_000 |
|
|
|
|
| def esc(text: str) -> str: |
| """HTML-escape a string.""" |
| return str(text).replace("&", "&").replace("<", "<").replace(">", ">") |
|
|
|
|
| |
| |
| |
| STATE_EMOJI = { |
| "idle": "💤", "starting": "🔄", "streaming": "🔴", |
| "paused": "⏸", "stopping": "🛑", "stopped": "⏹", |
| "completed": "✅", "error": "❌", "reconnecting": "🔁", |
| } |
|
|
|
|
| def get_uptime(start_time_obj) -> str: |
| if not start_time_obj: |
| return "—" |
| if start_time_obj.tzinfo is None: |
| start_time_obj = start_time_obj.replace(tzinfo=datetime.timezone.utc) |
| delta = datetime.datetime.now(datetime.timezone.utc) - start_time_obj |
| return str(delta).split('.')[0] |
|
|
|
|
| def format_settings_display(session: dict) -> str: |
| logo_status = "Disabled" |
| if session.get('logo_enabled') and session.get('logo_data_bytes'): |
| fname = session.get('logo_original_filename', 'Unknown') |
| logo_status = f"✅ {esc(fname)}" |
|
|
| loop = session.get('loop_count', 0) |
| loop_disp = "∞ Infinite" if loop == -1 else ("Once" if loop == 0 else f"{loop}×") |
|
|
| 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"📡 <b>Output URL:</b> <code>{esc(session.get('output_url', '—'))}</code>", |
| f"📜 <b>Playlist:</b> <code>{len(session.get('input_url_playlist', []))} item(s)</code>", |
| "", |
| f"🎨 <b>Quality Preset:</b> <code>{esc(session.get('quality_preset', '—'))}</code>", |
| f"🎥 <b>Video:</b> <code>{esc(session.get('video_codec'))} | {esc(session.get('resolution'))} | {session.get('fps')}fps | {esc(session.get('video_bitrate'))}</code>", |
| f"🔊 <b>Audio:</b> <code>{esc(session.get('audio_codec'))} | {esc(session.get('audio_bitrate'))}</code>", |
| f"⚡ <b>Preset:</b> <code>{esc(session.get('ffmpeg_preset'))}</code> 🔀 <b>GOP:</b> <code>{session.get('gop_size')}</code>", |
| f"📦 <b>Format:</b> <code>{esc(session.get('output_format', 'flv'))}</code>", |
| "", |
| f"🔁 <b>Loop:</b> <code>{loop_disp}</code>", |
| f"🔄 <b>Auto-Reconnect:</b> {auto_rc} (delay: <code>{session.get('reconnect_delay_seconds')}s</code>, max: <code>{session.get('max_reconnect_attempts')}</code>)", |
| f"🛑 <b>Stop on Error:</b> {stop_err}", |
| f"📋 <b>Verbose FFmpeg Log:</b> {verbose_log}", |
| f"⏱ <b>Timeouts:</b> open <code>{session.get('open_timeout_seconds')}s</code> read <code>{session.get('read_timeout_seconds')}s</code>", |
| f"🖼 <b>Logo:</b> {logo_status}", |
| ] |
| if session.get('logo_enabled') and session.get('logo_data_bytes'): |
| lines.append( |
| f" Position: <code>{session.get('logo_position')}</code> " |
| f"Scale: <code>{session.get('logo_scale')}</code> " |
| f"Opacity: <code>{session.get('logo_opacity')}</code>" |
| ) |
| return "\n".join(lines) |
|
|
|
|
| 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, "❓") |
|
|
| lines = [ |
| f"🤖 <b>Stream Bot v{APP_VERSION}</b>", |
| f"{icon} <b>State:</b> <code>{state}</code>", |
| ] |
|
|
| if session.get('error_notification_user'): |
| lines.append(f"⚠️ <b>Last Error:</b> <code>{esc(session['error_notification_user'])}</code>") |
|
|
| if state in ("streaming", "paused", "starting", "stopping", "reconnecting"): |
| playlist = session.get('input_url_playlist', []) |
| idx = session.get('current_playlist_index', 0) |
| cur_url = playlist[idx] if playlist and 0 <= idx < len(playlist) else "N/A" |
| loop = session.get('loop_count', 0) |
| 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"⏱ <b>Uptime:</b> <code>{get_uptime(session.get('stream_start_time'))}</code>", |
| f"🎞 <b>Frames:</b> <code>{frames:,}</code>", |
| f"📤 <b>Sent:</b> <code>{bytes_s / (1024*1024):.2f} MB</code>", |
| ] |
|
|
| lines += [ |
| *stats_lines, |
| f"🎬 <b>Input:</b> <code>{esc(cur_url)}</code> ({idx+1}/{len(playlist)})", |
| f"🔁 <b>Loop:</b> <code>{session.get('current_loop_iteration', 0)+1}/{loop_total}</code>", |
| ] |
| if reconnect_att > 0: |
| lines.append(f"🔄 <b>Reconnect attempts:</b> <code>{reconnect_att}</code>") |
|
|
| if include_config or state in ("idle", "stopped", "completed", "error"): |
| lines += ["", "⚙️ <b>Configuration:</b>", format_settings_display(session)] |
|
|
| if include_logs and state in ("streaming", "paused", "starting", "reconnecting"): |
| user_logs = session.get('live_log_lines_user', []) |
| last_logs = "\n".join(user_logs[-4:]) if user_logs else "No logs yet." |
| lines += [f"\n📋 <b>Recent Logs:</b>\n<pre>{esc(last_logs)}</pre>"] |
|
|
| return "\n".join(lines) |
|
|
|
|
| |
| |
| |
| def notify_state_change(chat_id: int, new_state: str, extra_msg: str = ""): |
| """Push a message to user when stream state changes.""" |
| session = get_user_session(chat_id) |
| icon = STATE_EMOJI.get(new_state, "❓") |
| text = f"{icon} <b>Stream state changed:</b> <code>{new_state}</code>" |
| if extra_msg: |
| text += f"\n{extra_msg}" |
| text += "\n\n" + compose_status_message(chat_id) |
| kb = get_main_keyboard(session) |
| |
| push_message(chat_id, text, reply_markup=kb) |
| enqueue_outbound_message(chat_id, send_message(chat_id, text, reply_markup=kb)) |
| |
| _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 |
| |
| 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 = ""): |
| """Thread-safe state update + push notification if state changed.""" |
| old_state = session.get('streaming_state') |
| session['streaming_state'] = new_state |
| should_notify = (old_state != new_state) |
|
|
| if should_notify: |
| |
| threading.Thread( |
| target=notify_state_change, |
| args=(chat_id, new_state, extra_msg), |
| daemon=True |
| ).start() |
| else: |
| |
| threading.Thread(target=_push_state_sse, args=(chat_id,), daemon=True).start() |
|
|
|
|
| |
| |
| |
| def watchdog_thread_target(chat_id: int): |
| """Monitors the stream thread for freezes and notifies user.""" |
| session = get_user_session(chat_id) |
| lock = session_locks[chat_id] |
| STALL_THRESHOLD = 30 |
|
|
| logger.info(f"[Chat {chat_id}] Watchdog started.") |
| while True: |
| time.sleep(5) |
| with lock: |
| state = session.get('streaming_state') |
| |
| if state not in ("streaming", "paused", "reconnecting", "starting"): |
| break |
|
|
| last_frame = session.get('last_frame_time') |
| thread_ref = session.get('stream_thread_ref') |
|
|
| |
| if thread_ref and not thread_ref.is_alive(): |
| with lock: |
| cur_state = session.get('streaming_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: |
| session['streaming_state'] = "error" |
| session['error_notification_user'] = "Stream thread terminated unexpectedly." |
| notify_state_change(chat_id, "error", "The stream thread crashed. Use /stream to restart.") |
| break |
|
|
| |
| if state == "streaming" and last_frame: |
| 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!") |
| enqueue_outbound_message(chat_id, send_message( |
| chat_id, |
| f"⚠️ <b>Stream may be frozen!</b>\nNo frames for <code>{stall_seconds:.0f}s</code>.\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.") |
|
|
|
|
| |
| |
| |
| def stream_engine_thread_target(chat_id: int): |
| session = get_user_session(chat_id) |
| lock = session_locks[chat_id] |
|
|
| def _cleanup_pyav(): |
| append_user_live_log(chat_id, "Cleaning up PyAV resources…") |
| with lock: |
| objs = session.get('pyav_objects', {}) |
| for key in ('output_container', 'input_container'): |
| container = objs.get(key) |
| if container: |
| try: |
| container.close() |
| except Exception as e: |
| append_user_live_log(chat_id, f"Error closing {key}: {e}") |
| objs[key] = None |
| objs['video_out_stream'] = None |
| objs['audio_out_stream'] = None |
| objs['logo_image_pil'] = None |
|
|
| active_output_container = None |
|
|
| |
| |
| |
| |
| _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_on = session.get('verbose_ffmpeg_log', False) |
| if not verbose_on and level > 24: |
| continue |
| |
| 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" |
| session['error_notification_user'] = "" |
| session['stream_start_time'] = datetime.datetime.now(datetime.timezone.utc) |
| session['frames_encoded'] = 0 |
| session['bytes_sent'] = 0 |
| session['current_loop_iteration'] = 0 |
| session['current_playlist_index'] = 0 |
| session['stop_gracefully_flag'] = False |
| session['reconnect_attempt'] = 0 |
| session['last_frame_time'] = None |
| session['live_log_lines_user'] = [] |
| session['pyav_objects'] = { |
| "input_container": None, "output_container": None, |
| "video_out_stream": None, "audio_out_stream": None, |
| "logo_image_pil": None, |
| } |
| |
| 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']}") |
|
|
| |
| notify_state_change(chat_id, "starting") |
|
|
| |
| |
| with lock: |
| output_url = session['output_url'] |
| output_format = session.get('output_format', 'flv') |
| open_timeout = session.get('open_timeout_seconds', 15) |
| |
| 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, |
| timeout=open_timeout |
| ) |
| with lock: |
| session['pyav_objects']['output_container'] = active_output_container |
| append_user_live_log(chat_id, f"Output container opened ({output_format}).") |
| except Exception as e: |
| raise Exception(f"Failed to open output stream: {e}") |
|
|
| |
| logo_pil_original = None |
| if session.get('logo_enabled') and session.get('logo_data_bytes'): |
| try: |
| logo_pil_original = Image.open(io.BytesIO(session['logo_data_bytes'])).convert("RGBA") |
| opacity = session.get('logo_opacity', 1.0) |
| if opacity < 1.0: |
| alpha = logo_pil_original.split()[-1] |
| alpha = ImageEnhance.Brightness(alpha).enhance(opacity) |
| logo_pil_original.putalpha(alpha) |
| with lock: |
| session['pyav_objects']['logo_image_pil'] = logo_pil_original |
| append_user_live_log(chat_id, f"Logo loaded: {session.get('logo_original_filename', 'N/A')}") |
| except Exception as e: |
| append_user_live_log(chat_id, f"Logo load error: {e}. Skipping logo.") |
| logo_pil_original = None |
|
|
| |
| total_loops = session.get('loop_count', 0) |
| playlist = session.get('input_url_playlist', []) |
|
|
| if not playlist: |
| raise Exception("Playlist is empty. Add URLs with /playlist add <url>") |
|
|
| active_video_out_stream = None |
| active_audio_out_stream = None |
| output_configured = False |
| current_logo_resized = None |
| logo_pos_x, logo_pos_y = 0, 0 |
|
|
| while True: |
| with lock: |
| if session.get('stop_gracefully_flag') or session['streaming_state'] == "stopping": |
| break |
| current_loop_iter = session['current_loop_iteration'] |
| |
| 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).") |
| break |
|
|
| append_user_live_log(chat_id, f"Loop iteration {current_loop_iter + 1}") |
| with lock: |
| session['current_playlist_index'] = 0 |
|
|
| while True: |
| with lock: |
| cur_idx = session['current_playlist_index'] |
| should_stop = session.get('stop_gracefully_flag') or session['streaming_state'] == "stopping" |
| playlist_done = cur_idx >= len(playlist) |
| cur_playlist = list(session['input_url_playlist']) |
|
|
| if should_stop or playlist_done: |
| break |
|
|
| current_input_url = cur_playlist[cur_idx] |
| append_user_live_log(chat_id, f"Input [{cur_idx+1}/{len(cur_playlist)}]: {current_input_url}") |
|
|
| _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', 3) |
| read_timeout = session.get('read_timeout_seconds', 30) |
|
|
| try: |
| _input_container = av.open(current_input_url, timeout=open_timeout) |
| with lock: |
| session['pyav_objects']['input_container'] = _input_container |
| session['reconnect_attempt'] = 0 |
|
|
| in_v_streams = _input_container.streams.video |
| in_a_streams = _input_container.streams.audio |
| if not in_v_streams: |
| raise Exception("No video stream found in input.") |
|
|
| in_v_s = in_v_streams[0] |
| target_w, target_h = in_v_s.width, in_v_s.height |
| if session.get('resolution', 'source') != 'source': |
| try: |
| target_w, target_h = map(int, session['resolution'].split('x')) |
| except Exception: |
| pass |
|
|
| |
| if not output_configured: |
| fps = session.get('fps', 30) |
| active_video_out_stream = active_output_container.add_stream( |
| session.get('video_codec', 'libx264'), rate=fps |
| ) |
| active_video_out_stream.width = target_w |
| active_video_out_stream.height = target_h |
| active_video_out_stream.pix_fmt = session.get('video_pix_fmt', 'yuv420p') |
| active_video_out_stream.codec_context.gop_size = session.get('gop_size', fps * 2) |
| active_video_out_stream.codec_context.bit_rate = parse_bitrate(session.get('video_bitrate', '1500k')) |
| tc = session.get('video_thread_count', 0) |
| if tc > 0: |
| active_video_out_stream.codec_context.thread_count = tc |
| active_video_out_stream.codec_context.thread_type = "AUTO" |
| preset = session.get('ffmpeg_preset', 'medium') |
| if session.get('video_codec', 'libx264') not in ('copy',): |
| active_video_out_stream.codec_context.options = {'preset': preset} |
|
|
| with lock: |
| session['pyav_objects']['video_out_stream'] = active_video_out_stream |
|
|
| if in_a_streams: |
| in_a_s = in_a_streams[0] |
| a_rate = session.get('audio_sample_rate') or in_a_s.rate |
| active_audio_out_stream = active_output_container.add_stream( |
| session.get('audio_codec', 'aac'), rate=a_rate |
| ) |
| active_audio_out_stream.codec_context.bit_rate = parse_bitrate(session.get('audio_bitrate', '128k')) |
| with lock: |
| session['pyav_objects']['audio_out_stream'] = active_audio_out_stream |
|
|
| |
| if logo_pil_original: |
| logo_scale = session.get('logo_scale', 0.10) |
| margin = session.get('logo_margin_px', 10) |
| logo_h = int(target_h * logo_scale) |
| logo_w = int(logo_h * (logo_pil_original.width / logo_pil_original.height)) |
| logo_w = min(logo_w, target_w) |
| logo_h = min(logo_h, target_h) |
| if logo_w > 0 and logo_h > 0: |
| current_logo_resized = logo_pil_original.resize((logo_w, logo_h), Image.Resampling.LANCZOS) |
| pos = session.get('logo_position', 'top_right') |
| if pos == 'top_right': |
| logo_pos_x, logo_pos_y = target_w - logo_w - margin, margin |
| elif pos == 'top_left': |
| logo_pos_x, logo_pos_y = margin, margin |
| elif pos == 'bottom_left': |
| logo_pos_x, logo_pos_y = margin, target_h - logo_h - margin |
| elif pos == 'bottom_right': |
| logo_pos_x, logo_pos_y = target_w - logo_w - margin, target_h - logo_h - margin |
| elif pos == 'center': |
| logo_pos_x, logo_pos_y = (target_w - logo_w) // 2, (target_h - logo_h) // 2 |
|
|
| output_configured = True |
| append_user_live_log(chat_id, f"Output streams configured: {target_w}x{target_h}@{fps}fps") |
|
|
| with lock: |
| if session['streaming_state'] != "paused": |
| old_st = session['streaming_state'] |
| session['streaming_state'] = "streaming" |
| if old_st != "streaming": |
| threading.Thread(target=notify_state_change, args=(chat_id, "streaming"), daemon=True).start() |
|
|
| |
| last_read_time = time.time() |
| for packet in _input_container.demux(video=0, audio=0 if in_a_streams else -1): |
| with lock: |
| if session['streaming_state'] == "stopping": |
| break |
| while session['streaming_state'] == "paused": |
| lock.release() |
| time.sleep(0.2) |
| lock.acquire() |
| if session['streaming_state'] == "stopping": |
| break |
|
|
| |
| if time.time() - last_read_time > read_timeout: |
| raise Exception(f"Read timeout after {read_timeout}s of no packets.") |
| last_read_time = time.time() |
|
|
| if packet.dts is None: |
| continue |
|
|
| if packet.stream.type == 'video' and packet.stream.index == in_v_s.index: |
| for frame in packet.decode(): |
| if not frame.width or not frame.height: |
| continue |
| if frame.width != target_w or frame.height != target_h: |
| frame = frame.reformat(target_w, target_h, format='yuv420p') |
|
|
| if current_logo_resized: |
| try: |
| pil_f = frame.to_image() |
| ax = max(0, min(logo_pos_x, pil_f.width - current_logo_resized.width)) |
| ay = max(0, min(logo_pos_y, pil_f.height - current_logo_resized.height)) |
| pil_f.paste(current_logo_resized, (ax, ay), current_logo_resized) |
| new_pts = frame.pts |
| frame = av.VideoFrame.from_image(pil_f) |
| frame.pts = new_pts |
| except Exception as e_logo: |
| pass |
|
|
| |
| |
| |
| |
| with lock: |
| session['frames_encoded'] += 1 |
| session['last_frame_time'] = datetime.datetime.now() |
| _fc = session['frames_encoded'] |
| |
| 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['bytes_sent'] += out_pkt.size |
|
|
| 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(): |
| if frame.pts is None: |
| continue |
| for out_pkt in active_audio_out_stream.encode(frame): |
| active_output_container.mux(out_pkt) |
| with lock: |
| session['bytes_sent'] += out_pkt.size |
|
|
| append_user_live_log(chat_id, f"Finished input: {current_input_url}") |
|
|
| except Exception as e_input: |
| tb = traceback.format_exc() |
| err_msg = str(e_input) |
| |
| 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[: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', 3) |
| stop_on_err = session.get('stop_on_error_in_playlist', True) |
|
|
| if should_stop_now: |
| break |
|
|
| if reconnect_on and cur_attempt < max_att: |
| with lock: |
| session['reconnect_attempt'] = cur_attempt + 1 |
| old_st = session['streaming_state'] |
| 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)}") |
| |
| if _input_container: |
| try: _input_container.close() |
| except Exception: pass |
| |
| 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 |
| |
| continue |
| elif stop_on_err: |
| append_user_live_log(chat_id, "Stopping due to error (stop_on_error=True).") |
| with lock: |
| session['streaming_state'] = "stopping" |
| break |
| else: |
| append_user_live_log(chat_id, "Skipping to next playlist item.") |
|
|
| finally: |
| if _input_container: |
| try: |
| _input_container.close() |
| except Exception: |
| pass |
| with lock: |
| if session['pyav_objects'].get('input_container') is _input_container: |
| session['pyav_objects']['input_container'] = None |
|
|
| with lock: |
| if session['streaming_state'] == "stopping": |
| break |
| session['current_playlist_index'] += 1 |
|
|
| with lock: |
| if session['streaming_state'] == "stopping": |
| break |
| session['current_loop_iteration'] += 1 |
|
|
| append_user_live_log(chat_id, "All items/loops processed or stopped.") |
|
|
| except Exception as e_fatal: |
| tb = traceback.format_exc() |
| append_user_live_log(chat_id, f"Fatal error: {e_fatal}") |
| logger.error(f"[Chat {chat_id}] Fatal stream error: {e_fatal}\n{tb}") |
| with lock: |
| session['error_notification_user'] = str(e_fatal) |
| session['streaming_state'] = "error" |
| notify_state_change(chat_id, "error", f"Fatal error: {esc(str(e_fatal))}") |
|
|
| finally: |
| |
| try: |
| av.logging.restore_default_callback() |
| except Exception: |
| pass |
| _fflog_stop.set() |
|
|
| append_user_live_log(chat_id, "Finalizing stream…") |
|
|
| |
| _final_output = None |
| _final_vid = None |
| _final_aud = None |
| with lock: |
| objs = session.get('pyav_objects', {}) |
| _final_output = objs.get('output_container') |
| _final_vid = objs.get('video_out_stream') |
| _final_aud = objs.get('audio_out_stream') |
|
|
| if _final_output: |
| append_user_live_log(chat_id, "Flushing encoders…") |
| for stream in [_final_vid, _final_aud]: |
| if stream: |
| try: |
| for pkt in stream.encode(): |
| try: |
| _final_output.mux(pkt) |
| except Exception: |
| pass |
| except Exception as fe: |
| append_user_live_log(chat_id, f"Flush error: {fe}") |
|
|
| _cleanup_pyav() |
|
|
| with lock: |
| cur = session['streaming_state'] |
| if cur == "stopping": |
| final_state = "stopped" |
| session['error_notification_user'] = "" |
| 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, |
| "Stream completed successfully! ✅" if final_state == "completed" else |
| "Stream was stopped." if final_state == "stopped" else |
| f"Error: {esc(session.get('error_notification_user', ''))}") |
|
|
|
|
| |
| |
| |
| 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 _reply( |
| f"⚠️ Stream already active (state: <code>{state}</code>).\n" |
| f"Use /abort to stop it first.", |
| get_main_keyboard(session)) |
| if not session.get('input_url_playlist'): |
| return _reply( |
| "📜 <b>Playlist is empty.</b>\nAdd at least one URL:\n" |
| "<code>/playlist add <url></code>", |
| get_main_keyboard(session)) |
| if not session.get('output_url') or session['output_url'] == DEFAULT_USER_SETTINGS['output_url']: |
| return _reply( |
| "📡 <b>Output URL not configured.</b>\n" |
| "Set it with: <code>/set output_url rtmp://your-url/key</code>", |
| get_main_keyboard(session)) |
| session['streaming_state'] = "starting" |
| session['error_notification_user'] = "" |
|
|
| t_stream = threading.Thread(target=stream_engine_thread_target, args=(chat_id,), |
| name=f"Stream-{chat_id}", daemon=True) |
| t_watch = threading.Thread(target=watchdog_thread_target, args=(chat_id,), |
| name=f"Watchdog-{chat_id}", daemon=True) |
| with lock: |
| session['stream_thread_ref'] = t_stream |
| session['watchdog_thread_ref'] = t_watch |
|
|
| t_stream.start() |
| await asyncio.sleep(0.3) |
| t_watch.start() |
|
|
| return _reply( |
| "🔄 <b>Stream starting…</b>\n\n" |
| "I'll notify you when it's live.\n\n" |
| "<b>Commands while streaming:</b>\n" |
| " /pause — pause the stream\n" |
| " /resume — resume\n" |
| " /abort — stop the stream\n" |
| " /status — current status", |
| get_main_keyboard(session)) |
|
|
|
|
| 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: |
| if session['streaming_state'] == "streaming": |
| session['streaming_state'] = "paused" |
| ok = True |
| else: |
| 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 _reply("⏸ <b>Stream paused.</b>\nUse /resume to continue or /abort to stop.", |
| get_main_keyboard(session)) |
| else: |
| return _reply(f"ℹ️ Cannot pause — state is <code>{st}</code>.", get_main_keyboard(session)) |
|
|
|
|
| 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: |
| if session['streaming_state'] == "paused": |
| session['streaming_state'] = "streaming" |
| ok = True |
| else: |
| 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 _reply("▶️ <b>Stream resumed.</b>", get_main_keyboard(session)) |
| else: |
| return _reply(f"ℹ️ Cannot resume — state is <code>{st}</code>.", 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.") |
|
|
| |
| with lock: |
| session['streaming_state'] = "stopping" |
| session['stop_gracefully_flag'] = True |
| thread_ref = session.get('stream_thread_ref') |
|
|
| |
| if thread_ref and thread_ref.is_alive(): |
| thread_ref.join(timeout=5.0) |
|
|
| |
| 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}") |
|
|
| |
| 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) |
|
|
| |
| 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 = ( |
| "⚡ <b>Force Reboot Complete!</b>\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, message_id: int = None): |
| session = get_user_session(chat_id) |
| lock = session_locks[chat_id] |
| thread_ref = None |
|
|
| with lock: |
| 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 = 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 _reply(f"ℹ️ No active stream (state: <code>{st}</code>).", 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=10.0) |
| if thread_ref.is_alive(): |
| append_user_live_log(chat_id, "Warning: thread did not stop cleanly in 10s — forcing.") |
|
|
| with lock: |
| |
| 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'] = "" |
|
|
| _push_state_sse(chat_id) |
| return _reply("⏹ <b>Stream stopped.</b>\n\n" + compose_status_message(chat_id), |
| get_main_keyboard(session)) |
|
|
|
|
| |
| |
| |
| async def handle_playlist_command(chat_id: int, text: str): |
| session = get_user_session(chat_id) |
| lock = session_locks[chat_id] |
| parts = text.split(maxsplit=2) |
| action = parts[1].lower() if len(parts) > 1 else "show" |
|
|
| with lock: |
| playlist = session.setdefault('input_url_playlist', []) |
|
|
| if action == "add": |
| if len(parts) < 3: |
| msg = "⚠️ Usage: <code>/playlist add <url></code>" |
| else: |
| url = parts[2].strip() |
| if validate_url(url): |
| playlist.append(url) |
| msg = f"✅ Added to playlist.\n<code>{esc(url)}</code>\nTotal: <b>{len(playlist)}</b>" |
| else: |
| msg = f"❌ Invalid URL.\nMust start with http/https/rtmp/rtsp/udp/srt/file.\n<code>{esc(url)}</code>" |
|
|
| elif action == "remove": |
| if len(parts) < 3: |
| msg = "⚠️ Usage: <code>/playlist remove <index|last></code>" |
| else: |
| idx_str = parts[2].strip() |
| removed = None |
| if idx_str.lower() == "last" and playlist: |
| removed = playlist.pop() |
| elif idx_str.isdigit(): |
| i = int(idx_str) - 1 |
| if 0 <= i < len(playlist): |
| removed = playlist.pop(i) |
| if removed: |
| msg = f"🗑 Removed: <code>{esc(removed[:80])}</code>\nRemaining: {len(playlist)}" |
| else: |
| msg = "❌ Invalid index or empty playlist." |
|
|
| elif action == "clear": |
| count = len(playlist) |
| session['input_url_playlist'] = [] |
| msg = f"🗑 Playlist cleared — {count} item(s) removed." |
|
|
| elif action == "show": |
| if playlist: |
| items = "\n".join(f" <code>{i+1}.</code> {esc(url)}" for i, url in enumerate(playlist)) |
| msg = f"📜 <b>Playlist ({len(playlist)} items):</b>\n{items}" |
| else: |
| msg = ("📜 <b>Playlist is empty.</b>\n" |
| "Add a URL:\n<code>/playlist add <url></code>") |
|
|
| else: |
| msg = ("📜 <b>Playlist Commands:</b>\n" |
| " <code>/playlist add <url></code>\n" |
| " <code>/playlist remove <index|last></code>\n" |
| " <code>/playlist clear</code>\n" |
| " <code>/playlist show</code>") |
|
|
| return send_message(chat_id, msg, reply_markup=get_main_keyboard(session)) |
|
|
|
|
| |
| |
| |
| SETTABLE_FIELDS = { |
| "output_url": (validate_url, "RTMP/HTTP output URL"), |
| "video_codec": (lambda x: x in SUPPORTED_VIDEO_CODECS, f"Video codec. Options: {', '.join(SUPPORTED_VIDEO_CODECS)}"), |
| "audio_codec": (lambda x: x in SUPPORTED_AUDIO_CODECS, f"Audio codec. Options: {', '.join(SUPPORTED_AUDIO_CODECS)}"), |
| "resolution": (validate_resolution, "e.g. 1280x720 or source"), |
| "fps": (lambda x: x.isdigit() and 1 <= int(x) <= 120, "FPS (1–120)"), |
| "video_bitrate": (lambda x: True, "e.g. 1500k or 3M"), |
| "audio_bitrate": (lambda x: True, "e.g. 128k"), |
| "ffmpeg_preset": (lambda x: x in FFMPEG_PRESETS, f"Options: {', '.join(FFMPEG_PRESETS)}"), |
| "gop_size": (lambda x: x.isdigit(), "Integer, e.g. 60"), |
| "loop_count": (lambda x: x.lstrip('-').isdigit(), "0=once, -1=infinite, N=N times"), |
| "output_format": (lambda x: x in OUTPUT_FORMATS, f"Options: {', '.join(OUTPUT_FORMATS)}"), |
| "reconnect_delay_seconds": (lambda x: x.isdigit(), "Seconds between reconnect attempts"), |
| "max_reconnect_attempts": (lambda x: x.isdigit(), "Max number of reconnect attempts"), |
| "open_timeout_seconds": (lambda x: x.isdigit(), "Connection open timeout in seconds"), |
| "read_timeout_seconds": (lambda x: x.isdigit(), "Packet read timeout in seconds"), |
| "video_pix_fmt": (lambda x: True, "Pixel format, e.g. yuv420p"), |
| "video_thread_count": (lambda x: x.isdigit(), "Thread count (0=auto)"), |
| "audio_sample_rate": (lambda x: x.isdigit(), "Sample rate (0=source), e.g. 44100"), |
| "audio_channels": (lambda x: x.isdigit(), "Channels (0=source), e.g. 2"), |
| "logo_position": (lambda x: x in LOGO_POSITIONS, f"Options: {', '.join(LOGO_POSITIONS)}"), |
| "logo_scale": (lambda x: True, "0.0–1.0, e.g. 0.1"), |
| "logo_opacity": (lambda x: True, "0.0–1.0, e.g. 0.8"), |
| "logo_margin_px": (lambda x: x.isdigit(), "Margin in pixels"), |
| "logo_enabled": (lambda x: x.lower() in ("on","off","true","false","1","0"), "on/off"), |
| "stop_on_error_in_playlist":(lambda x: x.lower() in ("on","off","true","false","1","0"), "on/off"), |
| "reconnect_on_stream_error":(lambda x: x.lower() in ("on","off","true","false","1","0"), "on/off"), |
| "quality_preset": (lambda x: x in QUALITY_PRESETS, f"Options: {', '.join(QUALITY_PRESETS.keys())}"), |
| } |
|
|
| INT_FIELDS = {"fps","gop_size","loop_count","reconnect_delay_seconds","max_reconnect_attempts", |
| "open_timeout_seconds","read_timeout_seconds","video_thread_count","audio_sample_rate","audio_channels","logo_margin_px"} |
| FLOAT_FIELDS = {"logo_scale","logo_opacity"} |
| BOOL_FIELDS = {"logo_enabled","stop_on_error_in_playlist","reconnect_on_stream_error"} |
|
|
|
|
| def parse_field_value(field: str, raw: str): |
| if field in BOOL_FIELDS: |
| return raw.lower() in ("on", "true", "1") |
| if field in INT_FIELDS: |
| return int(raw) |
| if field in FLOAT_FIELDS: |
| return float(raw) |
| return raw |
|
|
|
|
| async def handle_set_command(chat_id: int, text: str): |
| session = get_user_session(chat_id) |
| lock = session_locks[chat_id] |
| parts = text.split(maxsplit=2) |
|
|
| if len(parts) == 1: |
| |
| lines = ["⚙️ <b>All configurable fields:</b>\n", |
| "Usage: <code>/set <field> <value></code>\n"] |
| for field, (_, desc) in SETTABLE_FIELDS.items(): |
| cur = session.get(field) |
| lines.append(f" <code>{field}</code> = <code>{esc(str(cur))}</code>\n ↳ {esc(desc)}") |
| return send_message(chat_id, "\n".join(lines), reply_markup=get_main_keyboard(session)) |
|
|
| field = parts[1].lower() |
| if field not in SETTABLE_FIELDS: |
| close = [f for f in SETTABLE_FIELDS if field in f] |
| hint = f"\nDid you mean: <code>{', '.join(close)}</code>?" if close else "" |
| return send_message(chat_id, |
| f"❌ Unknown field <code>{esc(field)}</code>.{hint}\n" |
| f"Use /set to see all fields.", |
| reply_markup=get_main_keyboard(session)) |
|
|
| if len(parts) < 3: |
| validator, desc = SETTABLE_FIELDS[field] |
| cur = session.get(field) |
| return send_message(chat_id, |
| f"📝 <b>Field:</b> <code>{field}</code>\n" |
| f"<b>Current value:</b> <code>{esc(str(cur))}</code>\n" |
| f"<b>Expected:</b> {esc(desc)}\n\n" |
| f"Usage: <code>/set {field} <value></code>", |
| reply_markup=get_main_keyboard(session)) |
|
|
| raw_value = parts[2].strip() |
| validator, desc = SETTABLE_FIELDS[field] |
|
|
| if not validator(raw_value): |
| return send_message(chat_id, |
| f"❌ Invalid value for <code>{field}</code>.\n" |
| f"Expected: {esc(desc)}\n" |
| f"Got: <code>{esc(raw_value)}</code>", |
| reply_markup=get_main_keyboard(session)) |
|
|
| with lock: |
| parsed = parse_field_value(field, raw_value) |
| session[field] = parsed |
|
|
| |
| if field == "quality_preset" and raw_value in QUALITY_PRESETS: |
| for k, v in QUALITY_PRESETS[raw_value].items(): |
| session[k] = v |
|
|
| append_user_live_log(chat_id, f"Set {field} = {parsed}") |
| return send_message(chat_id, |
| f"✅ <b>{field}</b> set to <code>{esc(str(parsed))}</code>", |
| reply_markup=get_main_keyboard(session)) |
|
|
|
|
| |
| |
| |
| async def handle_logo_upload(chat_id: int, message: dict): |
| session = get_user_session(chat_id) |
| lock = session_locks[chat_id] |
|
|
| file_id = None |
| filename = None |
| mime_type = "image/png" |
|
|
| if message.get("document"): |
| doc = message["document"] |
| file_id = doc.get("file_id") |
| filename = doc.get("file_name", f"logo_{chat_id}.png") |
| mime_type = doc.get("mime_type", "image/png") |
| elif message.get("photo"): |
| |
| photos = message["photo"] |
| file_id = photos[-1]["file_id"] |
| filename = f"photo_{chat_id}.jpg" |
| mime_type = "image/jpeg" |
|
|
| if not file_id: |
| return send_message(chat_id, "❌ Could not get file from message.", reply_markup=get_main_keyboard(session)) |
|
|
| |
| |
| with lock: |
| session['current_step'] = None |
| return send_message(chat_id, |
| "⚠️ <b>Logo upload is unavailable in webhook-only mode.</b>\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)) |
|
|
|
|
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
|
|
| 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 = ["🕒 <b>Scheduled Streams</b> <i>(in-memory, lost on restart)</i>\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" • <b>{esc(j.name)}</b>\n" |
| f" ⏰ <code>{rt}</code>\n" |
| f" 📡 <code>{esc(rtmp[:60])}</code>\n" |
| f" 🎬 <code>{esc(inp[:60])}</code>\n" |
| f" Cancel: <code>/schedule cancel {esc(j.id)}</code>" |
| ) |
| else: |
| lines.append(" No scheduled streams.") |
| lines += [ |
| "", |
| "➕ <b>Add a schedule:</b> use <b>🕒 Schedule</b> button or <code>/schedule new</code>", |
| "❌ <b>Cancel:</b> <code>/schedule cancel <job_id></code>", |
| ] |
| 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) |
| lock = session_locks[chat_id] |
| parts = text.split(maxsplit=2) |
| sub = parts[1].lower() if len(parts) > 1 else "" |
|
|
| |
| if not sub or sub == "list": |
| return send_message(chat_id, _list_schedules(chat_id), |
| reply_markup=get_main_keyboard(session)) |
|
|
| |
| if sub == "new": |
| with lock: |
| session["current_step"] = "sched_when" |
| session["_sched_draft"] = {} |
| return send_message(chat_id, |
| "🕒 <b>New Scheduled Stream</b>\n\nWhen should it start?", |
| reply_markup=get_schedule_when_keyboard()) |
|
|
| |
| if sub == "cancel": |
| job_id = parts[2].strip() if len(parts) > 2 else None |
| if not job_id: |
| return send_message(chat_id, |
| "Usage: <code>/schedule cancel <job_id></code>", |
| 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: <code>{esc(job_id)}</code>", |
| reply_markup=get_main_keyboard(session)) |
| except Exception: |
| return send_message(chat_id, |
| f"❌ Job not found: <code>{esc(job_id)}</code>", |
| 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)) |
|
|
|
|
| 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": |
| |
| 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 <code>30</code>, <code>2h</code>, <code>1h30m</code>\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: <b>{minutes} minute(s)</b> from now\n" |
| f" → <code>{trigger_time.strftime('%Y-%m-%d %H:%M:%S UTC')}</code>\n\n" |
| f"Give this schedule a <b>name</b> (or type <code>skip</code>):") |
|
|
| elif step == "sched_datetime": |
| raw = text.strip() |
| try: |
| |
| 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 <code>YYYY-MM-DD HH:MM:SS</code> (UTC)\nOr /cancel.") |
| if trigger_time <= datetime.datetime.now(datetime.timezone.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: <code>{trigger_time.strftime('%Y-%m-%d %H:%M:%S UTC')}</code>\n\n" |
| f"Give this schedule a <b>name</b> (or type <code>skip</code>):") |
|
|
| 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"📡 <b>Step: RTMP Output URL</b>\n\n" |
| f"Enter the RTMP URL to stream to:\n" |
| f"<code>rtmp://a.rtmp.youtube.com/live2/YOUR_KEY</code>\n\n" |
| f"Or type <code>use</code> to use your current output URL (<code>{esc(session.get('output_url', 'not set'))}</code>)\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"🎬 <b>Step: Input URL</b>\n\n" |
| f"Enter the stream/video input URL:\n" |
| f"<code>https://example.com/video.mp4</code>\n\n" |
| f"Or type <code>use</code> to use your current playlist first item (<code>{esc((session.get('input_url_playlist') or ['not set'])[0])}</code>)\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 |
|
|
| |
| 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())}" |
| 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, |
| ) |
| |
| _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"✅ <b>Stream Scheduled!</b>\n\n" |
| f" 📛 <b>Name:</b> {esc(job_name)}\n" |
| f" ⏰ <b>Time:</b> <code>{trigger_time.strftime('%Y-%m-%d %H:%M:%S UTC')}</code> (~{mins_away}m from now)\n" |
| f" 📡 <b>Output:</b> <code>{esc(output_url[:60])}</code>\n" |
| f" 🎬 <b>Input:</b> <code>{esc(input_url[:60])}</code>\n" |
| f" 🔑 <b>ID:</b> <code>{job_id}</code>\n\n" |
| f"View all: /schedule\nCancel: <code>/schedule cancel {job_id}</code>", |
| reply_markup=get_main_keyboard(session)) |
|
|
| |
| 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, 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] |
|
|
| |
| 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] |
|
|
| |
| msg_text = ( |
| f"⏰ <b>Scheduled stream is starting!</b>\n\n" |
| f"📡 Output: <code>{output_url or session.get('output_url', '—')}</code>\n" |
| f"🎬 Input: <code>{input_url or (session.get('input_url_playlist') or ['—'])[0]}</code>\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_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) |
| else: |
| asyncio.run(start_stream_handler(chat_id)) |
|
|
|
|
| |
| |
| |
| def get_help_text() -> str: |
| return ( |
| f"<b>🤖 Advanced Stream Bot v{APP_VERSION}</b>\n" |
| "<i>In-memory: settings lost on restart</i>\n\n" |
| "<b>━━ Core Commands ━━</b>\n" |
| " /start — Home screen & status\n" |
| " /stream — Start streaming\n" |
| " /pause — Pause stream\n" |
| " /resume — Resume stream\n" |
| " /abort — Stop stream\n" |
| " /status — Detailed status\n" |
| " /reboot — ⚡ Force Reboot (last resort if bot stops responding)\n\n" |
| "<b>━━ Configuration ━━</b>\n" |
| " /set — List all settings\n" |
| " /set <field> <value> — Set a value\n" |
| " /settings — Interactive settings menu\n" |
| " /reset — Restore all defaults\n\n" |
| "<b>━━ Playlist ━━</b>\n" |
| " /playlist show\n" |
| " /playlist add <url>\n" |
| " /playlist remove <index|last>\n" |
| " /playlist clear\n\n" |
| "<b>━━ Logo ━━</b>\n" |
| " /set_logo — Upload logo (PNG/JPG)\n" |
| " /set logo_enabled on|off\n" |
| " /set logo_position top_right|...\n" |
| " /set logo_scale 0.1\n" |
| " /set logo_opacity 0.8\n\n" |
| "<b>━━ Schedule ━━</b>\n" |
| " /schedule — List & manage schedules\n" |
| " /schedule new — Create a new schedule (conversation)\n" |
| " /schedule cancel <id> — Cancel a scheduled stream\n\n" |
| "<b>📅 Scheduling supports:</b>\n" |
| " ⏱ Timer: <i>in X minutes/hours</i>\n" |
| " 📅 Date & time: <code>YYYY-MM-DD HH:MM:SS</code> (UTC)\n" |
| " Each schedule has its own RTMP output + input URL\n\n" |
| "<b>━━ Diagnostics ━━</b>\n" |
| " /logs — Recent user logs\n" |
| " /globallogs — System logs\n\n" |
| "<b>━━ Emergency ━━</b>\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" |
| "<b>⚠️ Data is in-memory only — lost on restart.</b>" |
| ) |
|
|
|
|
| |
| |
| |
| async def handle_telegram_update(update: dict): |
| try: |
| return await _handle_update_inner(update) |
| except Exception as e: |
| logger.error(f"Unhandled update error: {e}", exc_info=True) |
| return {"status": "ok"} |
|
|
|
|
| async def _handle_update_inner(update: dict): |
| |
| if "message" in update: |
| msg = update["message"] |
| chat_id = msg["chat"]["id"] |
| session = get_user_session(chat_id) |
| lock = session_locks[chat_id] |
| text = msg.get("text", "") |
|
|
| |
| if msg.get("photo") or msg.get("document"): |
| if session.get("current_step") == "awaiting_logo": |
| return await handle_logo_upload(chat_id, msg) |
| else: |
| return send_message(chat_id, |
| "📎 Got a file — but I wasn't expecting one.\n" |
| "Use /set_logo first if you want to upload a logo.", |
| reply_markup=get_main_keyboard(session)) |
|
|
| if not text: |
| return {"status": "ok"} |
|
|
| command = text.split()[0].lower() if text.startswith('/') else None |
|
|
| |
| if text.strip().lower() == "/cancel": |
| with lock: |
| session['current_step'] = None |
| session['current_step_index'] = 0 |
| session['conversation_fields_list'] = [] |
| session['settings_editing_field'] = None |
| return send_message(chat_id, "✅ Cancelled.", reply_markup=get_main_keyboard(session)) |
|
|
| |
| if session.get('current_step') and not command: |
| return await handle_conversation_input(chat_id, text) |
|
|
| |
| if command: |
| logger.info(f"[Chat {chat_id}] Command: {text[:80]}") |
|
|
| if command in ("/start", "/menu", "/home"): |
| with lock: |
| session['current_step'] = None |
| return send_message(chat_id, compose_status_message(chat_id, include_config=True), |
| reply_markup=get_main_keyboard(session)) |
|
|
| elif command == "/help": |
| return send_message(chat_id, get_help_text(), reply_markup=get_main_keyboard(session)) |
|
|
| elif command == "/stream": |
| return await start_stream_handler(chat_id) |
|
|
| elif command == "/pause": |
| return await pause_stream_handler(chat_id) |
|
|
| elif command == "/resume": |
| return await resume_stream_handler(chat_id) |
|
|
| 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)) |
|
|
| elif command == "/settings": |
| return send_message(chat_id, |
| "⚙️ <b>Settings</b>\n\n" |
| + format_settings_display(session) + "\n\n" |
| "<b>To change a setting:</b>\n" |
| "<code>/set <field> <value></code>\n\n" |
| "Or use the buttons below to navigate interactively.", |
| reply_markup=get_settings_keyboard(session)) |
|
|
| elif command == "/set": |
| return await handle_set_command(chat_id, text) |
|
|
| elif command == "/reset": |
| return send_message(chat_id, |
| "⚠️ <b>Reset all settings to defaults?</b>\nThis cannot be undone.", |
| reply_markup=get_reset_confirm_keyboard()) |
|
|
| elif command == "/playlist": |
| return await handle_playlist_command(chat_id, text) |
|
|
| elif command == "/set_logo": |
| with lock: |
| session['current_step'] = "awaiting_logo" |
| return send_message(chat_id, |
| "🖼 <b>Upload Logo</b>\n\n" |
| "Send a PNG or JPG image now.\n" |
| "<i>Max 5 MB. /cancel to abort.</i>", |
| reply_markup=get_main_keyboard(session)) |
|
|
| elif command == "/schedule": |
| return await handle_schedule_command(chat_id, text) |
|
|
| elif command == "/logs": |
| |
| logs = session.get('live_log_lines_user', []) |
| log_tail = "\n".join(logs[-30:]) if logs else "No logs yet." |
| return send_message(chat_id, |
| f"📋 <b>Stream Logs (last 30):</b>\n<pre>{esc(log_tail)}</pre>\n\n" |
| f"<i>💡 Tap the 📋 Logs button below for a live-updating view.</i>", |
| reply_markup=get_main_keyboard(session)) |
|
|
| elif command == "/globallogs": |
| last = "\n".join(live_log_lines_global[-20:]) |
| return send_message(chat_id, |
| "📜 <b>Global Bot Logs (last 20):</b>\n<pre>" + esc(last) + "</pre>", |
| reply_markup=get_main_keyboard(session)) |
|
|
| else: |
| |
| if session.get('current_step'): |
| return await handle_conversation_input(chat_id, text) |
| return send_message(chat_id, |
| f"❓ Unknown command: <code>{esc(text.split()[0])}</code>\nUse /help for available commands.", |
| reply_markup=get_main_keyboard(session)) |
|
|
| else: |
| |
| if session.get('current_step'): |
| return await handle_conversation_input(chat_id, text) |
| return send_message(chat_id, |
| "💬 Type a <b>command</b> or tap a button below.\n/help — see all commands.", |
| reply_markup=get_main_keyboard(session)) |
|
|
| |
| elif "callback_query" in update: |
| cq = update["callback_query"] |
| chat_id = cq["message"]["chat"]["id"] |
| message_id = cq["message"]["message_id"] |
| data = cq["data"] |
| session = get_user_session(chat_id) |
| lock = session_locks[chat_id] |
|
|
| logger.info(f"[Chat {chat_id}] Callback: {data}") |
| ack = answer_callback_query(cq["id"]) |
|
|
| |
| |
| |
| 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)) |
|
|
| |
| |
| if data == "stream_start": |
| 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": |
| mid = message_id if ux == "edit" else None |
| return [ack, await pause_stream_handler(chat_id, message_id=mid)] |
| elif data == "stream_resume": |
| mid = message_id if ux == "edit" else None |
| return [ack, await resume_stream_handler(chat_id, message_id=mid)] |
| elif data == "stream_abort": |
| 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": |
| |
| |
| 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)] |
|
|
| |
| |
| elif data == "lv_pause": |
| set_live_view_show(chat_id, False) |
| |
| 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": |
| |
| _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, edit("⏳ <b>Will stop after current loop finishes.</b>\n\n" |
| + compose_status_message(chat_id))] |
|
|
| |
| elif data == "open_settings": |
| return [ack, edit( |
| "⚙️ <b>Settings</b>\n\n" + format_settings_display(session) + "\n\n" |
| "Tap a parameter to change it, or use <code>/set <field> <value></code>", |
| 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"✅ <b>UX Mode switched to:</b> <code>{label}</code>\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, reply(compose_status_message(chat_id, True))] |
|
|
| elif data == "pick_quality_preset": |
| return [ack, edit("🎨 <b>Choose Quality Preset:</b>", get_quality_keyboard())] |
|
|
| elif data.startswith("apply_quality_"): |
| q = data.replace("apply_quality_", "") |
| if q in QUALITY_PRESETS: |
| with lock: |
| session['quality_preset'] = q |
| for k, v in QUALITY_PRESETS[q].items(): |
| session[k] = v |
| return [ack, edit( |
| f"✅ Quality preset <code>{q}</code> applied.\n\n" + format_settings_display(session), |
| get_settings_keyboard(session))] |
|
|
| elif data == "set_video_codec": |
| return [ack, edit("🎥 <b>Choose Video Codec:</b>", get_codec_keyboard("video"))] |
|
|
| elif data.startswith("set_vcodec_"): |
| codec = data.replace("set_vcodec_", "") |
| with lock: session['video_codec'] = codec |
| return [ack, edit(f"✅ Video codec: <code>{codec}</code>", get_settings_keyboard(session))] |
|
|
| elif data == "set_audio_codec": |
| return [ack, edit("🔊 <b>Choose Audio Codec:</b>", get_codec_keyboard("audio"))] |
|
|
| elif data.startswith("set_acodec_"): |
| codec = data.replace("set_acodec_", "") |
| with lock: session['audio_codec'] = codec |
| return [ack, edit(f"✅ Audio codec: <code>{codec}</code>", get_settings_keyboard(session))] |
|
|
| elif data == "set_ffmpeg_preset": |
| return [ack, edit("⚡ <b>Choose FFmpeg Preset:</b>", get_preset_keyboard())] |
|
|
| elif data.startswith("set_preset_"): |
| preset = data.replace("set_preset_", "") |
| with lock: session['ffmpeg_preset'] = preset |
| return [ack, edit(f"✅ Preset: <code>{preset}</code>", get_settings_keyboard(session))] |
|
|
| |
| elif data == "pick_resolution": |
| return [ack, edit("📐 <b>Choose Resolution:</b>", 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( |
| "📐 <b>Custom Resolution</b>\n" |
| "Current: <code>" + esc(str(session.get("resolution"))) + "</code>\n" |
| "Format: <code>WIDTHxHEIGHT</code> e.g. <code>1280x720</code> or <code>source</code>\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 <b>{label}</b> (<code>{val}</code>)", |
| get_settings_keyboard(session))] |
|
|
| |
| 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_fps": "fps", |
| "set_video_bitrate": "video_bitrate", |
| "set_audio_bitrate": "audio_bitrate", |
| "set_loop_count": "loop_count", |
| "set_gop_size": "gop_size", |
| "set_output_format": "output_format", |
| "set_reconnect_delay": "reconnect_delay_seconds", |
| "set_open_timeout": "open_timeout_seconds", |
| } |
| field = field_map[data] |
| _, desc = SETTABLE_FIELDS[field] |
| cur = session.get(field) |
| with lock: |
| session['current_step'] = "editing_field" |
| session['settings_editing_field'] = field |
| return [ack, reply( |
| f"📝 <b>Set {field}</b>\n" |
| f"Current: <code>{esc(str(cur))}</code>\n" |
| f"Expected: {esc(desc)}\n\n" |
| f"Type the new value now, or /cancel to abort.")] |
|
|
| elif data in ("toggle_reconnect", "toggle_stop_on_error", "toggle_verbose_log"): |
| field_map = { |
| "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, default_map[field]) |
| new_val = session[field] |
| 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"✅ <b>{labels[field]}</b> → <code>{'on' if new_val else 'off'}</code>", |
| get_settings_keyboard(session))] |
|
|
| |
| elif data == "view_playlist": |
| pl = session.get('input_url_playlist', []) |
| if pl: |
| items = "\n".join(f" <code>{i+1}.</code> {esc(url)}" for i, url in enumerate(pl)) |
| msg = (f"📜 <b>Playlist ({len(pl)} items):</b>\n{items}\n\n" |
| "<b>Manage:</b>\n" |
| " <code>/playlist add <url></code>\n" |
| " <code>/playlist remove <index|last></code>\n" |
| " <code>/playlist clear</code>") |
| else: |
| msg = ("📜 <b>Playlist is empty.</b>\n" |
| "<code>/playlist add <url></code>") |
| return [ack, reply(msg)] |
|
|
| |
| elif data == "cfg_logo": |
| fname = session.get('logo_original_filename', 'N/A') |
| enabled = session.get('logo_enabled', False) |
| has_logo = bool(session.get('logo_data_bytes')) |
| logo_btns = [] |
| if has_logo: |
| logo_btns.append([{"text": ("✅ Enabled — Disable" if enabled else "❌ Disabled — Enable"), |
| "callback_data": "toggle_logo"}]) |
| logo_btns.append([{"text": "📍 Change Position", "callback_data": "change_logo_pos"}]) |
| logo_btns.append([{"text": "🔄 Replace Logo", "callback_data": "upload_new_logo"}]) |
| else: |
| logo_btns.append([{"text": "📤 Upload Logo", "callback_data": "upload_new_logo"}]) |
| logo_btns.append([{"text": "↩ Back", "callback_data": "open_settings"}]) |
|
|
| lines = ["🖼 <b>Logo Configuration</b>\n"] |
| if has_logo: |
| lines.append(f"File: <code>{esc(fname)}</code>") |
| lines.append(f"Status: {'✅ Enabled' if enabled else '❌ Disabled'}") |
| lines.append(f"Position: <code>{session.get('logo_position')}</code>") |
| lines.append(f"Scale: <code>{session.get('logo_scale')}</code>") |
| lines.append(f"Opacity: <code>{session.get('logo_opacity')}</code>\n") |
| lines.append("Change scale/opacity: <code>/set logo_scale 0.15</code>") |
| else: |
| lines.append("No logo uploaded.") |
| 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, edit(f"🖼 Logo {'enabled ✅' if v else 'disabled ❌'}.")] |
|
|
| elif data == "change_logo_pos": |
| return [ack, edit("📍 <b>Choose logo position:</b>", get_logo_pos_keyboard())] |
|
|
| elif data.startswith("set_logo_pos_"): |
| pos = data.replace("set_logo_pos_", "") |
| with lock: session['logo_position'] = pos |
| return [ack, edit(f"✅ Logo position: <code>{pos}</code>")] |
|
|
| elif data == "upload_new_logo": |
| with lock: session['current_step'] = "awaiting_logo" |
| return [ack, reply("🖼 Send a PNG or JPG image now. /cancel to abort.")] |
|
|
| |
| elif data == "cfg_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( |
| "🕒 <b>New Scheduled Stream</b>\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 <b>Timer Setup</b>\n\n" |
| "How long from now? Type a duration:\n" |
| " <code>30</code> \u2192 30 minutes\n" |
| " <code>2h</code> \u2192 2 hours\n" |
| " <code>1h30m</code> \u2192 1h 30min\n\n" |
| "Or /cancel.")] |
|
|
| elif data == "sched_pick_datetime": |
| with lock: |
| session["current_step"] = "sched_datetime" |
| return [ack, reply( |
| "\U0001f4c5 <b>Date & Time (UTC)</b>\n\n" |
| "Enter the start time:\n" |
| "<code>YYYY-MM-DD HH:MM:SS</code>\n" |
| "Example: <code>2025-12-31 23:55:00</code>\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))] |
|
|
| |
| elif data == "quick_setup": |
| with lock: |
| session['current_step'] = "quick_output_url" |
| return [ack, reply( |
| "⚙️ <b>Quick Setup</b>\n\n" |
| "Step 1/2: Enter your <b>RTMP Output URL</b>:\n" |
| "<code>rtmp://a.rtmp.youtube.com/live2/YOUR_STREAM_KEY</code>\n\n" |
| "Or /cancel to abort.")] |
|
|
| |
| elif data == "confirm_reset": |
| return [ack, edit("⚠️ <b>Reset all settings to defaults?</b>", |
| get_reset_confirm_keyboard())] |
|
|
| elif data == "do_reset": |
| reset_session_settings(chat_id) |
| return [ack, reply("🔄 <b>Settings restored to defaults.</b>\n\n" |
| + compose_status_message(chat_id, True))] |
|
|
| |
| elif data == "show_user_logs": |
| 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, reply(get_help_text())] |
|
|
| return [ack, answer_callback_query(cq["id"], "Unknown action", show_alert=True)] |
|
|
| return {"status": "ok"} |
|
|
|
|
| |
| |
| |
| async def handle_conversation_input(chat_id: int, text: str): |
| session = get_user_session(chat_id) |
| lock = session_locks[chat_id] |
| step = session.get('current_step') |
|
|
| if step == "editing_field": |
| field = session.get('settings_editing_field') |
| if not field or field not in SETTABLE_FIELDS: |
| with lock: |
| session['current_step'] = None |
| return send_message(chat_id, "⚠️ State error — cancelled.", reply_markup=get_main_keyboard(session)) |
|
|
| validator, desc = SETTABLE_FIELDS[field] |
| raw = text.strip() |
| if not validator(raw): |
| return send_message(chat_id, |
| f"❌ Invalid value for <code>{field}</code>.\nExpected: {esc(desc)}\nTry again or /cancel.") |
|
|
| with lock: |
| session[field] = parse_field_value(field, raw) |
| if field == "quality_preset" and raw in QUALITY_PRESETS: |
| for k, v in QUALITY_PRESETS[raw].items(): |
| session[k] = v |
| session['current_step'] = None |
| session['settings_editing_field'] = None |
|
|
| append_user_live_log(chat_id, f"Set {field} = {raw}") |
| return send_message(chat_id, |
| f"✅ <b>{field}</b> set to <code>{esc(raw)}</code>\n\n" + format_settings_display(session), |
| reply_markup=get_settings_keyboard(session)) |
|
|
| elif step == "quick_output_url": |
| url = text.strip() |
| if not validate_url(url): |
| return send_message(chat_id, |
| "❌ Invalid URL. Must start with rtmp/http/https/rtsp/...\nTry again or /cancel.") |
| with lock: |
| session['output_url'] = url |
| session['current_step'] = "quick_input_url" |
| return send_message(chat_id, |
| f"✅ Output URL set.\n\n" |
| f"Step 2/2: Enter your <b>first Input URL</b> (stream/video URL):\n" |
| f"Or /cancel to abort.") |
|
|
| elif step == "quick_input_url": |
| url = text.strip() |
| if not validate_url(url): |
| return send_message(chat_id, |
| "❌ Invalid URL. Must start with http/https/rtmp/rtsp/...\nTry again or /cancel.") |
| with lock: |
| session['input_url_playlist'] = [url] |
| session['current_step'] = None |
| return send_message(chat_id, |
| f"✅ <b>Quick Setup Complete!</b>\n\n" |
| f"Input URL added to playlist.\n\n" |
| f"You're ready to stream! Use /stream to start.\n\n" |
| + compose_status_message(chat_id, True), |
| reply_markup=get_main_keyboard(session)) |
|
|
| elif step == "awaiting_logo": |
| return send_message(chat_id, |
| "🖼 Please <b>send an image file</b> (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 |
| return send_message(chat_id, |
| "💬 No active setup. Use a command or button below.", |
| reply_markup=get_main_keyboard(session)) |
|
|
|
|
| |
| |
| |
| @app.on_event("startup") |
| async def startup_event(): |
| global _main_event_loop |
| _main_event_loop = asyncio.get_event_loop() |
| if not scheduler.running: |
| scheduler.start() |
| logger.info("APScheduler started (MemoryJobStore).") |
| logger.info(f"Stream Bot v{APP_VERSION} started.") |
| logger.warning("All data is in-memory — lost on restart.") |
|
|
|
|
| @app.on_event("shutdown") |
| async def shutdown_event(): |
| if scheduler.running: |
| scheduler.shutdown() |
| logger.info("Stream Bot shutdown.") |
|
|
|
|
| @app.post("/webhook") |
| async def telegram_webhook_endpoint(request: Request): |
| try: |
| update = await request.json() |
|
|
| |
| 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) |
|
|
| |
| primary = None |
| if isinstance(response_data, list): |
| 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 |
|
|
| |
| |
| |
| 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] |
| |
| for m in queued_msgs[1:]: |
| enqueue_outbound_message(chat_id, m) |
| else: |
| |
| for m in queued_msgs: |
| enqueue_outbound_message(chat_id, m) |
|
|
| return primary or {"status": "ok"} |
|
|
| except json.JSONDecodeError: |
| raise HTTPException(status_code=400, detail="Invalid JSON") |
| except Exception as e: |
| logger.error(f"Webhook error: {e}", exc_info=True) |
| 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: |
| |
| get_user_session(chat_id) |
|
|
| q: asyncio.Queue = asyncio.Queue(maxsize=200) |
| _register_sse_subscriber(chat_id, q) |
|
|
| async def event_generator(): |
| try: |
| |
| _push_state_sse(chat_id) |
| |
| 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: |
| |
| if await request.is_disconnected(): |
| break |
| try: |
| |
| payload = await asyncio.wait_for(q.get(), timeout=3.0) |
| yield f"data: {payload}\n\n" |
| except asyncio.TimeoutError: |
| |
| 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: |
| |
| 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) |
| |
| 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 "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", |
| "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()), |
| } |
|
|
|
|
| @app.get("/health") |
| async def health(): |
| return { |
| "status": "ok", |
| "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<TOKEN>/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") |
|
|
| |
| _do_live_view_edit(chat_id, force=True) |
|
|
| with _outbound_queue_lock: |
| msgs = _outbound_message_queue.pop(chat_id, []) |
|
|
| |
| edit_msg = None |
| for m in reversed(msgs): |
| if m.get("method") == "editMessageText": |
| edit_msg = m |
| break |
| |
| 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(), |
| } |
|
|
|
|
| |
| |
| |
| if __name__ == "__main__": |
| import uvicorn |
| logger.info(f"Starting Advanced Stream Bot v{APP_VERSION}...") |
| uvicorn.run("stream_bot:app", host="0.0.0.0", port=8000, reload=False) |
