Update app.py

app.py

@@ -4,7 +4,6 @@ import time
 import datetime
 import traceback
 import fractions
-import collections
 
 from fastapi import FastAPI, Request
 import av
@@ -12,7 +11,7 @@ import av
 app = FastAPI()
 
 # -------------------------------------------------------------------
-# CONFIGURATION & GLOBAL VARIABLES
+# Configuration & Global Variables
 # -------------------------------------------------------------------
 # (No token or outgoing HTTP calls are used in this version.)
 # Conversation state
@@ -33,7 +32,7 @@ default_settings = {
 }
 
 # Streaming state & statistics
-streaming_state = "idle"  # Possible values: "idle", "streaming", "paused", "stopped"
+streaming_state = "idle"  # "idle", "streaming", "paused", "stopped"
 stream_chat_id = None     # Chat ID for periodic updates
 stream_start_time = None
 frames_encoded = 0
@@ -48,13 +47,13 @@ output_stream = None
 stream_thread = None
 live_log_thread = None
 
-# Live logging globals using collections.deque for efficiency
-live_log_lines = collections.deque(maxlen=50)  # Efficient log storage (max 50 lines)
-live_log_display = ""   # Updated by the live log updater thread
-error_notification = "" # Holds error details if any error occurs
+# Live logging globals
+live_log_lines = []     # Rolling list (max 50 log lines)
+live_log_display = ""   # Global variable updated every second by live_log_updater
+error_notification = "" # Global variable to hold error details if any
 
 # -------------------------------------------------------------------
