Update app.py

app.py

@@ -1,5 +1,11 @@
 """
 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
@@ -11,9 +17,6 @@ Advanced Stream Bot — v3.3.0
 - 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
-- NEW: Raw FFmpeg/libav log lines piped into user live log (bitrate, fps, codec init, errors)
-- Fixed: stats now pushed every frame for first 300 frames, then every 100
-- Fixed: delayed live-view force-refresh 4s after stream starts so stats show non-zero
 """
 
 import logging
@@ -156,6 +159,9 @@ DEFAULT_USER_SETTINGS = {
     "logo_opacity": 0.85,
     "logo_margin_px": 10,
 
+    # Logging
+    "verbose_ffmpeg_log": False,    # show raw FFmpeg/libav lines in live log
+
     # Conversation state
     "current_step": None,
     "current_step_index": 0,
@@ -394,7 +400,7 @@ def _build_live_view_text(chat_id: int, mode: str = "status") -> str:
     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[-8:]) if logs else "No logs yet."
+        log_tail = "\n".join(logs[-15:]) if logs else "No logs yet."
         pause_note = "\n\n<i>⏸ Auto-updates paused — tap ▶️ Resume to continue</i>" if is_paused else ""
 
         # Only show log tail while actively streaming
@@ -687,6 +693,7 @@ def get_settings_keyboard(session: dict = None):
          {"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"}],
@@ -824,6 +831,7 @@ def format_settings_display(session: dict) -> str:
 
     auto_rc = "✅" if session.get('reconnect_on_stream_error') else "❌"
     stop_err = "✅" if session.get('stop_on_error_in_playlist') else "❌"
+    verbose_log = "✅ On" if session.get('verbose_ffmpeg_log') else "❌ Off"
 
     lines = [
         f"📡 <b>Output URL:</b> <code>{esc(session.get('output_url', '—'))}</code>",
@@ -838,6 +846,7 @@ def format_settings_display(session: dict) -> str:
         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}",
     ]
@@ -1040,54 +1049,50 @@ def stream_engine_thread_target(chat_id: int):
 
     active_output_container = None
 
-    # ── Raw FFmpeg/libav log capture ──────────────────────────────────────────
-    # PyAV exposes av.logging.set_log_level() and av.logging.log_callback so we
-    # can intercept every libav log line (mux stats, bitrate, fps, speed, codec
-    # messages, errors) and pipe them straight into the user's live log.
-    #
-    # Level map (libav levels):
-    #   QUIET=-8  PANIC=0  FATAL=8  ERROR=16  WARNING=24  INFO=32
-    #   VERBOSE=40  DEBUG=48  TRACE=56
-    #
-    # We use INFO (32) so we get: stream open/close, mux progress, codec init,
-    # and error messages — without drowning in debug noise.
-    _LIBAV_LEVEL_NAMES = {
-        -8: "QUIET", 0: "PANIC", 8: "FATAL", 16: "ERROR",
-        24: "WARN", 32: "INFO", 40: "VERBOSE", 48: "DEBUG", 56: "TRACE",
-    }
-    _libav_log_active = threading.Event()
-    _libav_log_active.set()
-
-    # Deduplicate repeated identical lines (libav repeats some lines many times)
-    _libav_last_line = {"v": ""}
-
-    def _libav_log_callback(level, message, _user_ptr=None):
-        if not _libav_log_active.is_set():
-            return
-        msg = message.strip()
-        if not msg:
-            return
-        # Skip pure noise lines
-        if msg in ("Past duration 0.999992 too large",) or msg.startswith("deprecated pixel format"):
-            return
-        # Collapse duplicate consecutive lines
-        if msg == _libav_last_line["v"]:
-            return
-        _libav_last_line["v"] = msg
-
-        lvl_name = _LIBAV_LEVEL_NAMES.get(level, f"L{level}")
-        # Only prefix with level tag for warnings/errors; keep info lines clean
-        if level <= 24:  # WARNING and above
-            entry = f"[ffmpeg/{lvl_name}] {msg}"
-        else:
-            entry = f"[ffmpeg] {msg}"
-        append_user_live_log(chat_id, entry)
+    # ── Raw FFmpeg/libav log capture via queue ────────────────────────────────
+    # PyAV's av.logging.set_log_callback installs a C-level callback. We route
+    # every libav message into a thread-safe queue and drain it in a background
+    # thread so we never call Python from a C signal context.
+    _fflog_queue: queue.Queue = queue.Queue(maxsize=500)
+    _fflog_stop = threading.Event()
+
+    def _fflog_drain():
+        _LEVELS = {0:"PANIC",8:"FATAL",16:"ERROR",24:"WARN",32:"INFO",40:"VERBOSE"}
+        _NOISE = ("Past duration", "deprecated pixel format", "DTS", "PTS", "non monotonous")
+        _last = ""
+        while not _fflog_stop.is_set() or not _fflog_queue.empty():
+            try:
+                level, msg = _fflog_queue.get(timeout=0.3)
+            except queue.Empty:
+                continue
+            msg = msg.strip()
+            if not msg or msg == _last:
+                continue
+            # Verbose mode off → only pass warnings/errors (level <= 24)
+            verbose_on = session.get('verbose_ffmpeg_log', False)
+            if not verbose_on and level > 24:
+                continue
+            # Skip high-volume noise regardless of verbose setting
+            if any(s in msg for s in _NOISE):
+                continue
+            _last = msg
+            lvl = _LEVELS.get(level, f"L{level}")
+            prefix = f"[ffmpeg/{lvl}]" if level <= 24 else "[ffmpeg]"
+            append_user_live_log(chat_id, f"{prefix} {msg}")
+
+    def _fflog_callback(level, message):
+        try:
+            _fflog_queue.put_nowait((level, message))
+        except queue.Full:
+            pass
 
