Update app.py

app.py

@@ -1,5 +1,11 @@
 """
-Advanced Stream Bot — v3.2.0
+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
@@ -37,7 +43,7 @@ from apscheduler.schedulers.background import BackgroundScheduler
 # ──────────────────────────────────────────────
 # VERSION & APP
 # ──────────────────────────────────────────────
-APP_VERSION = "3.2.0"
+APP_VERSION = "3.3.0"
 
 app = FastAPI(title="Advanced Stream Bot", version=APP_VERSION)
 scheduler = BackgroundScheduler(timezone="UTC")
@@ -391,7 +397,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
@@ -870,6 +876,9 @@ def compose_status_message(chat_id: int, include_config: bool = False, include_l
 
         lines += [
             "",
+            f"⏱ <b>Uptime:</b> <code>{get_uptime(session.get('stream_start_time'))}</code>",
+            f"🎞 <b>Frames:</b> <code>{session.get('frames_encoded', 0):,}</code>",
+            f"📤 <b>Sent:</b> <code>{session.get('bytes_sent', 0) / (1024*1024):.2f} MB</code>",
             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>",
         ]
@@ -1034,6 +1043,48 @@ def stream_engine_thread_target(chat_id: int):
 
     active_output_container = None
 
+    # ── Raw FFmpeg/libav log capture via queue ────────────────────────────────
+    # PyAV's av.logging.set_log_callback installs a C-level callback. We route
+    # every libav message into a thread-safe queue and drain it in a background
+    # thread so we never call Python from a C signal context.
+    _fflog_queue: queue.Queue = queue.Queue(maxsize=500)
+    _fflog_stop = threading.Event()
+
+    def _fflog_drain():
+        _LEVELS = {0:"PANIC",8:"FATAL",16:"ERROR",24:"WARN",32:"INFO",40:"VERBOSE"}
+        _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
+            # Skip high-volume noise
+            if any(s in msg for s in ("Past duration", "deprecated pixel format",
+                                       "DTS", "PTS", "non monotonous")):
+                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"
@@ -1046,11 +1097,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']}")
 
@@ -1267,15 +1323,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():
@@ -1372,6 +1435,13 @@ def stream_engine_thread_target(chat_id: int):
         notify_state_change(chat_id, "error", f"Fatal error: {esc(str(e_fatal))}")
 
     finally:
+        # Shut down raw ffmpeg log capture
+        try:
+            av.logging.restore_default_callback()
+        except Exception:
+            pass
+        _fflog_stop.set()
+
         append_user_live_log(chat_id, "Finalizing stream…")
 
         # Flush encoders
@@ -3046,6 +3116,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
 # ────────────���─────────────────────────────────