no-name-here commited on
Commit
2d041db
·
verified ·
1 Parent(s): e06de82

Update app.py

Browse files
Files changed (1) hide show
  1. app.py +427 -265
app.py CHANGED
@@ -179,32 +179,40 @@ def format_settings_for_display(session: dict) -> str:
179
  if session.get('logo_enabled') and session.get('logo_data_bytes'):
180
  logo_status = f"Enabled ({session.get('logo_original_filename', 'Unknown name')})"
181
 
 
 
 
182
  display_data = {
183
  "Output URL": session.get('output_url'),
184
- "Quality": session.get('quality_settings'),
185
- "Video Codec": session.get('video_codec'),
186
- "Audio Codec": session.get('audio_codec'),
187
- "Resolution": session.get('resolution'),
188
- "FPS": session.get('fps'),
189
- "Video Bitrate": session.get('video_bitrate'),
190
- "Audio Bitrate": session.get('audio_bitrate'),
191
- "FFmpeg Preset": session.get('ffmpeg_preset'),
192
- "Loop Count": "Infinite" if session.get('loop_count') == -1 else session.get('loop_count'),
193
- "Playlist Length": len(session.get('input_url_playlist', [])),
194
  "Logo": logo_status,
195
  }
196
  if session.get('logo_enabled') and session.get('logo_data_bytes'):
197
- display_data["Logo Position"] = session.get('logo_position')
198
- display_data["Logo Scale"] = session.get('logo_scale')
199
- display_data["Logo Opacity"] = session.get('logo_opacity')
200
-
201
- return "\n".join([f"• *{k}*: `{v}`" for k, v in display_data.items() if v is not None])
 
 
 
 
 
202
 
203
 
204
  # --- UI Helpers (send_message, edit_message_text, answer_callback_query, get_main_keyboard, get_uptime, compose_status_message) ---
205
  # These functions remain largely the same as they don't interact with disk.
206
  # Minor change in compose_status_message for logo display from bytes.
207
- def send_message(chat_id: int, text: str, parse_mode="Markdown", reply_markup=None):
208
  logger.info(f"[Chat {chat_id}] Sending message: {text[:100]}...")
