Update app.py

app.py

@@ -4,6 +4,7 @@ import time
 import datetime
 import traceback
 import fractions
+import collections
 
 from fastapi import FastAPI, Request
 import av
@@ -13,13 +14,10 @@ app = FastAPI()
 # -------------------------------------------------------------------
 # Configuration & Global Variables
 # -------------------------------------------------------------------
-# (No token or outgoing HTTP calls are used in this version.)
-# Conversation state
+# (No outgoing HTTP calls or token usage in this version.)
+# Conversation state variables
 user_inputs = {}
-# The conversation fields will depend on the mode.
-# Simple mode (default): Only "input_url" and "output_url" are required.
-# Advanced mode (if user sends /setting): Additional fields are required.
-conversation_fields = []
+conversation_fields = []   # Depends on mode: simple vs. advanced
 current_step = None
 advanced_mode = False
 
@@ -33,7 +31,7 @@ default_settings = {
 
 # Streaming state & statistics
 streaming_state = "idle"  # "idle", "streaming", "paused", "stopped"
-stream_chat_id = None     # Chat ID for periodic updates
+stream_chat_id = None     # Chat ID for updates
 stream_start_time = None
 frames_encoded = 0
 bytes_sent = 0
@@ -48,9 +46,9 @@ stream_thread = None
 live_log_thread = None
 
 # 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
+live_log_lines = collections.deque(maxlen=50)  # Efficient log storage
+live_log_display = ""       # Global variable updated every second by live_log_updater
+error_notification = None   # Global variable to hold error details if any
 
 # -------------------------------------------------------------------
 # Enhanced Logging Setup
@@ -61,31 +59,32 @@ logging.basicConfig(
     datefmt="%Y-%m-%d %H:%M:%S",
 )
 logger = logging.getLogger()
+logger.setLevel(logging.DEBUG)
 
 def append_live_log(line: str):
-    global live_log_lines
+    """Append a log line to the live log deque."""
     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 for live display."""
     def emit(self, record):
         log_entry = self.format(record)
         append_live_log(log_entry)
 
 list_handler = ListHandler()
-list_handler.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s", "%Y-%m-%d %H:%M:%S"))
+list_handler.setLevel(logging.DEBUG)
+list_handler.setFormatter(logging.Formatter("%Y-%m-%d %H:%M:%S [%(levelname)s] %(message)s"))
 logger.addHandler(list_handler)
 
 # -------------------------------------------------------------------
 # Utility Functions & UI Helpers
 # -------------------------------------------------------------------
 def create_html_message(text: str):
-    # Wrap text in <pre> tags for monospaced output using HTML parse mode
+    """Create an HTML-formatted message (using <pre> for monospaced text)."""
     return {"parse_mode": "HTML", "text": f"<pre>{text}</pre>"}
 
 def get_inline_keyboard_for_stream():
-    # Inline keyboard for streaming controls after the stream has started
+    """Return the inline keyboard for controlling the stream."""
     keyboard = {
         "inline_keyboard": [
             [
@@ -101,7 +100,7 @@ def get_inline_keyboard_for_stream():
     return keyboard
 
 def get_inline_keyboard_for_start():
-    # Inline keyboard with a start button for when conversation is complete.
+    """Return the inline keyboard with a Start Streaming button."""
     keyboard = {
         "inline_keyboard": [
             [
@@ -112,12 +111,13 @@ def get_inline_keyboard_for_start():
     return keyboard
 
 def help_text():
+    """Return help text for the bot."""
     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 log history (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"
@@ -127,7 +127,7 @@ def help_text():
     )
 
 def send_guide_message(chat_id, message):
-    # Return a response dictionary to be sent as the webhook reply
+    """Return a response dictionary to send a guide message."""
     logging.info(f"Sending message to chat {chat_id}: {message}")
     return {
         "method": "sendMessage",
@@ -137,19 +137,21 @@ def send_guide_message(chat_id, message):
     }
 
 def reset_statistics():
+    """Reset the 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():
+    """Return the current uptime as a string."""
     if stream_start_time:
         uptime = datetime.datetime.now() - stream_start_time
         return str(uptime).split('.')[0]
     return "0"
 
 def validate_inputs():
-    # Ensure all fields in conversation_fields have been provided
+    """Ensure all required 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)}"
@@ -159,33 +161,32 @@ def validate_inputs():
 # Error Notification Helper
 # -------------------------------------------------------------------
 def notify_error(chat_id, error_message):
+    """Store the error message globally and log it."""
     global error_notification
     error_notification = error_message
-    logging.error(f"Error for chat {chat_id}: {error_message}")
+    logger.error(f"Error for chat {chat_id}: {error_message}")
 
 # -------------------------------------------------------------------
-# Live Log Updater (Background Thread)
+# Live Log Updater Thread
 # -------------------------------------------------------------------
 def live_log_updater():
+    """Continuously update the global live_log_display variable while streaming is active."""
     global live_log_display, streaming_state
     try:
         while streaming_state in ["streaming", "paused"]:
-            # Update the global live_log_display with the last 15 log lines in HTML format
-            live_log_display = "<pre>" + "\n".join(live_log_lines[-15:]) + "</pre>"
+            live_log_display = "<pre>" + "\n".join(list(live_log_lines)) + "</pre>"
             time.sleep(1)
     except Exception as e:
-        logging.error(f"Error in live log updater: {e}")
+        logger.exception("Error in live log updater")
 
 # -------------------------------------------------------------------
 # Logs History Handler (/logs)
 # -------------------------------------------------------------------
 def logs_history(chat_id):
+    """Return the live log display (including error notification if any) as HTML."""
     global live_log_display, error_notification
-    # Use the global live_log_display if available; otherwise, 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:
-        # Remove any opening <pre> tag from log_text and prepend error message
         if log_text.startswith("<pre>"):
             log_text = f"<pre>ERROR: {error_notification}\n\n" + log_text[5:]
         else:
@@ -202,7 +203,6 @@ def logs_history(chat_id):
 # -------------------------------------------------------------------
 def handle_start(chat_id):
     global current_step, user_inputs, conversation_fields, advanced_mode
-    # By default, use simple mode unless advanced_mode is set via /setting
     user_inputs = {}
     if not advanced_mode:
         conversation_fields = ["input_url", "output_url"]
@@ -213,7 +213,7 @@ def handle_start(chat_id):
             "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})")
+    logger.info(f"/start command from chat {chat_id} (advanced_mode={advanced_mode})")
     return {
         "method": "sendMessage",
         "chat_id": chat_id,
@@ -229,7 +229,7 @@ def handle_setting(chat_id):
     current_step = conversation_fields[0]
     text = ("⚙️ *Advanced Mode Activated!*\n\n"
             "Please enter the *input url*:")
-    logging.info(f"/setting command from chat {chat_id} - advanced mode enabled")
+    logger.info(f"/setting command from chat {chat_id} - advanced mode enabled")
     return {
         "method": "sendMessage",
         "chat_id": chat_id,
@@ -238,7 +238,7 @@ def handle_setting(chat_id):
     }
 
 def handle_help(chat_id):
-    logging.info(f"/help command from chat {chat_id}")
+    logger.info(f"/help command from chat {chat_id}")
     return {
         "method": "sendMessage",
         "chat_id": chat_id,
@@ -249,13 +249,12 @@ 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]}")
+            logger.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()}")
+            logger.info(f"Received {current_step}: {text.strip()}")
 
         idx = conversation_fields.index(current_step)
         if idx < len(conversation_fields) - 1:
@@ -265,17 +264,14 @@ 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,
@@ -291,7 +287,7 @@ def handle_conversation(chat_id, text):
 # -------------------------------------------------------------------
 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
-    logging.info("Initiating streaming to YouTube")
+    logger.info("Initiating streaming to YouTube")
     try:
         streaming_state = "streaming"
         reset_statistics()
@@ -323,17 +319,15 @@ def stream_to_youtube(input_url, quality_settings, video_codec, audio_codec, out
 
         video_stream.codec_context.time_base = fractions.Fraction(1, video_stream.rate)
 
-        logging.info("Streaming started successfully.")
+        logger.info("Streaming started successfully.")
 
-        # Start the live log updater in a background thread if not already running.
+        # Start live log updater 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)
-            live_log_thread.daemon = True
+            live_log_thread = threading.Thread(target=live_log_updater, daemon=True)
             live_log_thread.start()
-            logging.info("Live log updater thread started.")
+            logger.info("Live log updater thread started.")
 
-        # Stream loop: process packets until state changes
         while streaming_state in ["streaming", "paused"]:
             for packet in input_stream.demux():
                 if streaming_state == "stopped":
@@ -358,7 +352,6 @@ def stream_to_youtube(input_url, quality_settings, video_codec, audio_codec, out
                             if hasattr(out_packet, "size"):
                                 bytes_sent += out_packet.size
 
-            # Flush remaining packets
             for out_packet in video_stream.encode():
                 output_stream.mux(out_packet)
             for out_packet in out_audio_stream.encode():
@@ -367,22 +360,20 @@ def stream_to_youtube(input_url, quality_settings, video_codec, audio_codec, out
             if streaming_state == "paused":
                 time.sleep(1)
 
-        # Clean up resources
         try:
             video_stream.close()
             out_audio_stream.close()
             output_stream.close()
             input_stream.close()
         except Exception as cleanup_error:
-            logging.error(f"Error during cleanup: {cleanup_error}")
+            logger.error(f"Error during cleanup: {cleanup_error}")
 
-        logging.info("Streaming complete, resources cleaned up.")
+        logger.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)
+        logger.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):
@@ -393,7 +384,6 @@ def start_streaming(chat_id):
 
     stream_chat_id = chat_id
     try:
-        # Start the background streaming thread
         stream_thread = threading.Thread(
             target=stream_to_youtube,
             args=(
@@ -403,14 +393,11 @@ def start_streaming(chat_id):
                 user_inputs["audio_codec"],
                 user_inputs["output_url"],
                 chat_id,
-            )
+            ),
+            daemon=True
         )
-        stream_thread.daemon = True
         stream_thread.start()
-        logging.info("Streaming thread started.")
-
-        # Immediately inform the user that streaming has started;
-        # live log updates are available via the global variable and the /logs command.
+        logger.info("Streaming thread started.")
         return {
             "method": "sendMessage",
             "chat_id": chat_id,
@@ -420,7 +407,7 @@ def start_streaming(chat_id):
         }
     except Exception as e:
         error_message = f"Failed to start streaming: {str(e)}"
-        logging.error(error_message)
+        logger.error(error_message)
         notify_error(chat_id, error_message)
         return send_guide_message(chat_id, error_message)
 
@@ -431,7 +418,7 @@ def pause_stream(chat_id):
     global streaming_state
     if streaming_state == "streaming":
         streaming_state = "paused"
-        logging.info("Streaming paused.")
+        logger.info("Streaming paused.")
         return {
             "method": "sendMessage",
             "chat_id": chat_id,
@@ -444,7 +431,7 @@ def resume_stream(chat_id):
     global streaming_state
     if streaming_state == "paused":
         streaming_state = "streaming"
-        logging.info("Streaming resumed.")
+        logger.info("Streaming resumed.")
         return {
             "method": "sendMessage",
             "chat_id": chat_id,
@@ -457,7 +444,7 @@ def abort_stream(chat_id):
     global streaming_state
     if streaming_state in ["streaming", "paused"]:
         streaming_state = "stopped"
-        logging.info("Streaming aborted by user.")
+        logger.info("Streaming aborted by user.")
         return {
             "method": "sendMessage",
             "chat_id": chat_id,
@@ -487,7 +474,7 @@ def stream_status(chat_id):
 @app.post("/webhook")
 async def telegram_webhook(request: Request):
     update = await request.json()
-    logging.debug(f"Received update: {update}")
+    logger.debug(f"Received update: {update}")
 
     # Process messages from users
     if "message" in update:
@@ -511,7 +498,6 @@ 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
@@ -533,7 +519,6 @@ async def telegram_webhook(request: Request):
         else:
             response = send_guide_message(chat_id, "❓ Unknown callback command.")
 
-        # Edit the original message with updated information
         response["method"] = "editMessageText"
         response["message_id"] = message_id
         return response