Update app.py

app.py

@@ -1,13 +1,16 @@
 """
-Advanced Stream Bot — v3.1.0 "Final Edition"
-- Easy navigation: commands first, inline buttons as supplements
-- Full advanced settings: every parameter configurable
-- Default/restore mode
-- Anti-freeze: stream error recovery + watchdog
-- New message on command response (no stale edits)
-- Auto stream-state change notifications
-- Context-aware keyboards (streaming vs idle)
-- Improved UX flows
+Advanced Stream Bot — v3.2.0
+- 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
@@ -34,7 +37,7 @@ from apscheduler.schedulers.background import BackgroundScheduler
 # ──────────────────────────────────────────────
 # VERSION & APP
 # ──────────────────────────────────────────────
-APP_VERSION = "3.1.0"
+APP_VERSION = "3.2.0"
 
 app = FastAPI(title="Advanced Stream Bot", version=APP_VERSION)
 scheduler = BackgroundScheduler(timezone="UTC")
@@ -482,6 +485,8 @@ 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:
@@ -509,9 +514,18 @@ def _do_live_view_edit(chat_id: int, force: bool = False):
         if chat_id in _live_views:
             _live_views[chat_id]["last_sent"] = new_text
 
-    # Enqueue editMessageText — delivered on next webhook response
-    enqueue_outbound_message(chat_id,
-        edit_message_text(chat_id, msg_id, new_text, reply_markup=kb))
+    edit_payload = edit_message_text(chat_id, msg_id, new_text, reply_markup=kb)
+
+    # Replace (not append) any existing live-view edit in the queue so we never
+    # pile up stale edits that Telegram would reject or deliver out of order.
+    with _outbound_queue_lock:
+        existing = _outbound_message_queue.get(chat_id, [])
+        # Remove any previous edit targeting the same message_id
+        filtered = [m for m in existing
+                    if not (m.get("method") == "editMessageText"
+                            and m.get("message_id") == msg_id)]
+        filtered.append(edit_payload)
+        _outbound_message_queue[chat_id] = filtered
 
     # Also push SSE snapshot for web clients
     _push_state_sse(chat_id)
@@ -521,8 +535,8 @@ def _live_view_updater_loop():
     """
     Background thread — fires every LIVE_VIEW_UPDATE_INTERVAL seconds.
     For each registered live-view, calls _do_live_view_edit().
-    Also handles auto-unregister on terminal states and SSE for active
-    sessions that don't have a live-view registered (scheduler-triggered).
+    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:
@@ -538,11 +552,12 @@ def _live_view_updater_loop():
 
                     _do_live_view_edit(chat_id)
 
-                    # Auto-unregister on terminal state (leave one final edit queued)
+                    # Auto-unregister on terminal state after one forced final update
                     if state in ('idle', 'stopped', 'completed', 'error'):
-                        # Force a final update showing terminal state, then stop
                         _do_live_view_edit(chat_id, force=True)
                         unregister_live_view(chat_id)
+                        # Push final SSE
+                        _push_state_sse(chat_id)
 
                 except Exception as e_inner:
                     logger.warning(f"Live view updater error for {chat_id}: {e_inner}")
@@ -612,6 +627,7 @@ def get_main_keyboard(session: dict):
         if session.get('loop_count', 0) == -1:
             btns.append([{"text": "⏳ Finish After This Loop", "callback_data": "stream_stop_graceful"}])
         btns.append([{"text": "❔ Help", "callback_data": "show_help"}])
+        btns.append([{"text": "⚡ Force Reboot", "callback_data": "force_reboot"}])
     else:
         # Idle mode — full navigation
         btns = []
@@ -639,6 +655,10 @@ def get_main_keyboard(session: dict):
             {"text": "🔄 Reset to Defaults", "callback_data": "confirm_reset"},
             {"text": "❔ Help", "callback_data": "show_help"},
         ])
+        # Row 5: Emergency
+        btns.append([
+            {"text": "⚡ Force Reboot", "callback_data": "force_reboot"},
+        ])
 
     return {"inline_keyboard": btns}
 
@@ -956,8 +976,9 @@ def watchdog_thread_target(chat_id: int):
         time.sleep(5)
         with lock:
             state = session.get('streaming_state')
-            if state not in ("streaming", "paused", "reconnecting"):
-                break  # Stream ended, watchdog exits
+            # Exit watchdog if stream is in any terminal or stopping state
+            if state not in ("streaming", "paused", "reconnecting", "starting"):
+                break
 
             last_frame = session.get('last_frame_time')
             thread_ref = session.get('stream_thread_ref')