209
  response = {
210
  "method": "sendMessage",
@@ -216,7 +224,7 @@ def send_message(chat_id: int, text: str, parse_mode="Markdown", reply_markup=No
216
  response["reply_markup"] = reply_markup
217
  return response
218
 
219
- def edit_message_text(chat_id: int, message_id: int, text: str, parse_mode="Markdown", reply_markup=None):
220
  logger.info(f"[Chat {chat_id}] Editing message {message_id}: {text[:100]}...")
221
  response = {
222
  "method": "editMessageText",
@@ -282,40 +290,52 @@ def get_uptime(start_time_obj):
282
  def compose_status_message(chat_id: int, include_config: bool = False) -> str:
283
  session = get_user_session(chat_id)
284
  state = session['streaming_state']
285
-
286
- status_lines = [
287
- f"🤖 *Stream Bot Status for Chat ID {chat_id} (In-Memory Mode)*",
288
- f"_Settings, logos, and schedules are NOT persistent across bot restarts._",
289
- f"State: `{state}`",
 
 
 
 
 
 
 
 
290
  ]
 
291
  if session.get('error_notification_user'):
292
- status_lines.append(f"⚠️ Error: `{session['error_notification_user']}`")
 
293
 
294
  if state in ["streaming", "paused", "starting", "stopping"]:
295
- current_input_url_display = "N/A"
296
  playlist = session.get('input_url_playlist', [])
297
  playlist_idx = session.get('current_playlist_index', 0)
298
- if playlist and 0 <= playlist_idx < len(playlist):
299
- current_input_url_display = playlist[playlist_idx]
300
-
301
- status_lines.extend([
302
- f"Uptime: `{get_uptime(session['stream_start_time'])}`",
303
- f"Frames Encoded: `{session['frames_encoded']}`",
304
- f"Bytes Sent: `{session['bytes_sent'] / (1024*1024):.2f} MB`",
305
- f"Current Input: `{current_input_url_display}` (Item {playlist_idx+1}/{len(playlist)})",
306
- f"Loop Iteration: {session.get('current_loop_iteration',0)+1}/{'Infinite' if session.get('loop_count',0) == -1 else session.get('loop_count',0) or 1}"
307
- ])
308
-
 
 
 
309
  if include_config or state not in ["streaming", "paused"]:
310
- status_lines.append("\n📝 *Current Configuration:*")
311
- status_lines.append(format_settings_for_display(session))
312
 
313
  if state in ["streaming", "paused", "starting"]:
314
- status_lines.append("\n📋 *User Live Logs (last 10):*")
315
- user_logs_preview = "\n".join(session['live_log_lines_user'][-10:])
316
- status_lines.append(f"<pre>{user_logs_preview if user_logs_preview else 'No user logs yet.'}</pre>")
317
-
318
- return "\n".join(status_lines)
 
319
 
320
 
321
  # --- Core Streaming Logic ---
@@ -691,6 +711,14 @@ def stream_engine_thread_target(chat_id: int):
691
  # get_help_text needs to be updated about non-persistence.
692
 
693
  async def handle_telegram_update(update: dict):
 
 
 
 
 
 
 
 
694
  if "message" in update:
695
  message = update["message"]
696
  chat_id = message["chat"]["id"]
@@ -703,38 +731,46 @@ async def handle_telegram_update(update: dict):
703
 
704
  if message.get("photo") or message.get("document"):
705
  if session.get("current_step") == "awaiting_logo_upload":
706
- # Need to pass the actual UploadFile if using FastAPI's file handling for webhook
707
- # For direct JSON, we'd get file_id and need a bot instance to download.
708
- # Assuming direct JSON from Telegram with file_id:
709
  return await handle_logo_data_from_telegram(chat_id, message)
710
  else:
711
- return send_message(chat_id, "Received a file, but I wasn't expecting one. Use /set_logo first if you want to upload a logo.")
712
 
713
  if command:
714
  logger.info(f"[Chat {chat_id} ({user_name})] Received command: {text}")
715
- if command == "/start" or command == "/menu":
716
- with lock: session['current_step'] = None
717
- return send_message(chat_id, compose_status_message(chat_id, include_config=True), reply_markup=get_main_keyboard(session), parse_mode="HTML")
718
  elif command == "/help":
719
- return send_message(chat_id, get_help_text(), parse_mode="Markdown")
720
  elif command == "/settings":
721
- return start_settings_conversation(chat_id, advanced=True)
 
722
  elif command == "/stream":
723
- return await start_stream_command_handler(chat_id)
724
  elif command == "/playlist":
725
  return await handle_playlist_command(chat_id, text)
726
  elif command == "/set_logo":
727
- with lock: session["current_step"] = "awaiting_logo_upload"
728
- return send_message(chat_id, "Please upload your logo image (PNG or JPG). Max 1MB recommended due to in-memory storage.")
729
  elif command == "/schedule":
730
  return await handle_schedule_command(chat_id, text)
731
  elif command == "/status":
732
- return send_message(chat_id, compose_status_message(chat_id, include_config=True), reply_markup=get_main_keyboard(session), parse_mode="HTML")
733
  elif command == "/abort":
734
- return await stream_abort_handler(chat_id)
735
- else:
736
- return await handle_conversation_step(chat_id, text)
737
- else:
 
 
 
 
 
 
 
 
 
 
738
  return await handle_conversation_step(chat_id, text)
739
 
740
  elif "callback_query" in update:
@@ -752,25 +788,25 @@ async def handle_telegram_update(update: dict):
752
 
753
  if data == "cfg_start":
754
  if not session.get('input_url_playlist') or not session.get('output_url'):
755
- # This is tricky as start_settings_conversation returns a message dict, not just text.
756
  settings_response = start_settings_conversation(chat_id, advanced=session.get('advanced_mode', False))
757
- response_text = "Playlist or Output URL is not set. Let's configure.\n" + settings_response['text']
758
- # If settings_response has its own reply_markup, it might conflict.
759
- # For simplicity, let's assume start_settings_conversation is primarily text.
760
  return [ack, send_message(chat_id, response_text, reply_markup=settings_response.get('reply_markup') or get_main_keyboard(session))]
761
-
762
  else:
763
  return [ack, await start_stream_command_handler(chat_id, message_id_to_edit=message_id)]
 
764
  elif data == "cfg_advanced":
765
  settings_response = start_settings_conversation(chat_id, advanced=True)
766
- return [ack, edit_message_text(chat_id, message_id, settings_response['text'] + "\n\n" + compose_status_message(chat_id, include_config=True), reply_markup=settings_response.get('reply_markup') or get_main_keyboard(session), parse_mode="HTML")]
 
 
 
767
  elif data == "cfg_playlist":
768
  return [ack, await handle_playlist_config_interaction(chat_id, message_id)]
769
  elif data == "cfg_logo":
770
  return [ack, await handle_logo_config_interaction(chat_id, message_id)]
771
  elif data == "cfg_schedule":
772
  return [ack, await handle_schedule_config_interaction(chat_id, message_id)]
773
-
774
  elif data == "stream_pause":
775
  return [ack, await stream_pause_handler(chat_id, message_id)]
776
  elif data == "stream_resume":
@@ -778,30 +814,46 @@ async def handle_telegram_update(update: dict):
778
  elif data == "stream_abort":
779
  return [ack, await stream_abort_handler(chat_id, message_id_to_edit=message_id)]
780
  elif data == "stream_status":
781
- return [ack, edit_message_text(chat_id, message_id, compose_status_message(chat_id, include_config=False), reply_markup=get_main_keyboard(session), parse_mode="HTML")]
 
 
782
  elif data == "stream_stop_loop_graceful":
783
  with lock: session['stop_gracefully_flag'] = True
784
  append_user_live_log(chat_id, "Graceful loop stop requested.")
785
- return [ack, edit_message_text(chat_id, message_id, "Graceful loop stop initiated. Current iteration will complete.\n" + compose_status_message(chat_id), reply_markup=get_main_keyboard(session), parse_mode="HTML")]
 
 
786
 
787
  elif data == "show_global_logs":
788
- global_logs_text = "📜 *Global Bot Logs (last 20):*\n<pre>" + "\n".join(live_log_lines_global[-20:]) + "</pre>"
789
- return [ack, edit_message_text(chat_id, message_id, global_logs_text + "\n\n" + compose_status_message(chat_id), reply_markup=get_main_keyboard(session), parse_mode="HTML")]
 
 
 
 
790
  elif data == "show_help":
791
- return [ack, edit_message_text(chat_id, message_id, get_help_text() + "\n\n" + compose_status_message(chat_id), reply_markup=get_main_keyboard(session))]
792
-
 
 
793
  elif data.startswith("set_quality_"):
794
- quality = data.split("_")[-1]
795
- with lock:
796
- session["quality_settings"] = quality
797
- if quality in QUALITY_SETTINGS_MAP and quality != "custom":
798
- for key, val in QUALITY_SETTINGS_MAP[quality].items():
799
- session[key] = val
800
- session["current_step"] = None
801
- # save_user_settings_to_file(chat_id) # REMOVED
802
- return [ack, edit_message_text(chat_id, message_id, f"Quality set to `{quality}`.\n" + compose_status_message(chat_id, True), reply_markup=get_main_keyboard(session), parse_mode="HTML")]
803
-
804
- return [ack, send_message(chat_id, "Unknown action.")]
 
 
 
 
 
 
805
 
806
  return {"status": "ok", "message": "Update type not handled"}
807
 
@@ -832,15 +884,12 @@ def start_settings_conversation(chat_id: int, advanced: bool):
832
  session['conversation_fields_list'] = CONVERSATION_STEPS_ADVANCED if advanced else CONVERSATION_STEPS_SIMPLE
833
  session['current_step_index'] = 0
834
  session['current_step'] = session['conversation_fields_list'][0][0]
835
-
836
  first_field_key, first_prompt, _ = session['conversation_fields_list'][0]
837
-
838
- # Prepare initial message (send_message or part of edit_message_text)
839
- initial_message_content = {}
840
-
841
  if first_field_key == "quality_settings_prompt":
842
- initial_message_content = {
843
- "text": first_prompt,
844
  "reply_markup": {
845
  "inline_keyboard": [
846
  [{"text": q.title(), "callback_data": f"set_quality_{q}"} for q in ["low", "medium", "high", "custom"]]
@@ -848,11 +897,11 @@ def start_settings_conversation(chat_id: int, advanced: bool):
848
  }
849
  }
850
  else:
851
- current_value_msg = f"(Current: `{session.get(first_field_key, 'Not set')}`)" if session.get(first_field_key) else ""
852
- initial_message_content = {"text": f"{first_prompt}\n{current_value_msg}\n\nType /cancel to abort setup."}
853
-
854
- # This function is now called from callbacks too, so it needs to return a dict that can be used by send_message or edit_message_text
855
- return initial_message_content
856
 
857
 
858
  async def handle_conversation_step(chat_id: int, user_text: str):
@@ -862,100 +911,114 @@ async def handle_conversation_step(chat_id: int, user_text: str):
862
  current_step_key_local = None
863
  validator_local = None
864
  current_step_index_local = -1
 
 
 
865
 
866
  with lock:
867
  if not session.get('current_step') or not session.get('conversation_fields_list'):
868
- return send_message(chat_id, "No setup in progress. Use /start or a menu button.", reply_markup=get_main_keyboard(session))
869
-
870
- if user_text.lower() == "/cancel":
871
  session['current_step'] = None
872
- session['current_step_index'] = 0 # Reset index
873
  cancelled = True
874
- else:
875
- cancelled = False
876
-
877
- # Ensure current_step_index is valid
878
- if not cancelled and not (0 <= session.get('current_step_index', -1) < len(session['conversation_fields_list'])):
879
- session['current_step'] = None # Invalid state, reset
880
  session['current_step_index'] = 0
881
  logger.warning(f"[Chat {chat_id}] Invalid current_step_index, resetting conversation.")
882
  invalid_index = True
883
  else:
884
- invalid_index = False
885
-
886
- if not cancelled and not invalid_index:
887
  current_step_index_local = session['current_step_index']
888
  current_step_key_local, _, validator_local = session['conversation_fields_list'][current_step_index_local]
889
-
890
- # Handle early exits outside the lock
 
 
 
 
 
 
891
  if cancelled:
892
- return send_message(chat_id, "Setup cancelled.", reply_markup=get_main_keyboard(session))
893
  if invalid_index:
894
- return send_message(chat_id, "Conversation state error, setup reset. Please start again.", reply_markup=get_main_keyboard(session))
 
895
  user_input = user_text.strip()
896
 
897
  if validator_local and not validator_local(user_input):
898
- # Re-fetch prompt as it might have been clobbered if lock wasn't held long enough (should be fine here)
899
- _, prompt, _ = session['conversation_fields_list'][current_step_index_local] # Use local index; tuple is (key, prompt, validator)
900
- return send_message(chat_id, f"Invalid input for {current_step_key_local.replace('_',' ').title()}. Please try again.\n{prompt}")
 
 
 
 
 
 
 
 
 
 
901
 
902
  with lock:
903
- # Re-verify current_step_key just in case, though primary logic relies on index
904
  if current_step_key_local != session['conversation_fields_list'][session['current_step_index']][0]:
905
  logger.error(f"[Chat {chat_id}] Conversation step mismatch! Aborting step.")
906
- # Build response outside lock below
907
  step_mismatch = True
908
  else:
909
- step_mismatch = False
910
-
911
- if not step_mismatch:
912
  if current_step_key_local == "input_url_playlist":
913
  session["input_url_playlist"] = [user_input] if user_input else []
914
  elif current_step_key_local == "quality_settings_prompt":
915
  quality_input = user_input.lower()
916
- if quality_input in QUALITY_SETTINGS_MAP.keys() or quality_input == "custom":
917
  session["quality_settings"] = quality_input
918
  if quality_input in QUALITY_SETTINGS_MAP and quality_input != "custom":
919
- for key, val in QUALITY_SETTINGS_MAP[quality_input].items(): session[key] = val
920
- append_user_live_log(chat_id, f"Quality set to {quality_input} via text.")
 
921
  else:
922
- logger.warning(f"[Chat {chat_id}] Invalid quality input '{quality_input}' in conversation step. Not applied.")
923
- elif current_step_key_local == "loop_count" or current_step_key_local == "fps":
924
  session[current_step_key_local] = int(user_input)
 
925
  else:
926
  session[current_step_key_local] = user_input
927
-
928
- append_user_live_log(chat_id, f"Set {current_step_key_local} = {user_input}")
929
 
930
  session['current_step_index'] += 1
931
  if session['current_step_index'] < len(session['conversation_fields_list']):
932
  next_step_key, next_prompt, _ = session['conversation_fields_list'][session['current_step_index']]
933
  session['current_step'] = next_step_key
934
  next_step_is_quality = (next_step_key == "quality_settings_prompt")
935
- current_value_msg = f"(Current: `{session.get(next_step_key, 'Not set')}`)" if session.get(next_step_key) else ""
936
  else:
937
- next_step_key = None
938
- next_prompt = None
939
- next_step_is_quality = False
940
- current_value_msg = ""
941
  session['current_step'] = None
942
  session['current_step_index'] = 0
943
  logger.info(f"[Chat {chat_id}] In-memory settings configured.")
944
 
945
- # All returns happen outside the lock to avoid holding it during send_message calls
 
 
 
946
  if step_mismatch:
947
- return send_message(chat_id, "A state error occurred. Please try /cancel and restart setup.", reply_markup=get_main_keyboard(session))
948
 
949
  if next_step_key is not None:
950
  if next_step_is_quality:
951
- return send_message(chat_id, next_prompt, reply_markup={
952
  "inline_keyboard": [
953
  [{"text": q.title(), "callback_data": f"set_quality_{q}"} for q in ["low", "medium", "high", "custom"]]
954
- ]})
955
- return send_message(chat_id, f"{next_prompt}\n{current_value_msg}\n\nType /cancel to abort setup.")
 
 
 
 
956
  else:
957
- # save_user_settings_to_file(chat_id) # REMOVED
958
- return send_message(chat_id, "All settings configured (in memory for this session)!\n" + compose_status_message(chat_id, True), reply_markup=get_main_keyboard(session), parse_mode="HTML")
 
 
 
959
 
960
 
961
  async def handle_playlist_command(chat_id: int, text: str):
@@ -964,48 +1027,78 @@ async def handle_playlist_command(chat_id: int, text: str):
964
  parts = text.split(maxsplit=2)
965
  action = parts[1].lower() if len(parts) > 1 else "show"
966
  msg = ""
967
-
968
  with lock:
969
  playlist = session.get('input_url_playlist', [])
970
- if action == "add" and len(parts) > 2:
971
- url_to_add = parts[2]
972
- if validate_url(url_to_add):
973
- playlist.append(url_to_add)
974
- # session['input_url_playlist'] = playlist # Modified in place
975
- msg = f"Added to playlist. Total items: {len(playlist)}."
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
976
  else:
977
- msg = "Invalid URL format."
978
- elif action == "remove" and len(parts) > 2:
979
- idx_str = parts[2]
980
- removed_item = None
981
- if idx_str.lower() == "last" and playlist: removed_item = playlist.pop()
982
- elif idx_str.isdigit():
983
- idx_to_remove = int(idx_str) -1
984
- if 0 <= idx_to_remove < len(playlist): removed_item = playlist.pop(idx_to_remove)
985
- if removed_item: msg = f"Removed '{removed_item[:50]}...' from playlist. Total items: {len(playlist)}."
986
- else: msg = "Invalid index or playlist empty."
987
  elif action == "clear":
 
988
  session['input_url_playlist'] = []
989
- msg = "Playlist cleared."
990
  elif action == "show":
991
  if playlist:
992
- plist_str = "\n".join([f"{i+1}. `{url}`" for i, url in enumerate(playlist)])
993
- msg = f"*Current Playlist ({len(playlist)} items):*\n{plist_str}"
994
- else: msg = "Playlist is empty."
995
- else: msg = "Usage: /playlist [add <url>|remove <index|last>|clear|show]"
996
-
997
- # save_user_settings_to_file(chat_id) # REMOVED (for playlist changes)
 
 
 
 
 
 
 
 
 
 
 
998
  return send_message(chat_id, msg, reply_markup=get_main_keyboard(session))
999
 
1000
  async def handle_playlist_config_interaction(chat_id: int, message_id: int):
1001
  session = get_user_session(chat_id)
1002
  playlist = session.get('input_url_playlist', [])
1003
- msg = ""
1004
  if playlist:
1005
- plist_str = "\n".join([f"{i+1}. `{url}`" for i, url in enumerate(playlist)])
1006
- msg = f"*Current Playlist ({len(playlist)} items):*\n{plist_str}\n\nUse `/playlist add <url>`, `/playlist remove <index>`, `/playlist clear`."
 
 
 
 
 
 
 
 
 
1007
  else:
1008
- msg = "Playlist is empty. Use `/playlist add <url>` to add your first stream input."
1009
  return edit_message_text(chat_id, message_id, msg + "\n\n" + compose_status_message(chat_id), reply_markup=get_main_keyboard(session))
1010
 
1011
 
@@ -1051,27 +1144,47 @@ async def handle_logo_data_from_telegram(chat_id: int, message: dict):
1051
  session['logo_enabled'] = True
1052
  session['current_step'] = None
1053
 
1054
- # save_user_settings_to_file(chat_id) # REMOVED
1055
  append_user_live_log(chat_id, f"CONCEPTUAL: Logo data for '{dummy_filename}' stored in memory.")
1056
- return send_message(chat_id, f"Conceptual: Logo '{dummy_filename}' data stored in memory and enabled! (Actual file download/processing requires full bot capabilities beyond this webhook example.)\nLogo settings (position, etc.) are in Advanced Settings.", reply_markup=get_main_keyboard(session))
 
 
 
 
 
 
 
1057
 
1058
  async def handle_logo_config_interaction(chat_id: int, message_id: int):
1059
  session = get_user_session(chat_id)
1060
- msg_parts = ["*Logo Configuration (In-Memory):*"]
1061
  if session.get('logo_data_bytes') and session.get('logo_enabled'):
1062
- msg_parts.append(f"Status: Enabled (`{session.get('logo_original_filename', 'N/A')}`)")
1063
- msg_parts.append(f"Position: `{session.get('logo_position')}` (Change in Advanced Settings)")
1064
- msg_parts.append(f"Scale: `{session.get('logo_scale')}` (Change in Advanced Settings)")
1065
- msg_parts.append(f"Opacity: `{session.get('logo_opacity')}` (Change in Advanced Settings)")
1066
- msg_parts.append("\nTo change logo, send /set_logo then upload a new image.")
1067
- msg_parts.append("To disable: `/configure_logo enabled false` (conceptual command).")
 
 
 
 
1068
  elif session.get('logo_data_bytes') and not session.get('logo_enabled'):
1069
- msg_parts.append(f"Status: Disabled (`{session.get('logo_original_filename', 'N/A')}`)")
1070
- msg_parts.append("To enable: `/configure_logo enabled true` (conceptual command).")
 
 
 
1071
  else:
1072
- msg_parts.append("No logo uploaded/stored. Send /set_logo then upload an image.")
1073
-
1074
- return edit_message_text(chat_id, message_id, "\n".join(msg_parts) + "\n\n" + compose_status_message(chat_id), reply_markup=get_main_keyboard(session))
 
 
 
 
 
 
 
1075
 
1076
 
1077
  def schedule_stream_job(chat_id: int, trigger_time: datetime.datetime, job_name: str = None):
@@ -1115,7 +1228,13 @@ def start_stream_for_scheduler(chat_id: int):
1115
  async def handle_schedule_command(chat_id: int, text: str):
1116
  parts = text.split(maxsplit=3)
1117
  if len(parts) < 3:
1118
- return send_message(chat_id, "Usage: `/schedule YYYY-MM-DD HH:MM:SS [Optional Job Name]`\nExample: `/schedule 2024-12-31 23:55:00 NYE`\nTimezone is UTC. Schedules are in-memory and lost on restart.")
 
 
 
 
 
 
1119
 
1120
  try:
1121
  date_str, time_str = parts[1], parts[2]
@@ -1124,73 +1243,92 @@ async def handle_schedule_command(chat_id: int, text: str):
1124
  trigger_time_utc = trigger_time_naive.replace(tzinfo=datetime.timezone.utc)
1125
 
1126
  if trigger_time_utc <= datetime.datetime.now(datetime.timezone.utc):
1127
- return send_message(chat_id, "Scheduled time must be in the future.")
1128
 
1129
  schedule_stream_job(chat_id, trigger_time_utc, job_name)
1130
  session = get_user_session(chat_id)
1131
- return send_message(chat_id, f"Stream scheduled (in-memory) as '{job_name}' for {trigger_time_utc.strftime('%Y-%m-%d %H:%M:%S %Z')}. Current settings will be used.", reply_markup=get_main_keyboard(session))
 
 
 
 
 
 
 
 
1132
  except ValueError as e:
1133
- return send_message(chat_id, f"Invalid date/time format: {e}. Use YYYY-MM-DD HH:MM:SS.")
 
 
 
 
 
1134
 
1135
  async def handle_schedule_config_interaction(chat_id: int, message_id: int):
1136
  session = get_user_session(chat_id)
1137
  jobs = scheduler.get_jobs()
1138
  user_jobs = [job for job in jobs if job.id.startswith(f"stream_{chat_id}_")]
1139
-
1140
- msg_parts = ["*Scheduled Streams (In-Memory - Lost on Restart):*"]
1141
  if user_jobs:
1142
  for job in user_jobs:
1143
- run_time_str = job.next_run_time.strftime('%Y-%m-%d %H:%M:%S %Z') if job.next_run_time else "N/A"
1144
- msg_parts.append(f"• `{job.name}` at `{run_time_str}` (ID: `{job.id}`)")
1145
- msg_parts.append(f" To cancel: `/cancel_schedule {job.id}` (conceptual command).")
 
 
1146
  else:
1147
- msg_parts.append("No streams currently scheduled.")
1148
- msg_parts.append("\nUse `/schedule YYYY-MM-DD HH:MM:SS [Name]` to schedule a new stream.")
1149
-
1150
- return edit_message_text(chat_id, message_id, "\n".join(msg_parts) + "\n\n" + compose_status_message(chat_id), reply_markup=get_main_keyboard(session))
 
 
 
 
 
 
 
 
1151
 
1152
 
1153
  async def start_stream_command_handler(chat_id: int, message_id_to_edit: int = None, is_scheduled: bool = False):
1154
  session = get_user_session(chat_id)
1155
  lock = session_locks[chat_id]
1156
- response = None
 
1157
 
1158
  with lock:
1159
  if session['streaming_state'] in ["streaming", "paused", "starting"]:
1160
- msg = f"A stream is already active ({session['streaming_state']})."
1161
- if message_id_to_edit: response = edit_message_text(chat_id, message_id_to_edit, msg + "\n" + compose_status_message(chat_id), reply_markup=get_main_keyboard(session), parse_mode="HTML")
1162
- else: response = send_message(chat_id, msg, reply_markup=get_main_keyboard(session))
1163
- return response
1164
-
1165
- if not session.get('input_url_playlist') or not session.get('output_url'):
1166
- msg = "Input playlist or Output URL is not set. Please configure first using /settings or menu."
1167
- if message_id_to_edit: response = edit_message_text(chat_id, message_id_to_edit, msg + "\n" + compose_status_message(chat_id), reply_markup=get_main_keyboard(session), parse_mode="HTML")
1168
- else: response = send_message(chat_id, msg, reply_markup=get_main_keyboard(session))
1169
- return response
1170
-
1171
- session['streaming_state'] = "starting"
1172
- session['error_notification_user'] = ""
1173
-
1174
- # save_user_settings_to_file(chat_id) # REMOVED
1175
 
1176
- thread = threading.Thread(target=stream_engine_thread_target, args=(chat_id,), name=f"StreamThread-{chat_id}", daemon=True)
 
 
 
 
 
 
 
1177
  session['stream_thread_ref'] = thread
1178
- thread.start()
1179
- append_user_live_log(chat_id, "Stream thread initiated.")
1180
-
1181
  await asyncio.sleep(0.2)
1182
 
1183
  status_msg = compose_status_message(chat_id)
1184
  if is_scheduled and not message_id_to_edit:
1185
  logger.info(f"[Chat {chat_id}] Scheduled stream started. Status: {status_msg[:100]}")
1186
- # Conceptual proactive message:
1187
- # await send_proactive_message(chat_id, "Your scheduled stream is now starting!\n" + status_msg)
1188
  return {"status": "scheduled_stream_started_silently_inmemory"}
1189
  elif message_id_to_edit:
1190
- response = edit_message_text(chat_id, message_id_to_edit, "Stream starting...\n" + status_msg, reply_markup=get_main_keyboard(session), parse_mode="HTML")
1191
  else:
1192
- response = send_message(chat_id, "Stream starting...\n" + status_msg, reply_markup=get_main_keyboard(session), parse_mode="HTML")
1193
- return response
1194
 
1195
 
1196
  async def stream_pause_handler(chat_id: int, message_id: int):
@@ -1199,11 +1337,17 @@ async def stream_pause_handler(chat_id: int, message_id: int):
1199
  with lock:
1200
  if session['streaming_state'] == "streaming":
1201
  session['streaming_state'] = "paused"
1202
- append_user_live_log(chat_id, "Stream paused by user.")
1203
- msg = "Stream paused."
1204
  else:
1205
- msg = f"Stream is not streaming (state: {session['streaming_state']}). Cannot pause."
1206
- return edit_message_text(chat_id, message_id, msg + "\n" + compose_status_message(chat_id), reply_markup=get_main_keyboard(session), parse_mode="HTML")
 
 
 
 
 
 
 
1207
 
1208
  async def stream_resume_handler(chat_id: int, message_id: int):
1209
  session = get_user_session(chat_id)
@@ -1211,28 +1355,39 @@ async def stream_resume_handler(chat_id: int, message_id: int):
1211
  with lock:
1212
  if session['streaming_state'] == "paused":
1213
  session['streaming_state'] = "streaming"
1214
- append_user_live_log(chat_id, "Stream resumed by user.")
1215
- msg = "Stream resumed."
1216
  else:
1217
- msg = f"Stream is not paused (state: {session['streaming_state']}). Cannot resume."
1218
- return edit_message_text(chat_id, message_id, msg + "\n" + compose_status_message(chat_id), reply_markup=get_main_keyboard(session), parse_mode="HTML")
 
 
 
 
 
 
1219
 
1220
  async def stream_abort_handler(chat_id: int, message_id_to_edit: int = None):
1221
  session = get_user_session(chat_id)
1222
  lock = session_locks[chat_id]
1223
  thread_to_join = None
1224
- msg = ""
1225
-
1226
  with lock:
1227
  if session['streaming_state'] in ["streaming", "paused", "starting"]:
1228
- session['streaming_state'] = "stopping"
1229
  thread_to_join = session.get('stream_thread_ref')
1230
- append_user_live_log(chat_id, "Stream abort sequence initiated by user.")
1231
- msg = "Stream aborting..."
1232
  else:
1233
- msg = f"No active stream to abort (state: {session['streaming_state']})."
1234
- if message_id_to_edit: return edit_message_text(chat_id, message_id_to_edit, msg + "\n" + compose_status_message(chat_id), reply_markup=get_main_keyboard(session), parse_mode="HTML")
1235
- return send_message(chat_id, msg, reply_markup=get_main_keyboard(session))
 
 
 
 
 
 
 
 
1236
 
1237
  if thread_to_join and thread_to_join.is_alive():
1238
  logger.info(f"[Chat {chat_id}] Waiting for stream thread {thread_to_join.name} to join...")
@@ -1242,46 +1397,53 @@ async def stream_abort_handler(chat_id: int, message_id_to_edit: int = None):
1242
  logger.warning(f"[Chat {chat_id}] Stream thread {thread_to_join.name} still alive after join timeout.")
1243
  else:
1244
  append_user_live_log(chat_id, "Stream thread terminated.")
1245
-
1246
  with lock:
1247
  if session['streaming_state'] == "stopping":
1248
  session['streaming_state'] = "stopped"
1249
  session['stream_thread_ref'] = None
1250
 
1251
- final_status_msg = "Stream aborted.\n" + compose_status_message(chat_id)
1252
  if message_id_to_edit:
1253
- return edit_message_text(chat_id, message_id_to_edit, final_status_msg, reply_markup=get_main_keyboard(session), parse_mode="HTML")
1254
- else:
1255
- return send_message(chat_id, final_status_msg, reply_markup=get_main_keyboard(session), parse_mode="HTML")
1256
 
1257
 
1258
  def get_help_text():
1259
  return (
1260
- f"*Advanced Stream Bot v{APP_VERSION} (In-Memory Edition) Help*\n\n"
1261
- "⚠️ *IMPORTANT: All settings, uploaded logos, and schedules are stored in memory "
1262
- "and will be LOST if the bot restarts due to environment restrictions.*\n\n"
1263
- "Use the interactive menu buttons or the following commands:\n"
1264
- "*/start* or */menu* - Show main menu and status.\n"
1265
- "*/settings* - Start advanced configuration.\n"
1266
- "*/stream* - Start stream with current settings.\n"
1267
- "*/playlist add <url>* - Add URL to playlist.\n"
1268
- "*/playlist remove <index|last>* - Remove URL from playlist.\n"
1269
- "*/playlist clear* - Clear entire playlist.\n"
1270
- "*/playlist show* - Display current playlist.\n"
1271
- "*/set_logo* - Prompts to upload a logo image (stored in memory).\n"
1272
- " (Logo settings like position, scale are in Advanced Settings via menu)\n"
1273
- "*/schedule YYYY-MM-DD HH:MM:SS [Name]* - Schedule a stream (UTC time, in-memory).\n"
1274
- " Example: `/schedule 2024-07-04 18:30:00 My Stream`\n"
1275
- "*/status* - Show detailed stream status and config.\n"
1276
- "*/abort* - Force stop the current stream.\n"
1277
- "\n*While Streaming (via menu buttons):*\n"
1278
- "• Pause/Resume Stream.\n"
1279
- "• Abort Stream.\n"
1280
- "• Status Update (refreshes message).\n"
1281
- "• Stop Loop Gracefully (for infinite loops, current iteration completes).\n"
1282
- "\n*Configuration Conversation:*\n"
1283
- "When in a setup process (e.g., after /settings), reply to prompts.\n"
1284
- "Type `/cancel` to exit setup anytime.\n"
 
 
 
 
 
 
 
 
1285
  )
1286
 
1287
  # --- FastAPI Webhook Endpoint ---
 
179
  if session.get('logo_enabled') and session.get('logo_data_bytes'):
180
  logo_status = f"Enabled ({session.get('logo_original_filename', 'Unknown name')})"
181
 
182
+ loop_count = session.get('loop_count', 0)
183
+ loop_display = "&#8734; Infinite" if loop_count == -1 else ("Once" if loop_count == 0 else f"{loop_count}x")
184
+
185
  display_data = {
186
  "Output URL": session.get('output_url'),
187
+ "Quality": session.get('quality_settings', '-'),
188
+ "Video Codec": session.get('video_codec', '-'),
189
+ "Audio Codec": session.get('audio_codec', '-'),
190
+ "Resolution": session.get('resolution', '-'),
191
+ "FPS": session.get('fps', '-'),
192
+ "Video Bitrate": session.get('video_bitrate', '-'),
193
+ "Audio Bitrate": session.get('audio_bitrate', '-'),
194
+ "FFmpeg Preset": session.get('ffmpeg_preset', '-'),
195
+ "Loop": loop_display,
196
+ "Playlist Items": len(session.get('input_url_playlist', [])),
197
  "Logo": logo_status,
198
  }
199
  if session.get('logo_enabled') and session.get('logo_data_bytes'):
200
+ display_data["Logo Position"] = session.get('logo_position', '-')
201
+ display_data["Logo Scale"] = session.get('logo_scale', '-')
202
+ display_data["Logo Opacity"] = session.get('logo_opacity', '-')
203
+
204
+ lines = []
205
+ for k, v in display_data.items():
206
+ if v is not None:
207
+ safe_v = str(v).replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
208
+ lines.append(f" \u2022 <b>{k}</b>: <code>{safe_v}</code>")
209
+ return "\n".join(lines)
210
 
211
 
212
  # --- UI Helpers (send_message, edit_message_text, answer_callback_query, get_main_keyboard, get_uptime, compose_status_message) ---
213
  # These functions remain largely the same as they don't interact with disk.
214
  # Minor change in compose_status_message for logo display from bytes.
215
+ def send_message(chat_id: int, text: str, parse_mode="HTML", reply_markup=None):
216
  logger.info(f"[Chat {chat_id}] Sending message: {text[:100]}...")
217
  response = {
218
  "method": "sendMessage",
 
224
  response["reply_markup"] = reply_markup
225
  return response
226
 
227
+ def edit_message_text(chat_id: int, message_id: int, text: str, parse_mode="HTML", reply_markup=None):
228
  logger.info(f"[Chat {chat_id}] Editing message {message_id}: {text[:100]}...")
229
  response = {
230
  "method": "editMessageText",
 
290
  def compose_status_message(chat_id: int, include_config: bool = False) -> str:
291
  session = get_user_session(chat_id)
292
  state = session['streaming_state']
293
+
294
+ STATE_EMOJI = {
295
+ "idle": "💤", "starting": "🔄", "streaming": "🔴",
296
+ "paused": "⏸", "stopping": "🛑", "stopped": "⏹",
297
+ "completed": "✅", "error": "❌",
298
+ }
299
+ state_icon = STATE_EMOJI.get(state, "❓")
300
+
301
+ lines = [
302
+ f"🤖 <b>Stream Bot</b> — Chat <code>{chat_id}</code>",
303
+ f"<i>⚠️ Settings are in-memory only — lost on restart.</i>",
304
+ f"",
305
+ f"{state_icon} <b>State:</b> <code>{state}</code>",
306
  ]
307
+
308
  if session.get('error_notification_user'):
309
+ safe_err = str(session['error_notification_user']).replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
310
+ lines.append(f"⚠️ <b>Last Error:</b> <code>{safe_err}</code>")
311
 
312
  if state in ["streaming", "paused", "starting", "stopping"]:
 
313
  playlist = session.get('input_url_playlist', [])
314
  playlist_idx = session.get('current_playlist_index', 0)
315
+ current_input_url_display = playlist[playlist_idx] if playlist and 0 <= playlist_idx < len(playlist) else "N/A"
316
+ safe_url = current_input_url_display.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
317
+ loop_count = session.get('loop_count', 0)
318
+ loop_total = "&#8734;" if loop_count == -1 else str(max(loop_count, 1))
319
+
320
+ lines += [
321
+ f"",
322
+ f"⏱ <b>Uptime:</b> <code>{get_uptime(session['stream_start_time'])}</code>",
323
+ f"🎞 <b>Frames:</b> <code>{session['frames_encoded']:,}</code>",
324
+ f"📤 <b>Sent:</b> <code>{session['bytes_sent'] / (1024*1024):.2f} MB</code>",
325
+ f"🎬 <b>Input:</b> <code>{safe_url}</code> ({playlist_idx+1}/{len(playlist)})",
326
+ f"🔁 <b>Loop:</b> <code>{session.get('current_loop_iteration', 0)+1}/{loop_total}</code>",
327
+ ]
328
+
329
  if include_config or state not in ["streaming", "paused"]:
330
+ lines += ["", "📝 <b>Current Configuration:</b>", format_settings_for_display(session)]
 
331
 
332
  if state in ["streaming", "paused", "starting"]:
333
+ user_logs = session.get('live_log_lines_user', [])
334
+ last_logs = "\n".join(user_logs[-8:]) if user_logs else "No logs yet."
335
+ safe_logs = last_logs.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
336
+ lines += ["", "📋 <b>Recent Logs:</b>", f"<pre>{safe_logs}</pre>"]
337
+
338
+ return "\n".join(lines)
339
 
340
 
341
  # --- Core Streaming Logic ---
 
711
  # get_help_text needs to be updated about non-persistence.
712
 
713
  async def handle_telegram_update(update: dict):
714
+ try:
715
+ return await _handle_telegram_update_inner(update)
716
+ except Exception as e:
717
+ logger.error(f"Unhandled exception in handle_telegram_update: {e}", exc_info=True)
718
+ return {"status": "ok", "message": "Internal error — please try again."}
719
+
720
+
721
+ async def _handle_telegram_update_inner(update: dict):
722
  if "message" in update:
723
  message = update["message"]
724
  chat_id = message["chat"]["id"]
 
731
 
732
  if message.get("photo") or message.get("document"):
733
  if session.get("current_step") == "awaiting_logo_upload":
 
 
 
734
  return await handle_logo_data_from_telegram(chat_id, message)
735
  else:
736
+ return send_message(chat_id, "📎 Got a file, but I wasn't expecting one.\nUse /set_logo first if you want to upload a logo.", reply_markup=get_main_keyboard(session))
737
 
738
  if command:
739
  logger.info(f"[Chat {chat_id} ({user_name})] Received command: {text}")
740
+ if command in ("/start", "/menu"):
741
+ with lock: session['current_step'] = None
742
+ return send_message(chat_id, compose_status_message(chat_id, include_config=True), reply_markup=get_main_keyboard(session))
743
  elif command == "/help":
744
+ return send_message(chat_id, get_help_text(), reply_markup=get_main_keyboard(session))
745
  elif command == "/settings":
746
+ resp = start_settings_conversation(chat_id, advanced=True)
747
+ return send_message(chat_id, resp['text'], reply_markup=resp.get('reply_markup') or get_main_keyboard(session))
748
  elif command == "/stream":
749
+ return await start_stream_command_handler(chat_id)
750
  elif command == "/playlist":
751
  return await handle_playlist_command(chat_id, text)
752
  elif command == "/set_logo":
753
+ with lock: session["current_step"] = "awaiting_logo_upload"
754
+ return send_message(chat_id, "🖼 Please upload your logo image (PNG or JPG).\n<i>Max 1 MB recommended due to in-memory storage.</i>", reply_markup=get_main_keyboard(session))
755
  elif command == "/schedule":
756
  return await handle_schedule_command(chat_id, text)
757
  elif command == "/status":
758
+ return send_message(chat_id, compose_status_message(chat_id, include_config=True), reply_markup=get_main_keyboard(session))
759
  elif command == "/abort":
760
+ return await stream_abort_handler(chat_id)
761
+ elif command == "/cancel":
762
+ # /cancel works both inside and outside a conversation
763
+ return await handle_conversation_step(chat_id, "/cancel")
764
+ else:
765
+ # Unknown slash command — if in a conversation pass it through, else show hint
766
+ if session.get('current_step'):
767
+ return await handle_conversation_step(chat_id, text)
768
+ return send_message(
769
+ chat_id,
770
+ f"❓ Unknown command <code>{text.split()[0]}</code>.\nTap <b>❔ Help</b> below or use /help for a list of commands.",
771
+ reply_markup=get_main_keyboard(session)
772
+ )
773
+ else:
774
  return await handle_conversation_step(chat_id, text)
775
 
776
  elif "callback_query" in update:
 
788
 
789
  if data == "cfg_start":
790
  if not session.get('input_url_playlist') or not session.get('output_url'):
 
791
  settings_response = start_settings_conversation(chat_id, advanced=session.get('advanced_mode', False))
792
+ response_text = "⚙️ <b>Playlist or Output URL not set.</b> Let's configure:\n\n" + settings_response['text']
 
 
793
  return [ack, send_message(chat_id, response_text, reply_markup=settings_response.get('reply_markup') or get_main_keyboard(session))]
 
794
  else:
795
  return [ack, await start_stream_command_handler(chat_id, message_id_to_edit=message_id)]
796
+
797
  elif data == "cfg_advanced":
798
  settings_response = start_settings_conversation(chat_id, advanced=True)
799
+ return [ack, edit_message_text(chat_id, message_id,
800
+ settings_response['text'] + "\n\n" + compose_status_message(chat_id, include_config=True),
801
+ reply_markup=settings_response.get('reply_markup') or get_main_keyboard(session))]
802
+
803
  elif data == "cfg_playlist":
804
  return [ack, await handle_playlist_config_interaction(chat_id, message_id)]
805
  elif data == "cfg_logo":
806
  return [ack, await handle_logo_config_interaction(chat_id, message_id)]
807
  elif data == "cfg_schedule":
808
  return [ack, await handle_schedule_config_interaction(chat_id, message_id)]
809
+
810
  elif data == "stream_pause":
811
  return [ack, await stream_pause_handler(chat_id, message_id)]
812
  elif data == "stream_resume":
 
814
  elif data == "stream_abort":
815
  return [ack, await stream_abort_handler(chat_id, message_id_to_edit=message_id)]
816
  elif data == "stream_status":
817
+ return [ack, edit_message_text(chat_id, message_id,
818
+ compose_status_message(chat_id, include_config=False),
819
+ reply_markup=get_main_keyboard(session))]
820
  elif data == "stream_stop_loop_graceful":
821
  with lock: session['stop_gracefully_flag'] = True
822
  append_user_live_log(chat_id, "Graceful loop stop requested.")
823
+ return [ack, edit_message_text(chat_id, message_id,
824
+ "⏳ <b>Graceful stop initiated.</b> Current iteration will finish before stopping.\n\n" + compose_status_message(chat_id),
825
+ reply_markup=get_main_keyboard(session))]
826
 
827
  elif data == "show_global_logs":
828
+ safe_logs = "\n".join(live_log_lines_global[-20:]).replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
829
+ global_logs_text = "📜 <b>Global Bot Logs (last 20):</b>\n<pre>" + (safe_logs or "No logs yet.") + "</pre>"
830
+ return [ack, edit_message_text(chat_id, message_id,
831
+ global_logs_text + "\n\n" + compose_status_message(chat_id),
832
+ reply_markup=get_main_keyboard(session))]
833
+
834
  elif data == "show_help":
835
+ return [ack, edit_message_text(chat_id, message_id,
836
+ get_help_text() + "\n\n" + compose_status_message(chat_id),
837
+ reply_markup=get_main_keyboard(session))]
838
+
839
  elif data.startswith("set_quality_"):
840
+ quality = data.split("_", 2)[-1]
841
+ if quality in QUALITY_SETTINGS_MAP or quality == "custom":
842
+ with lock:
843
+ session["quality_settings"] = quality
844
+ if quality in QUALITY_SETTINGS_MAP and quality != "custom":
845
+ for key, val in QUALITY_SETTINGS_MAP[quality].items():
846
+ session[key] = val
847
+ session["current_step"] = None
848
+ return [ack, edit_message_text(chat_id, message_id,
849
+ f"✅ <b>Quality set to</b> <code>{quality}</code>.\n\n" + compose_status_message(chat_id, True),
850
+ reply_markup=get_main_keyboard(session))]
851
+ else:
852
+ return [ack, answer_callback_query(cb_query["id"], text=f"Unknown quality: {quality}", show_alert=True)]
853
+
854
+ return [ack, send_message(chat_id,
855
+ "❓ <b>Unknown action.</b> Please use the menu buttons below.",
856
+ reply_markup=get_main_keyboard(session))]
857
 
858
  return {"status": "ok", "message": "Update type not handled"}
859
 
 
884
  session['conversation_fields_list'] = CONVERSATION_STEPS_ADVANCED if advanced else CONVERSATION_STEPS_SIMPLE
885
  session['current_step_index'] = 0
886
  session['current_step'] = session['conversation_fields_list'][0][0]
887
+
888
  first_field_key, first_prompt, _ = session['conversation_fields_list'][0]
889
+
 
 
 
890
  if first_field_key == "quality_settings_prompt":
891
+ return {
892
+ "text": f"<b>{first_prompt}</b>",
893
  "reply_markup": {
894
  "inline_keyboard": [
895
  [{"text": q.title(), "callback_data": f"set_quality_{q}"} for q in ["low", "medium", "high", "custom"]]
 
897
  }
898
  }
899
  else:
900
+ current_val = session.get(first_field_key)
901
+ current_value_msg = f"\nCurrent: <code>{str(current_val).replace('&','&amp;').replace('<','&lt;').replace('>','&gt;')}</code>" if current_val else ""
902
+ return {
903
+ "text": f"<b>{first_prompt}</b>{current_value_msg}\n\nType <code>/cancel</code> to abort setup."
904
+ }
905
 
906
 
907
  async def handle_conversation_step(chat_id: int, user_text: str):
 
911
  current_step_key_local = None
912
  validator_local = None
913
  current_step_index_local = -1
914
+ no_setup = False
915
+ cancelled = False
916
+ invalid_index = False
917
 
918
  with lock:
919
  if not session.get('current_step') or not session.get('conversation_fields_list'):
920
+ no_setup = True
921
+ elif user_text.strip().lower() == "/cancel":
 
922
  session['current_step'] = None
923
+ session['current_step_index'] = 0
924
  cancelled = True
925
+ elif not (0 <= session.get('current_step_index', -1) < len(session['conversation_fields_list'])):
926
+ session['current_step'] = None
 
 
 
 
927
  session['current_step_index'] = 0
928
  logger.warning(f"[Chat {chat_id}] Invalid current_step_index, resetting conversation.")
929
  invalid_index = True
930
  else:
 
 
 
931
  current_step_index_local = session['current_step_index']
932
  current_step_key_local, _, validator_local = session['conversation_fields_list'][current_step_index_local]
933
+
934
+ # All responses happen outside the lock to prevent deadlocks
935
+ if no_setup:
936
+ return send_message(
937
+ chat_id,
938
+ "💬 No setup in progress. Use /start or tap a menu button to begin.",
939
+ reply_markup=get_main_keyboard(session)
940
+ )
941
  if cancelled:
942
+ return send_message(chat_id, "✅ <b>Setup cancelled.</b> Your previous settings are unchanged.", reply_markup=get_main_keyboard(session))
943
  if invalid_index:
944
+ return send_message(chat_id, "⚠️ <b>Conversation state error — setup reset.</b>\nPlease start again.", reply_markup=get_main_keyboard(session))
945
+
946
  user_input = user_text.strip()
947
 
948
  if validator_local and not validator_local(user_input):
949
+ _, prompt, _ = session['conversation_fields_list'][current_step_index_local]
950
+ field_label = current_step_key_local.replace('_', ' ').title()
951
+ return send_message(
952
+ chat_id,
953
+ f"❌ <b>Invalid input for {field_label}.</b>\nPlease try again.\n\n<i>{prompt}</i>\n\nType <code>/cancel</code> to abort setup."
954
+ )
955
+
956
+ next_step_key = None
957
+ next_prompt = None
958
+ next_step_is_quality = False
959
+ current_value_msg = ""
960
+ step_mismatch = False
961
+ log_line = None
962
 
963
  with lock:
 
964
  if current_step_key_local != session['conversation_fields_list'][session['current_step_index']][0]:
965
  logger.error(f"[Chat {chat_id}] Conversation step mismatch! Aborting step.")
 
966
  step_mismatch = True
967
  else:
 
 
 
968
  if current_step_key_local == "input_url_playlist":
969
  session["input_url_playlist"] = [user_input] if user_input else []
970
  elif current_step_key_local == "quality_settings_prompt":
971
  quality_input = user_input.lower()
972
+ if quality_input in QUALITY_SETTINGS_MAP or quality_input == "custom":
973
  session["quality_settings"] = quality_input
974
  if quality_input in QUALITY_SETTINGS_MAP and quality_input != "custom":
975
+ for key, val in QUALITY_SETTINGS_MAP[quality_input].items():
976
+ session[key] = val
977
+ log_line = f"Quality set to {quality_input}"
978
  else:
979
+ logger.warning(f"[Chat {chat_id}] Invalid quality input '{quality_input}'. Not applied.")
980
+ elif current_step_key_local in ("loop_count", "fps"):
981
  session[current_step_key_local] = int(user_input)
982
+ log_line = f"Set {current_step_key_local} = {user_input}"
983
  else:
984
  session[current_step_key_local] = user_input
985
+ log_line = f"Set {current_step_key_local} = {user_input}"
 
986
 
987
  session['current_step_index'] += 1
988
  if session['current_step_index'] < len(session['conversation_fields_list']):
989
  next_step_key, next_prompt, _ = session['conversation_fields_list'][session['current_step_index']]
990
  session['current_step'] = next_step_key
991
  next_step_is_quality = (next_step_key == "quality_settings_prompt")
992
+ current_value_msg = f"(Current: <code>{session.get(next_step_key, 'Not set')}</code>)" if session.get(next_step_key) else ""
993
  else:
 
 
 
 
994
  session['current_step'] = None
995
  session['current_step_index'] = 0
996
  logger.info(f"[Chat {chat_id}] In-memory settings configured.")
997
 
998
+ # Log outside the lock to avoid deadlock (append_user_live_log acquires same lock)
999
+ if log_line:
1000
+ append_user_live_log(chat_id, log_line)
1001
+
1002
  if step_mismatch:
1003
+ return send_message(chat_id, "⚠️ <b>State error during setup.</b>\nType <code>/cancel</code> and restart.", reply_markup=get_main_keyboard(session))
1004
 
1005
  if next_step_key is not None:
1006
  if next_step_is_quality:
1007
+ return send_message(chat_id, f"<b>{next_prompt}</b>", reply_markup={
1008
  "inline_keyboard": [
1009
  [{"text": q.title(), "callback_data": f"set_quality_{q}"} for q in ["low", "medium", "high", "custom"]]
1010
+ ]
1011
+ })
1012
+ return send_message(
1013
+ chat_id,
1014
+ f"<b>{next_prompt}</b>\n{current_value_msg}\n\nType <code>/cancel</code> to abort setup."
1015
+ )
1016
  else:
1017
+ return send_message(
1018
+ chat_id,
1019
+ "✅ <b>All settings saved</b> (in-memory for this session)!\n\n" + compose_status_message(chat_id, True),
1020
+ reply_markup=get_main_keyboard(session)
1021
+ )
1022
 
1023
 
1024
  async def handle_playlist_command(chat_id: int, text: str):
 
1027
  parts = text.split(maxsplit=2)
1028
  action = parts[1].lower() if len(parts) > 1 else "show"
1029
  msg = ""
1030
+
1031
  with lock:
1032
  playlist = session.get('input_url_playlist', [])
1033
+ if action == "add":
1034
+ if len(parts) > 2:
1035
+ url_to_add = parts[2]
1036
+ if validate_url(url_to_add):
1037
+ playlist.append(url_to_add)
1038
+ msg = f"✅ <b>Added to playlist.</b>\nTotal items: <code>{len(playlist)}</code>"
1039
+ else:
1040
+ msg = f"❌ <b>Invalid URL format.</b>\nMust start with http/https/rtmp/rtsp/udp/srt/file.\n<code>{url_to_add}</code>"
1041
+ else:
1042
+ msg = "⚠️ Usage: <code>/playlist add &lt;url&gt;</code>"
1043
+ elif action == "remove":
1044
+ if len(parts) > 2:
1045
+ idx_str = parts[2]
1046
+ removed_item = None
1047
+ if idx_str.lower() == "last" and playlist:
1048
+ removed_item = playlist.pop()
1049
+ elif idx_str.isdigit():
1050
+ idx_to_remove = int(idx_str) - 1
1051
+ if 0 <= idx_to_remove < len(playlist):
1052
+ removed_item = playlist.pop(idx_to_remove)
1053
+ safe_removed = removed_item[:60].replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;") if removed_item else None
1054
+ if removed_item:
1055
+ msg = f"🗑 <b>Removed:</b> <code>{safe_removed}</code>\nTotal items remaining: <code>{len(playlist)}</code>"
1056
+ else:
1057
+ msg = f"❌ <b>Invalid index or playlist is empty.</b>"
1058
  else:
1059
+ msg = "⚠️ Usage: <code>/playlist remove &lt;index|last&gt;</code>"
 
 
 
 
 
 
 
 
 
1060
  elif action == "clear":
1061
+ count = len(playlist)
1062
  session['input_url_playlist'] = []
1063
+ msg = f"🗑 <b>Playlist cleared.</b> Removed {count} item(s)."
1064
  elif action == "show":
1065
  if playlist:
1066
+ safe_urls = "\n".join([
1067
+ f" <code>{i+1}.</code> {url.replace('&','&amp;').replace('<','&lt;').replace('>','&gt;')}"
1068
+ for i, url in enumerate(playlist)
1069
+ ])
1070
+ msg = f"📜 <b>Playlist ({len(playlist)} items):</b>\n{safe_urls}"
1071
+ else:
1072
+ msg = "📜 <b>Playlist is empty.</b>\nAdd items with <code>/playlist add &lt;url&gt;</code>"
1073
+ else:
1074
+ msg = (
1075
+ "⚠️ <b>Unknown playlist action.</b>\n"
1076
+ "Usage:\n"
1077
+ " <code>/playlist add &lt;url&gt;</code>\n"
1078
+ " <code>/playlist remove &lt;index|last&gt;</code>\n"
1079
+ " <code>/playlist clear</code>\n"
1080
+ " <code>/playlist show</code>"
1081
+ )
1082
+
1083
  return send_message(chat_id, msg, reply_markup=get_main_keyboard(session))
1084
 
1085
  async def handle_playlist_config_interaction(chat_id: int, message_id: int):
1086
  session = get_user_session(chat_id)
1087
  playlist = session.get('input_url_playlist', [])
 
1088
  if playlist:
1089
+ safe_urls = "\n".join([
1090
+ f" <code>{i+1}.</code> {url.replace('&','&amp;').replace('<','&lt;').replace('>','&gt;')}"
1091
+ for i, url in enumerate(playlist)
1092
+ ])
1093
+ msg = (
1094
+ f"📜 <b>Playlist ({len(playlist)} items):</b>\n{safe_urls}\n\n"
1095
+ "Manage with:\n"
1096
+ " <code>/playlist add &lt;url&gt;</code>\n"
1097
+ " <code>/playlist remove &lt;index&gt;</code>\n"
1098
+ " <code>/playlist clear</code>"
1099
+ )
1100
  else:
1101
+ msg = "📜 <b>Playlist is empty.</b>\nAdd your first stream URL:\n<code>/playlist add &lt;url&gt;</code>"
1102
  return edit_message_text(chat_id, message_id, msg + "\n\n" + compose_status_message(chat_id), reply_markup=get_main_keyboard(session))
1103
 
1104
 
 
1144
  session['logo_enabled'] = True
1145
  session['current_step'] = None
1146
 
 
1147
  append_user_live_log(chat_id, f"CONCEPTUAL: Logo data for '{dummy_filename}' stored in memory.")
1148
+ return send_message(
1149
+ chat_id,
1150
+ f"✅ <b>Logo received</b> (conceptual).\n"
1151
+ f"File <code>{dummy_filename}</code> stored in memory and enabled.\n"
1152
+ f"<i>Note: Actual file download requires full bot API capabilities beyond this webhook example.</i>\n\n"
1153
+ f"Logo settings (position, scale, opacity) are in ⚙️ <b>Advanced Settings</b>.",
1154
+ reply_markup=get_main_keyboard(session)
1155
+ )
1156
 
1157
  async def handle_logo_config_interaction(chat_id: int, message_id: int):
1158
  session = get_user_session(chat_id)
1159
+ lines = ["🖼 <b>Logo Configuration</b>"]
1160
  if session.get('logo_data_bytes') and session.get('logo_enabled'):
1161
+ fname = session.get('logo_original_filename', 'N/A').replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
1162
+ lines += [
1163
+ f" Status: ✅ <b>Enabled</b> — <code>{fname}</code>",
1164
+ f" Position: <code>{session.get('logo_position', '-')}</code>",
1165
+ f" Scale: <code>{session.get('logo_scale', '-')}</code>",
1166
+ f" Opacity: <code>{session.get('logo_opacity', '-')}</code>",
1167
+ "",
1168
+ "To replace: /set_logo then upload a new image.",
1169
+ "To change position/scale/opacity: use ⚙️ Settings.",
1170
+ ]
1171
  elif session.get('logo_data_bytes') and not session.get('logo_enabled'):
1172
+ fname = session.get('logo_original_filename', 'N/A').replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
1173
+ lines += [
1174
+ f" Status: ⏸ <b>Disabled</b> — <code>{fname}</code>",
1175
+ "To enable: update via ⚙️ Settings.",
1176
+ ]
1177
  else:
1178
+ lines += [
1179
+ " No logo uploaded.",
1180
+ "Use /set_logo then upload a PNG or JPG image.",
1181
+ ]
1182
+
1183
+ return edit_message_text(
1184
+ chat_id, message_id,
1185
+ "\n".join(lines) + "\n\n" + compose_status_message(chat_id),
1186
+ reply_markup=get_main_keyboard(session)
1187
+ )
1188
 
1189
 
1190
  def schedule_stream_job(chat_id: int, trigger_time: datetime.datetime, job_name: str = None):
 
1228
  async def handle_schedule_command(chat_id: int, text: str):
1229
  parts = text.split(maxsplit=3)
1230
  if len(parts) < 3:
1231
+ return send_message(
1232
+ chat_id,
1233
+ "🕒 <b>Schedule a Stream</b>\n\n"
1234
+ "Usage: <code>/schedule YYYY-MM-DD HH:MM:SS [Name]</code>\n"
1235
+ "Example: <code>/schedule 2025-12-31 23:55:00 NYE Stream</code>\n\n"
1236
+ "<i>Timezone: UTC. Schedules are in-memory and lost on restart.</i>"
1237
+ )
1238
 
1239
  try:
1240
  date_str, time_str = parts[1], parts[2]
 
1243
  trigger_time_utc = trigger_time_naive.replace(tzinfo=datetime.timezone.utc)
1244
 
1245
  if trigger_time_utc <= datetime.datetime.now(datetime.timezone.utc):
1246
+ return send_message(chat_id, "❌ <b>Scheduled time must be in the future.</b> Please use a future UTC date/time.")
1247
 
1248
  schedule_stream_job(chat_id, trigger_time_utc, job_name)
1249
  session = get_user_session(chat_id)
1250
+ safe_name = job_name.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
1251
+ return send_message(
1252
+ chat_id,
1253
+ f"✅ <b>Stream scheduled!</b>\n"
1254
+ f" Name: <b>{safe_name}</b>\n"
1255
+ f" Time: <code>{trigger_time_utc.strftime('%Y-%m-%d %H:%M:%S UTC')}</code>\n\n"
1256
+ f"<i>Current settings will be used when it fires. Schedule is in-memory only.</i>",
1257
+ reply_markup=get_main_keyboard(session)
1258
+ )
1259
  except ValueError as e:
1260
+ return send_message(
1261
+ chat_id,
1262
+ f"❌ <b>Invalid date/time format.</b>\n"
1263
+ f"Error: <code>{str(e).replace('&','&amp;').replace('<','&lt;').replace('>','&gt;')}</code>\n\n"
1264
+ f"Expected: <code>YYYY-MM-DD HH:MM:SS</code>"
1265
+ )
1266
 
1267
  async def handle_schedule_config_interaction(chat_id: int, message_id: int):
1268
  session = get_user_session(chat_id)
1269
  jobs = scheduler.get_jobs()
1270
  user_jobs = [job for job in jobs if job.id.startswith(f"stream_{chat_id}_")]
1271
+
1272
+ lines = ["🕒 <b>Scheduled Streams</b> <i>(in-memory — lost on restart)</i>"]
1273
  if user_jobs:
1274
  for job in user_jobs:
1275
+ run_time_str = job.next_run_time.strftime('%Y-%m-%d %H:%M:%S UTC') if job.next_run_time else "N/A"
1276
+ safe_name = str(job.name).replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
1277
+ safe_id = str(job.id).replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
1278
+ lines.append(f" \u2022 <b>{safe_name}</b> at <code>{run_time_str}</code>")
1279
+ lines.append(f" ID: <code>{safe_id}</code>")
1280
  else:
1281
+ lines.append(" No streams currently scheduled.")
1282
+
1283
+ lines += [
1284
+ "",
1285
+ "To schedule: <code>/schedule YYYY-MM-DD HH:MM:SS [Name]</code>",
1286
+ ]
1287
+
1288
+ return edit_message_text(
1289
+ chat_id, message_id,
1290
+ "\n".join(lines) + "\n\n" + compose_status_message(chat_id),
1291
+ reply_markup=get_main_keyboard(session)
1292
+ )
1293
 
1294
 
1295
  async def start_stream_command_handler(chat_id: int, message_id_to_edit: int = None, is_scheduled: bool = False):
1296
  session = get_user_session(chat_id)
1297
  lock = session_locks[chat_id]
1298
+
1299
+ early_exit_msg = None
1300
 
1301
  with lock:
1302
  if session['streaming_state'] in ["streaming", "paused", "starting"]:
1303
+ early_exit_msg = f"⚠️ <b>Stream already active</b> (state: <code>{session['streaming_state']}</code>)."
1304
+ elif not session.get('input_url_playlist') or not session.get('output_url'):
1305
+ early_exit_msg = "⚙️ <b>Input playlist or Output URL not configured.</b>\nUse /settings or the menu to set them up first."
1306
+ else:
1307
+ session['streaming_state'] = "starting"
1308
+ session['error_notification_user'] = ""
 
 
 
 
 
 
 
 
 
1309
 
1310
+ if early_exit_msg:
1311
+ status_snippet = compose_status_message(chat_id)
1312
+ if message_id_to_edit:
1313
+ return edit_message_text(chat_id, message_id_to_edit, early_exit_msg + "\n\n" + status_snippet, reply_markup=get_main_keyboard(session))
1314
+ return send_message(chat_id, early_exit_msg + "\n\n" + status_snippet, reply_markup=get_main_keyboard(session))
1315
+
1316
+ thread = threading.Thread(target=stream_engine_thread_target, args=(chat_id,), name=f"StreamThread-{chat_id}", daemon=True)
1317
+ with lock:
1318
  session['stream_thread_ref'] = thread
1319
+ thread.start()
1320
+ append_user_live_log(chat_id, "Stream thread initiated.")
1321
+
1322
  await asyncio.sleep(0.2)
1323
 
1324
  status_msg = compose_status_message(chat_id)
1325
  if is_scheduled and not message_id_to_edit:
1326
  logger.info(f"[Chat {chat_id}] Scheduled stream started. Status: {status_msg[:100]}")
 
 
1327
  return {"status": "scheduled_stream_started_silently_inmemory"}
1328
  elif message_id_to_edit:
1329
+ return edit_message_text(chat_id, message_id_to_edit, "🔴 <b>Stream starting…</b>\n\n" + status_msg, reply_markup=get_main_keyboard(session))
1330
  else:
1331
+ return send_message(chat_id, "🔴 <b>Stream starting…</b>\n\n" + status_msg, reply_markup=get_main_keyboard(session))
 
1332
 
1333
 
1334
  async def stream_pause_handler(chat_id: int, message_id: int):
 
1337
  with lock:
1338
  if session['streaming_state'] == "streaming":
1339
  session['streaming_state'] = "paused"
1340
+ paused_ok = True
 
1341
  else:
1342
+ paused_ok = False
1343
+ state_now = session['streaming_state']
1344
+ if paused_ok:
1345
+ append_user_live_log(chat_id, "Stream paused by user.")
1346
+ msg = "⏸ <b>Stream paused.</b>"
1347
+ else:
1348
+ msg = f"ℹ️ Cannot pause — current state is <code>{state_now}</code>."
1349
+ return edit_message_text(chat_id, message_id, msg + "\n\n" + compose_status_message(chat_id), reply_markup=get_main_keyboard(session))
1350
+
1351
 
1352
  async def stream_resume_handler(chat_id: int, message_id: int):
1353
  session = get_user_session(chat_id)
 
1355
  with lock:
1356
  if session['streaming_state'] == "paused":
1357
  session['streaming_state'] = "streaming"
1358
+ resumed_ok = True
 
1359
  else:
1360
+ resumed_ok = False
1361
+ state_now = session['streaming_state']
1362
+ if resumed_ok:
1363
+ append_user_live_log(chat_id, "Stream resumed by user.")
1364
+ msg = "▶️ <b>Stream resumed.</b>"
1365
+ else:
1366
+ msg = f"ℹ️ Cannot resume — current state is <code>{state_now}</code>."
1367
+ return edit_message_text(chat_id, message_id, msg + "\n\n" + compose_status_message(chat_id), reply_markup=get_main_keyboard(session))
1368
 
1369
  async def stream_abort_handler(chat_id: int, message_id_to_edit: int = None):
1370
  session = get_user_session(chat_id)
1371
  lock = session_locks[chat_id]
1372
  thread_to_join = None
1373
+ no_active_stream = False
1374
+
1375
  with lock:
1376
  if session['streaming_state'] in ["streaming", "paused", "starting"]:
1377
+ session['streaming_state'] = "stopping"
1378
  thread_to_join = session.get('stream_thread_ref')
 
 
1379
  else:
1380
+ no_active_stream = True
1381
+
1382
+ # Handle "nothing to abort" outside the lock
1383
+ if no_active_stream:
1384
+ state = session['streaming_state']
1385
+ msg = f"ℹ️ <b>No active stream to abort.</b>\nCurrent state: <code>{state}</code>"
1386
+ if message_id_to_edit:
1387
+ return edit_message_text(chat_id, message_id_to_edit, msg + "\n\n" + compose_status_message(chat_id), reply_markup=get_main_keyboard(session))
1388
+ return send_message(chat_id, msg, reply_markup=get_main_keyboard(session))
1389
+
1390
+ append_user_live_log(chat_id, "Stream abort sequence initiated by user.")
1391
 
1392
  if thread_to_join and thread_to_join.is_alive():
1393
  logger.info(f"[Chat {chat_id}] Waiting for stream thread {thread_to_join.name} to join...")
 
1397
  logger.warning(f"[Chat {chat_id}] Stream thread {thread_to_join.name} still alive after join timeout.")
1398
  else:
1399
  append_user_live_log(chat_id, "Stream thread terminated.")
1400
+
1401
  with lock:
1402
  if session['streaming_state'] == "stopping":
1403
  session['streaming_state'] = "stopped"
1404
  session['stream_thread_ref'] = None
1405
 
1406
+ final_msg = "⏹ <b>Stream aborted.</b>\n\n" + compose_status_message(chat_id)
1407
  if message_id_to_edit:
1408
+ return edit_message_text(chat_id, message_id_to_edit, final_msg, reply_markup=get_main_keyboard(session))
1409
+ return send_message(chat_id, final_msg, reply_markup=get_main_keyboard(session))
 
1410
 
1411
 
1412
  def get_help_text():
1413
  return (
1414
+ f"<b>🤖 Advanced Stream Bot v{APP_VERSION}</b>\n"
1415
+ f"<i>In-Memory Edition — all data lost on restart.</i>\n"
1416
+ "\n"
1417
+ "<b>⚠️ Important:</b> Settings, logos and schedules are stored only in RAM.\n"
1418
+ "\n"
1419
+ "<b>📋 Commands:</b>\n"
1420
+ " /start or /menu — Main menu &amp; status\n"
1421
+ " /settings — Open advanced configuration\n"
1422
+ " /stream — Start stream with current settings\n"
1423
+ " /status — Detailed stream status\n"
1424
+ " /abort — Force stop the current stream\n"
1425
+ "\n"
1426
+ "<b>🎬 Playlist:</b>\n"
1427
+ " /playlist add &lt;url&gt;\n"
1428
+ " /playlist remove &lt;index|last&gt;\n"
1429
+ " /playlist clear\n"
1430
+ " /playlist show\n"
1431
+ "\n"
1432
+ "<b>🖼 Logo:</b>\n"
1433
+ " /set_logo — Upload a logo image (PNG/JPG)\n"
1434
+ " Position, scale &amp; opacity via ⚙️ Settings menu\n"
1435
+ "\n"
1436
+ "<b>🕒 Schedule:</b>\n"
1437
+ " /schedule YYYY-MM-DD HH:MM:SS [Name] (UTC)\n"
1438
+ " Example: <code>/schedule 2025-12-31 23:55:00 NYE</code>\n"
1439
+ "\n"
1440
+ "<b>▶️ While Streaming (menu buttons):</b>\n"
1441
+ " ⏸ Pause / ▶️ Resume / ⏹ Abort / 📊 Status\n"
1442
+ " ⏳ Stop Loop Gracefully (infinite loops only)\n"
1443
+ "\n"
1444
+ "<b>⚙️ Setup Conversations:</b>\n"
1445
+ " Reply to prompts during /settings flow.\n"
1446
+ " Type <code>/cancel</code> anytime to exit setup.\n"
1447
  )
1448
 
1449
  # --- FastAPI Webhook Endpoint ---