-# ENHANCED LOGGING SETUP
+# Enhanced Logging Setup
 # -------------------------------------------------------------------
 logging.basicConfig(
     level=logging.DEBUG,
@@ -62,33 +61,31 @@ logging.basicConfig(
     datefmt="%Y-%m-%d %H:%M:%S",
 )
 logger = logging.getLogger()
-logger.setLevel(logging.DEBUG)
 
 def append_live_log(line: str):
-    """ Append a log line to the global deque. """
+    global live_log_lines
     live_log_lines.append(line)
+    if len(live_log_lines) > 50:
+        live_log_lines.pop(0)
 
 class ListHandler(logging.Handler):
-    """ Custom logging handler to store logs in the deque for live display. """
     def emit(self, record):
         log_entry = self.format(record)
         append_live_log(log_entry)
 
-# Attach custom logging handler
 list_handler = ListHandler()
-list_handler.setLevel(logging.DEBUG)
-list_handler.setFormatter(logging.Formatter("%Y-%m-%d %H:%M:%S [%(levelname)s] %(message)s"))
+list_handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s", "%Y-%m-%d %H:%M:%S"))
 logger.addHandler(list_handler)
 
 # -------------------------------------------------------------------
-# UTILITY FUNCTIONS & UI HELPERS
+# Utility Functions & UI Helpers
 # -------------------------------------------------------------------
 def create_html_message(text: str):
-    """ Returns a message dictionary with HTML formatting. """
+    # Wrap text in <pre> tags for monospaced output using HTML parse mode
     return {"parse_mode": "HTML", "text": f"<pre>{text}</pre>"}
 
 def get_inline_keyboard_for_stream():
-    """ Returns the inline keyboard for stream controls. """
+    # Inline keyboard for streaming controls (when streaming is active)
     keyboard = {
         "inline_keyboard": [
             [
@@ -104,7 +101,7 @@ def get_inline_keyboard_for_stream():
     return keyboard
 
 def get_inline_keyboard_for_start():
-    """ Returns the inline keyboard with a Start Streaming button. """
+    # Inline keyboard with a start button (when inputs are complete)
     keyboard = {
         "inline_keyboard": [
             [
@@ -115,13 +112,12 @@ def get_inline_keyboard_for_start():
     return keyboard
 
 def help_text():
-    """ Returns the help text message. """
     return (
         "*Stream Bot Help*\n\n"
         "*/start* - Begin setup for streaming (simple mode: only Input & Output URL)\n"
         "*/setting* - Enter advanced settings (Input URL, Quality Settings, Video Codec, Audio Codec, Output URL)\n"
         "*/help* - Display this help text\n"
-        "*/logs* - Show the live log display\n\n"
+        "*/logs* - Show the live log history\n\n"
         "After inputs are collected, press the inline *Start Streaming* button.\n\n"
         "While streaming, you can use inline buttons or commands:\n"
         "*/pause* - Pause the stream\n"
@@ -131,7 +127,7 @@ def help_text():
     )
 
 def send_guide_message(chat_id, message):
-    """ Returns a response dictionary for sending a guide message. """
+    # Return a response dictionary to be sent as the webhook reply
     logging.info(f"Sending message to chat {chat_id}: {message}")
     return {
         "method": "sendMessage",
@@ -141,58 +137,55 @@ def send_guide_message(chat_id, message):
     }
 
 def reset_statistics():
-    """ Resets streaming statistics. """
     global stream_start_time, frames_encoded, bytes_sent
     stream_start_time = datetime.datetime.now()
     frames_encoded = 0
     bytes_sent = 0
 
 def get_uptime():
-    """ Returns the uptime as a formatted string. """
     if stream_start_time:
         uptime = datetime.datetime.now() - stream_start_time
         return str(uptime).split('.')[0]
     return "0"
 
 def validate_inputs():
-    """ Validates that all required conversation fields have been provided. """
+    # Ensure all fields in conversation_fields have been provided
     missing = [field for field in conversation_fields if field not in user_inputs or not user_inputs[field]]
     if missing:
         return False, f"Missing fields: {', '.join(missing)}"
     return True, ""
 
 # -------------------------------------------------------------------
-# ERROR NOTIFICATION VIA GLOBAL VARIABLE
+# Error Notification Helper
 # -------------------------------------------------------------------
 def notify_error(chat_id, error_message):
-    """ Stores the error message in a global variable and logs it. """
     global error_notification
     error_notification = error_message
     logging.error(f"Error for chat {chat_id}: {error_message}")
 
 # -------------------------------------------------------------------
-# LIVE LOG UPDATER THREAD
+# Live Log Updater (Background Thread)
 # -------------------------------------------------------------------
 def live_log_updater():
-    """ Continuously updates the global live_log_display with the latest log lines. """
     global live_log_display, streaming_state
     try:
         while streaming_state in ["streaming", "paused"]:
-            # Update with all lines from the deque, joined by newlines
-            live_log_display = "<pre>" + "\n".join(list(live_log_lines)) + "</pre>"
+            # Include a header with the current update time in the log display.
+            current_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+            live_log_display = f"<pre>Live Logs (Updated at {current_time}):\n" + "\n".join(live_log_lines[-15:]) + "</pre>"
             time.sleep(1)
     except Exception as e:
-        logging.exception("Error in live log updater")
+        logging.error(f"Error in live log updater: {e}")
 
 # -------------------------------------------------------------------
-# LOGS HISTORY HANDLER (/logs)
+# Logs History Handler (/logs)
 # -------------------------------------------------------------------
 def logs_history(chat_id):
-    """ Returns the current live log display, prepending any error notification if present. """
     global live_log_display, error_notification
+    # Use the global live_log_display; if empty, show a default message.
     log_text = live_log_display if live_log_display else "<pre>No logs available yet.</pre>"
+    # If an error occurred, prepend the error notification.
     if error_notification:
-        # Prepend the error message to the log display
         if log_text.startswith("<pre>"):
             log_text = f"<pre>ERROR: {error_notification}\n\n" + log_text[5:]
         else:
@@ -205,23 +198,21 @@ def logs_history(chat_id):
     }
 
 # -------------------------------------------------------------------
-# CONVERSATION HANDLERS
+# Conversation Handlers
 # -------------------------------------------------------------------
 def handle_start(chat_id):
     global current_step, user_inputs, conversation_fields, advanced_mode
-    # Default to simple mode unless /setting has been issued
+    # By default, simple mode is used unless advanced_mode is set via /setting.
     user_inputs = {}
     if not advanced_mode:
         conversation_fields = ["input_url", "output_url"]
     else:
         conversation_fields = ["input_url", "quality_settings", "video_codec", "audio_codec", "output_url"]
     current_step = conversation_fields[0]
-    text = (
-        "👋 *Welcome to the Stream Bot!*\n\n"
-        "Let's set up your stream.\n"
-        f"Please enter the *{current_step.replace('_', ' ')}*"
-        f"{' (no default)' if current_step not in default_settings else f' _(default: {default_settings[current_step]})_'}:"
-    )
+    text = ("👋 *Welcome to the Stream Bot!*\n\n"
+            "Let's set up your stream.\n"
+            f"Please enter the *{current_step.replace('_', ' ')}*"
+            f"{' (no default)' if current_step not in default_settings else f' _(default: {default_settings[current_step]})_'}:")
     logging.info(f"/start command from chat {chat_id} (advanced_mode={advanced_mode})")
     return {
         "method": "sendMessage",
@@ -236,10 +227,8 @@ def handle_setting(chat_id):
     conversation_fields = ["input_url", "quality_settings", "video_codec", "audio_codec", "output_url"]
     user_inputs = {}
     current_step = conversation_fields[0]
-    text = (
-        "⚙️ *Advanced Mode Activated!*\n\n"
-        "Please enter the *input url*:"
-    )
+    text = ("⚙️ *Advanced Mode Activated!*\n\n"
+            "Please enter the *input url*:")
     logging.info(f"/setting command from chat {chat_id} - advanced mode enabled")
     return {
         "method": "sendMessage",
@@ -260,13 +249,13 @@ def handle_help(chat_id):
 def handle_conversation(chat_id, text):
     global current_step, user_inputs, conversation_fields
     if current_step:
+        # If the response is empty and a default exists, use the default.
         if text.strip() == "" and current_step in default_settings:
             user_inputs[current_step] = default_settings[current_step]
             logging.info(f"Using default for {current_step}: {default_settings[current_step]}")
         else:
             user_inputs[current_step] = text.strip()
             logging.info(f"Received {current_step}: {text.strip()}")
-
         idx = conversation_fields.index(current_step)
         if idx < len(conversation_fields) - 1:
             current_step = conversation_fields[idx + 1]
@@ -275,18 +264,21 @@ def handle_conversation(chat_id, text):
                 prompt += f" _(default: {default_settings[current_step]})_"
             return send_guide_message(chat_id, prompt)
         else:
+            # All inputs have been collected.
             current_step = None
             valid, msg = validate_inputs()
             if not valid:
                 return send_guide_message(chat_id, f"Validation error: {msg}")
+            # In simple mode, fill in advanced fields with defaults.
             if not advanced_mode:
                 user_inputs.setdefault("quality_settings", default_settings["quality_settings"])
                 user_inputs.setdefault("video_codec", default_settings["video_codec"])
                 user_inputs.setdefault("audio_codec", default_settings["audio_codec"])
+            # Instead of asking the user to type "start", send an inline button.
             return {
                 "method": "sendMessage",
                 "chat_id": chat_id,
-                "text": "All inputs received. Press *🚀 Start Streaming* to begin.",
+                "text": "All inputs received. Press *🚀 Start Streaming* to begin.\n\n" + live_log_display,
                 "reply_markup": get_inline_keyboard_for_start(),
                 "parse_mode": "Markdown"
             }
@@ -294,7 +286,7 @@ def handle_conversation(chat_id, text):
         return send_guide_message(chat_id, "Unrecognized input. Type /help for available commands.")
 
 # -------------------------------------------------------------------
-# BACKGROUND STREAMING FUNCTIONS
+# Background Streaming Functions
 # -------------------------------------------------------------------
 def stream_to_youtube(input_url, quality_settings, video_codec, audio_codec, output_url, chat_id):
     global video_stream, audio_stream_in, output_stream, streaming_state, frames_encoded, bytes_sent
@@ -302,17 +294,14 @@ def stream_to_youtube(input_url, quality_settings, video_codec, audio_codec, out
     try:
         streaming_state = "streaming"
         reset_statistics()
-
         input_stream = av.open(input_url)
         output_stream = av.open(output_url, mode='w', format='flv')
-
         # Configure video stream
         video_stream = output_stream.add_stream(video_codec, rate=30)
         video_stream.width = input_stream.streams.video[0].width
         video_stream.height = input_stream.streams.video[0].height
         video_stream.pix_fmt = input_stream.streams.video[0].format.name
         video_stream.codec_context.options.update({'g': '30'})
-
         if quality_settings.lower() == "high":
             video_stream.bit_rate = 3000000
             video_stream.bit_rate_tolerance = 1000000
@@ -322,24 +311,20 @@ def stream_to_youtube(input_url, quality_settings, video_codec, audio_codec, out
         elif quality_settings.lower() == "low":
             video_stream.bit_rate = 800000
             video_stream.bit_rate_tolerance = 200000
-
         # Configure audio stream
         audio_stream_in = input_stream.streams.audio[0]
         out_audio_stream = output_stream.add_stream(audio_codec, rate=audio_stream_in.rate)
         out_audio_stream.layout = "stereo"
-
         video_stream.codec_context.time_base = fractions.Fraction(1, video_stream.rate)
-
         logging.info("Streaming started successfully.")
-
-        # Start live log updater thread if not already running
+        # Start the live log updater in a background thread if not already running.
         global live_log_thread
         if live_log_thread is None or not live_log_thread.is_alive():
-            live_log_thread = threading.Thread(target=live_log_updater, daemon=True)
+            live_log_thread = threading.Thread(target=live_log_updater)
+            live_log_thread.daemon = True
             live_log_thread.start()
             logging.info("Live log updater thread started.")
-
-        # Main streaming loop
+        # Stream loop: process packets until state changes
         while streaming_state in ["streaming", "paused"]:
             for packet in input_stream.demux():
                 if streaming_state == "stopped":
@@ -363,16 +348,13 @@ def stream_to_youtube(input_url, quality_settings, video_codec, audio_codec, out
                             output_stream.mux(out_packet)
                             if hasattr(out_packet, "size"):
                                 bytes_sent += out_packet.size
-
-            # Flush any remaining packets
+            # Flush remaining packets
             for out_packet in video_stream.encode():
                 output_stream.mux(out_packet)
             for out_packet in out_audio_stream.encode():
                 output_stream.mux(out_packet)
-
             if streaming_state == "paused":
                 time.sleep(1)
-
         # Clean up resources
         try:
             video_stream.close()
@@ -381,13 +363,13 @@ def stream_to_youtube(input_url, quality_settings, video_codec, audio_codec, out
             input_stream.close()
         except Exception as cleanup_error:
             logging.error(f"Error during cleanup: {cleanup_error}")
-
         logging.info("Streaming complete, resources cleaned up.")
         streaming_state = "idle"
     except Exception as e:
         error_message = f"An error occurred during streaming: {str(e)}\n\n{traceback.format_exc()}"
         logging.error(error_message)
         streaming_state = "idle"
+        # Notify the user about the error by storing it in a global variable
         notify_error(chat_id, error_message)
 
 def start_streaming(chat_id):
@@ -395,7 +377,6 @@ def start_streaming(chat_id):
     valid, msg = validate_inputs()
     if not valid:
         return send_guide_message(chat_id, f"Validation error: {msg}")
-
     stream_chat_id = chat_id
     try:
         stream_thread = threading.Thread(
@@ -412,11 +393,12 @@ def start_streaming(chat_id):
         stream_thread.daemon = True
         stream_thread.start()
         logging.info("Streaming thread started.")
-
+        # Immediately inform the user that streaming has started, along with current live logs.
+        response_text = "🚀 *Streaming initiated!* Live logs are now updating below:\n\n" + live_log_display
         return {
             "method": "sendMessage",
             "chat_id": chat_id,
-            "text": "🚀 *Streaming initiated!* Live logs are now updating. Use the inline keyboard to control the stream.",
+            "text": response_text,
             "reply_markup": get_inline_keyboard_for_stream(),
             "parse_mode": "Markdown"
         }
@@ -427,7 +409,7 @@ def start_streaming(chat_id):
         return send_guide_message(chat_id, error_message)
 
 # -------------------------------------------------------------------
-# STREAM CONTROL HANDLERS
+# Stream Control Handlers
 # -------------------------------------------------------------------
 def pause_stream(chat_id):
     global streaming_state
@@ -484,7 +466,7 @@ def stream_status(chat_id):
     }
 
 # -------------------------------------------------------------------
-# FASTAPI WEBHOOK ENDPOINT FOR TELEGRAM UPDATES
+# FastAPI Webhook Endpoint for Telegram Updates
 # -------------------------------------------------------------------
 @app.post("/webhook")
 async def telegram_webhook(request: Request):
@@ -495,7 +477,6 @@ async def telegram_webhook(request: Request):
     if "message" in update:
         chat_id = update["message"]["chat"]["id"]
         text = update["message"].get("text", "").strip()
-
         if text.startswith("/setting"):
             return handle_setting(chat_id)
         elif text.startswith("/start"):
@@ -513,14 +494,13 @@ async def telegram_webhook(request: Request):
         elif text.startswith("/status"):
             return stream_status(chat_id)
         else:
+            # Process conversation setup inputs
             return handle_conversation(chat_id, text)
-
     # Process inline keyboard callback queries
     elif "callback_query" in update:
         callback_data = update["callback_query"]["data"]
         chat_id = update["callback_query"]["message"]["chat"]["id"]
         message_id = update["callback_query"]["message"]["message_id"]
-
         if callback_data == "pause":
             response = pause_stream(chat_id)
         elif callback_data == "resume":
@@ -533,9 +513,12 @@ async def telegram_webhook(request: Request):
             response = start_streaming(chat_id)
         else:
             response = send_guide_message(chat_id, "❓ Unknown callback command.")
-
+        # Ensure the inline keyboard remains in the updated message.
+        if streaming_state in ["streaming", "paused"]:
+            response["reply_markup"] = get_inline_keyboard_for_stream()
+        elif current_step is None:
+            response["reply_markup"] = get_inline_keyboard_for_start()
         response["method"] = "editMessageText"
         response["message_id"] = message_id
         return response
-
     return {"status": "ok"}