@@ -966,7 +987,8 @@ def watchdog_thread_target(chat_id: int):
         if thread_ref and not thread_ref.is_alive():
             with lock:
                 cur_state = session.get('streaming_state')
-            if cur_state in ("streaming", "paused", "reconnecting"):
+            # Only mark error if not already in a clean terminal/stopping state
+            if cur_state in ("streaming", "paused", "reconnecting", "starting"):
                 logger.error(f"[Chat {chat_id}] Watchdog: stream thread died unexpectedly!")
                 append_user_live_log(chat_id, "⚠️ Stream thread died unexpectedly. Marking as error.")
                 with lock:
@@ -980,12 +1002,12 @@ def watchdog_thread_target(chat_id: int):
             stall_seconds = (datetime.datetime.now() - last_frame).total_seconds()
             if stall_seconds > STALL_THRESHOLD:
                 append_user_live_log(chat_id, f"⚠️ Watchdog: No frames for {stall_seconds:.0f}s — possible freeze!")
-                push_message(
+                enqueue_outbound_message(chat_id, send_message(
                     chat_id,
                     f"⚠️ <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.")
 
@@ -1271,10 +1293,13 @@ def stream_engine_thread_target(chat_id: int):
 
                 except Exception as e_input:
                     tb = traceback.format_exc()
-                    err_msg = f"{e_input}"
+                    err_msg = str(e_input)
+                    # Strip any log-format prefix that might have leaked in
+                    if "] " in err_msg and "[" in err_msg[:40]:
+                        err_msg = err_msg.split("] ", 1)[-1] if "] " in err_msg else err_msg
                     append_user_live_log(chat_id, f"Error on input {current_input_url}: {err_msg}")
                     with lock:
-                        session['error_notification_user'] = err_msg
+                        session['error_notification_user'] = err_msg[:300]
 
                     with lock:
                         should_stop_now = session['streaming_state'] == "stopping"
@@ -1290,7 +1315,8 @@ def stream_engine_thread_target(chat_id: int):
                         with lock:
                             session['reconnect_attempt'] = cur_attempt + 1
                             old_st = session['streaming_state']
-                            session['streaming_state'] = "reconnecting"
+                            if session['streaming_state'] not in ('stopping',):
+                                session['streaming_state'] = "reconnecting"
                         append_user_live_log(chat_id, f"Reconnecting in {reconnect_delay}s (attempt {cur_attempt+1}/{max_att})…")
                         if old_st != "reconnecting":
                             notify_state_change(chat_id, "reconnecting", f"Attempt {cur_attempt+1}/{max_att}. Error: {esc(err_msg)}")
@@ -1298,7 +1324,15 @@ def stream_engine_thread_target(chat_id: int):
                         if _input_container:
                             try: _input_container.close()
                             except Exception: pass
-                        time.sleep(reconnect_delay)
+                        # Sleep in small increments so abort flag is checked promptly
+                        for _ in range(int(reconnect_delay * 5)):
+                            with lock:
+                                if session['streaming_state'] == 'stopping':
+                                    break
+                            time.sleep(0.2)
+                        with lock:
+                            if session['streaming_state'] == 'stopping':
+                                break
                         # Retry same URL (re-enter inner loop iteration)
                         continue
                     elif stop_on_err:
@@ -1372,15 +1406,18 @@ def stream_engine_thread_target(chat_id: int):
             cur = session['streaming_state']
             if cur == "stopping":
                 final_state = "stopped"
+                session['error_notification_user'] = ""  # clear error on clean stop
             elif cur == "error":
                 final_state = "error"
             elif session.get('error_notification_user'):
                 final_state = "error"
             else:
                 final_state = "completed"
+                session['error_notification_user'] = ""
             session['streaming_state'] = final_state
             session['stream_thread_ref'] = None
             session['watchdog_thread_ref'] = None
+            session['stop_gracefully_flag'] = False
 
         append_user_live_log(chat_id, f"Stream ended. Final state: {final_state}.")
         notify_state_change(chat_id, final_state,
@@ -1493,19 +1530,87 @@ async def resume_stream_handler(chat_id: int, message_id: int = None):
         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.")
+
+    # 1. Signal any running thread to stop
+    with lock:
+        session['streaming_state'] = "stopping"
+        session['stop_gracefully_flag'] = True
+        thread_ref = session.get('stream_thread_ref')
+
+    # 2. Wait briefly for thread to die
+    if thread_ref and thread_ref.is_alive():
+        thread_ref.join(timeout=5.0)
+
+    # 3. Hard-close any open PyAV containers
+    try:
+        with lock:
+            objs = session.get('pyav_objects', {})
+            for key in ('output_container', 'input_container'):
+                c = objs.get(key)
+                if c:
+                    try: c.close()
+                    except Exception: pass
+                objs[key] = None
+            objs['video_out_stream'] = None
+            objs['audio_out_stream'] = None
+            objs['logo_image_pil'] = None
+    except Exception as e:
+        logger.warning(f"[Reboot] PyAV close error: {e}")
+
+    # 4. Reset ALL runtime state — keep user settings intact
+    with lock:
+        for k, v in DEFAULT_SESSION_RUNTIME_STATE.items():
+            session[k] = dict(v) if isinstance(v, dict) else (list(v) if isinstance(v, list) else v)
+
+    # 5. Clear queues and live view
+    unregister_live_view(chat_id)
+    with _outbound_queue_lock:
+        _outbound_message_queue.pop(chat_id, None)
+
+    _push_state_sse(chat_id)
+    append_user_live_log(chat_id, "Force Reboot complete — bot is in clean idle state.")
+
+    reply_text = (
+        "⚡ <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:
-        if session['streaming_state'] in ("streaming", "paused", "starting", "reconnecting"):
+        state = session['streaming_state']
+        if state in ("streaming", "paused", "starting", "reconnecting", "stopping"):
             session['streaming_state'] = "stopping"
+            session['stop_gracefully_flag'] = True
             thread_ref = session.get('stream_thread_ref')
             aborted = True
         else:
             aborted = False
-            st = session['streaming_state']
+            st = state
 
     def _reply(text, kb):
         if message_id:
@@ -1517,16 +1622,21 @@ async def abort_stream_handler(chat_id: int, message_id: int = None):
 
     append_user_live_log(chat_id, "Abort requested by user.")
     if thread_ref and thread_ref.is_alive():
-        thread_ref.join(timeout=8.0)
+        thread_ref.join(timeout=10.0)
         if thread_ref.is_alive():
-            append_user_live_log(chat_id, "Warning: thread did not stop cleanly.")
+            append_user_live_log(chat_id, "Warning: thread did not stop cleanly in 10s — forcing.")
 
     with lock:
-        if session['streaming_state'] == "stopping":
+        # Force stopped state regardless of what thread set it to
+        if session['streaming_state'] in ("stopping", "reconnecting", "starting", "streaming", "paused"):
             session['streaming_state'] = "stopped"
         session['stream_thread_ref'] = None
+        session['watchdog_thread_ref'] = None
+        session['stop_gracefully_flag'] = False
+        session['error_notification_user'] = ""   # clear error on clean stop
 
-    return _reply("⏹ <b>Stream aborted.</b>\n\n" + compose_status_message(chat_id),
+    _push_state_sse(chat_id)
+    return _reply("⏹ <b>Stream stopped.</b>\n\n" + compose_status_message(chat_id),
                   get_main_keyboard(session))
 
 
@@ -2027,7 +2137,8 @@ def get_help_text() -> str:
         "  /pause — Pause stream\n"
         "  /resume — Resume stream\n"
         "  /abort — Stop stream\n"
-        "  /status — Detailed status\n\n"
+        "  /status — Detailed status\n"
+        "  /reboot — ⚡ Force Reboot (last resort if bot stops responding)\n\n"
         "<b>━━ Configuration ━━</b>\n"
         "  /set — List all settings\n"
         "  /set &lt;field&gt; &lt;value&gt; — Set a value\n"
@@ -2055,6 +2166,10 @@ def get_help_text() -> str:
         "<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>"
     )
 
@@ -2132,6 +2247,9 @@ async def _handle_update_inner(update: dict):
             elif command == "/abort":
                 return await abort_stream_handler(chat_id)
 
+            elif command in ("/reboot", "/forcereboot"):
+                return await force_reboot_handler(chat_id)
+
             elif command == "/status":
                 return send_message(chat_id, compose_status_message(chat_id, include_config=True),
                                     reply_markup=get_main_keyboard(session))
@@ -2243,6 +2361,10 @@ async def _handle_update_inner(update: dict):
             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 this message as the live-view bubble (status+log mode)
             # Background thread will edit it every 2s — exactly like logg.py