+    _fflog_thread = threading.Thread(target=_fflog_drain, name=f"FFlogDrain-{chat_id}", daemon=True)
+    _fflog_thread.start()
     try:
-        av.logging.set_log_level(av.logging.INFO)
-        av.logging.set_log_callback(_libav_log_callback)
-    except Exception as _e_log:
-        logger.warning(f"Could not set av log callback: {_e_log}")
+        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:
@@ -1102,11 +1107,16 @@ def stream_engine_thread_target(chat_id: int):
             session['stop_gracefully_flag'] = False
             session['reconnect_attempt'] = 0
             session['last_frame_time'] = None
+            session['live_log_lines_user'] = []   # clear stale logs from previous run
             session['pyav_objects'] = {
                 "input_container": None, "output_container": None,
                 "video_out_stream": None, "audio_out_stream": None,
                 "logo_image_pil": None,
             }
+        # Also reset last_sent so the live bubble always refreshes immediately
+        with _live_view_lock:
+            if chat_id in _live_views:
+                _live_views[chat_id]["last_sent"] = ""
 
         append_user_live_log(chat_id, f"Stream engine started → {session['output_url']}")
 
@@ -1323,15 +1333,22 @@ def stream_engine_thread_target(chat_id: int):
                                     except Exception as e_logo:
                                         pass
 
+                                # Count decoded input frames immediately.
+                                # The encoder buffers many frames before emitting
+                                # the first output packet, so counting only encoded
+                                # output kept stats stuck at 0 for a long time.
+                                with lock:
+                                    session['frames_encoded'] += 1
+                                    session['last_frame_time'] = datetime.datetime.now()
+                                _fc = session['frames_encoded']
+                                # Push live stats every 30 decoded frames (~1s at 30fps)
+                                if _fc % 30 == 0:
+                                    threading.Thread(target=_push_state_sse, args=(chat_id,), daemon=True).start()
+
                                 for out_pkt in active_video_out_stream.encode(frame):
                                     active_output_container.mux(out_pkt)
                                     with lock:
-                                        session['frames_encoded'] += 1
                                         session['bytes_sent'] += out_pkt.size
-                                        session['last_frame_time'] = datetime.datetime.now()
-                                    # Push live stats to SSE subscribers every 100 frames
-                                    if session['frames_encoded'] % 100 == 0:
-                                        threading.Thread(target=_push_state_sse, args=(chat_id,), daemon=True).start()
 
                         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():
@@ -1428,12 +1445,12 @@ def stream_engine_thread_target(chat_id: int):
         notify_state_change(chat_id, "error", f"Fatal error: {esc(str(e_fatal))}")
 
     finally:
-        # Stop raw FFmpeg log capture first so no stray lines land after teardown
-        _libav_log_active.clear()
+        # Shut down raw ffmpeg log capture
         try:
             av.logging.restore_default_callback()
         except Exception:
             pass
+        _fflog_stop.set()
 
         append_user_live_log(chat_id, "Finalizing stream…")
 
@@ -2588,17 +2605,28 @@ async def _handle_update_inner(update: dict):
                 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"):
+        elif data in ("toggle_reconnect", "toggle_stop_on_error", "toggle_verbose_log"):
             field_map = {
-                "toggle_reconnect": "reconnect_on_stream_error",
+                "toggle_reconnect":    "reconnect_on_stream_error",
                 "toggle_stop_on_error": "stop_on_error_in_playlist",
+                "toggle_verbose_log":  "verbose_ffmpeg_log",
+            }
+            default_map = {
+                "reconnect_on_stream_error": True,
+                "stop_on_error_in_playlist": True,
+                "verbose_ffmpeg_log":        False,
             }
             field = field_map[data]
             with lock:
-                session[field] = not session.get(field, True)
+                session[field] = not session.get(field, default_map[field])
                 new_val = session[field]
+            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>{field}</b> → <code>{'on' if new_val else 'off'}</code>",
+                f"✅ <b>{labels[field]}</b> → <code>{'on' if new_val else 'off'}</code>",
                 get_settings_keyboard(session))]
 
         # Playlist view
@@ -3109,6 +3137,54 @@ async def get_status_endpoint(chat_id: int):
     }
 
 
+@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")
+
+    # Force-build a fresh live-view edit right now
+    _do_live_view_edit(chat_id, force=True)
+
+    with _outbound_queue_lock:
+        msgs = _outbound_message_queue.pop(chat_id, [])
+
+    # Pick the most recent editMessageText (last one wins — it's the freshest)
+    edit_msg = None
+    for m in reversed(msgs):
+        if m.get("method") == "editMessageText":
+            edit_msg = m
+            break
+    # Re-queue any non-edit messages (sendMessage notifications etc.)
+    for m in msgs:
+        if m.get("method") != "editMessageText":
+            enqueue_outbound_message(chat_id, m)
+
+    session = get_user_session(chat_id)
+    return {
+        "pending": edit_msg is not None,
+        "edit": edit_msg,
+        "status": {
+            "state": session.get("streaming_state", "idle"),
+            "frames_encoded": session.get("frames_encoded", 0),
+            "bytes_sent": session.get("bytes_sent", 0),
+            "uptime_str": get_uptime(session.get("stream_start_time")),
+            "logs": session.get("live_log_lines_user", [])[-15:],
+        },
+        "ts": datetime.datetime.now(datetime.timezone.utc).isoformat(),
+    }
+
+
 # ──────────────────────────────────────────────
 # MAIN
 # ──────────────────────────────────────────────