feat: integrate web support

Author: ihmily

app/core/record_manager.py

@@ -1,4 +1,5 @@
 import asyncio
+import threading
 from datetime import datetime, timedelta
 
 from ..messages.message_pusher import MessagePusher
@@ -10,11 +11,15 @@
 from .stream_manager import LiveStreamRecorder
 
 
+class GlobalRecordingState:
+    recordings = []
+    lock = threading.Lock()
+
+
 class RecordingManager:
     def __init__(self, app):
         self.app = app
         self.settings = app.settings
-        self.recordings = []
         self.periodic_task_started = False
         self.loop_time_seconds = None
         self.app.language_manager.add_observer(self)
@@ -23,6 +28,14 @@ def __init__(self, app):
         self.load()
         self.initialize_dynamic_state()
 
+    @property
+    def recordings(self):
+        return GlobalRecordingState.recordings
+
+    @recordings.setter
+    def recordings(self, value):
+        raise AttributeError("Please use add_recording/update_recording methods to modify data")
+
     def load(self):
         language = self.app.language_manager.language
         for key in ("recording_manager", "video_quality"):
@@ -31,7 +44,8 @@ def load(self):
     def load_recordings(self):
         """Load recordings from a JSON file into objects."""
         recordings_data = self.app.config_manager.load_recordings_config()
-        self.recordings = [Recording.from_dict(rec) for rec in recordings_data]
+        if not GlobalRecordingState.recordings:
+            GlobalRecordingState.recordings = [Recording.from_dict(rec) for rec in recordings_data]
         logger.info(f"Live Recordings: Loaded {len(self.recordings)} items")
 
     def initialize_dynamic_state(self):
@@ -42,11 +56,31 @@ def initialize_dynamic_state(self):
             recording.loop_time_seconds = self.loop_time_seconds
             recording.update_title(self._[recording.quality])
 
-    async def update_recording(self, recording: Recording, updated_info: dict):
+    async def add_recording(self, recording):
+        with GlobalRecordingState.lock:
+            GlobalRecordingState.recordings.append(recording)
+            await self.persist_recordings()
+
+    async def remove_recording(self, recording: Recording):
+        with GlobalRecordingState.lock:
+            GlobalRecordingState.recordings.remove(recording)
+            await self.persist_recordings()
+
+    async def clear_all_recordings(self):
+        with GlobalRecordingState.lock:
+            GlobalRecordingState.recordings.clear()
+            await self.persist_recordings()
+
+    async def persist_recordings(self):
+        """Persist recordings to a JSON file."""
+        data_to_save = [rec.to_dict() for rec in self.recordings]
+        await self.app.config_manager.save_recordings_config(data_to_save)
+
+    async def update_recording_card(self, recording: Recording, updated_info: dict):
         """Update an existing recording object and persist changes to a JSON file."""
         if recording:
             recording.update(updated_info)
-            self.app.page.run_task(self.save_to_json)
+            self.app.page.run_task(self.persist_recordings)
 
     @staticmethod
     async def _update_recording(
@@ -74,9 +108,10 @@ async def start_monitor_recording(self, recording: Recording, auto_save: bool =
                 selected=False,
             )
             self.app.page.run_task(self.check_if_live, recording)
-            self.app.page.run_task(self.app.record_card_manager.update_cards, recording)
+            self.app.page.run_task(self.app.record_card_manager.update_card, recording)
+            self.app.page.pubsub.send_others_on_topic("update", recording)
             if auto_save:
-                self.app.page.run_task(self.save_to_json)
+                self.app.page.run_task(self.persist_recordings)
 
     async def stop_monitor_recording(self, recording: Recording, auto_save: bool = True):
         """
@@ -91,9 +126,10 @@ async def stop_monitor_recording(self, recording: Recording, auto_save: bool = T
                 selected=False,
             )
             self.stop_recording(recording)
-            self.app.page.run_task(self.app.record_card_manager.update_cards, recording)
+            self.app.page.run_task(self.app.record_card_manager.update_card, recording)
+            self.app.page.pubsub.send_others_on_topic("update", recording)
             if auto_save:
-                self.app.page.run_task(self.save_to_json)
+                self.app.page.run_task(self.persist_recordings)
 
     async def start_monitor_recordings(self):
         """
@@ -105,7 +141,7 @@ async def start_monitor_recordings(self):
         for recording in pre_start_monitor_recordings:
             if cards_obj[recording.rec_id]["card"].visible:
                 self.app.page.run_task(self.start_monitor_recording, recording, auto_save=False)
-        self.app.page.run_task(self.save_to_json)
+        self.app.page.run_task(self.persist_recordings)
         logger.info(f"Batch Start Monitor Recordings: {[i.rec_id for i in pre_start_monitor_recordings]}")
 
     async def stop_monitor_recordings(self, selected_recordings: list[Recording | None] | None = None):
@@ -119,19 +155,18 @@ async def stop_monitor_recordings(self, selected_recordings: list[Recording | No
         for recording in pre_stop_monitor_recordings:
             if cards_obj[recording.rec_id]["card"].visible:
                 self.app.page.run_task(self.stop_monitor_recording, recording, auto_save=False)
-        self.app.page.run_task(self.save_to_json)
+        self.app.page.run_task(self.persist_recordings)
         logger.info(f"Batch Stop Monitor Recordings: {[i.rec_id for i in pre_stop_monitor_recordings]}")
 
     async def get_selected_recordings(self):
         return [recording for recording in self.recordings if recording.selected]
 
-    def remove_recordings(self, recordings: list[Recording]):
+    async def remove_recordings(self, recordings: list[Recording]):
         """Remove a recording from the list and update the JSON file."""
         for recording in recordings:
             if recording in self.recordings:
-                self.recordings.remove(recording)
+                await self.remove_recording(recording)
                 logger.info(f"Delete Items: {recording.rec_id}-{recording.streamer_name}")
-        self.app.page.run_task(self.save_to_json)
 
     def find_recording_by_id(self, rec_id: str):
         """Find a recording by its ID (hash of dict representation)."""
@@ -140,11 +175,6 @@ def find_recording_by_id(self, rec_id: str):
                 return rec
         return None
 
-    async def save_to_json(self):
-        """Persist recordings to a JSON file."""
-        recordings_data = [rec.to_dict() for rec in self.recordings]
-        await self.app.config_manager.save_recordings_config(recordings_data)
-
     async def check_all_live_status(self):
         """Check the live status of all recordings and update their display titles."""
         for recording in self.recordings:
@@ -264,7 +294,8 @@ async def check_if_live(self, recording: Recording):
                     self.start_update(recording)
                     self.app.page.run_task(recorder.start_recording, stream_info)
 
-                self.app.page.run_task(self.app.record_card_manager.update_cards, recording)
+                self.app.page.run_task(self.app.record_card_manager.update_card, recording)
+                self.app.page.pubsub.send_others_on_topic("update", recording)
 
             else:
                 recording.status_info = RecordingStatus.MONITORING
@@ -278,8 +309,9 @@ async def check_if_live(self, recording: Recording):
                             "display_title": title,
                         }
                     )
-                    self.app.page.run_task(self.app.record_card_manager.update_cards, recording)
-                    self.app.page.run_task(self.save_to_json)
+                    self.app.page.run_task(self.app.record_card_manager.update_card, recording)
+                    self.app.page.pubsub.send_others_on_topic("update", recording)
+                    self.app.page.run_task(self.persist_recordings)
 
     @staticmethod
     def start_update(recording: Recording):
@@ -323,8 +355,9 @@ def get_duration(self, recording: Recording):
             return str(total_duration).split(".")[0]
 
     async def delete_recording_cards(self, recordings: list[Recording]):
-        self.remove_recordings(recordings)
         self.app.page.run_task(self.app.record_card_manager.remove_recording_card, recordings)
+        self.app.page.pubsub.send_others_on_topic('delete', recordings)
+        await self.remove_recordings(recordings)
 
     async def check_free_space(self, output_dir: str | None = None):
         disk_space_limit = float(self.settings.user_config.get("recording_space_threshold"))
@@ -342,4 +375,4 @@ async def check_free_space(self, output_dir: str | None = None):
             )
 
         else:
-            self.app.recording_enabled = True
+            self.app.recording_enabled = True

app/core/stream_manager.py

@@ -93,7 +93,7 @@ def _get_output_dir(self, stream_info: StreamData) -> str:
                 output_dir = os.path.join(output_dir, f"{now[:10]}_{live_title}")
         os.makedirs(output_dir, exist_ok=True)
         self.recording.recording_dir = output_dir
-        self.app.page.run_task(self.app.record_manager.save_to_json)
+        self.app.page.run_task(self.app.record_manager.persist_recordings)
         return output_dir
 
     def _get_save_path(self, filename: str) -> str:
@@ -225,18 +225,19 @@ async def start_ffmpeg(
 
                 await asyncio.sleep(1)
 
+            return_code = process.returncode
+            safe_return_code = [0, 255]
             stdout, stderr = await process.communicate()
-            if stderr:
+            if return_code not in safe_return_code and stderr:
                 logger.error(f"FFmpeg Stderr Output: {str(stderr.decode()).splitlines()[0]}")
                 self.recording.status_info = RecordingStatus.RECORDING_ERROR
                 self.app.record_manager.stop_recording(self.recording)
-                await self.app.record_card_manager.update_cards(self.recording)
+                await self.app.record_card_manager.update_card(self.recording)
+                self.app.page.pubsub.send_others_on_topic("update", self.recording)
                 await self.app.snack_bar.show_snack_bar(
                     record_name + " " + self._["record_stream_error"], duration=2000
                 )
 
-            return_code = process.returncode
-            safe_return_code = [0, 255]
             if return_code in safe_return_code:
                 if self.recording.monitor_status:
                     self.recording.status_info = RecordingStatus.MONITORING
@@ -253,7 +254,8 @@ async def start_ffmpeg(
                     logger.success(f"Live recording completed: {record_name}")
 
                 self.recording.update({"display_title": display_title})
-                self.app.page.run_task(self.app.record_card_manager.update_cards, self.recording)
+                await self.app.record_card_manager.update_card(self.recording)
+                self.app.page.pubsub.send_others_on_topic("update", self.recording)
                 if self.app.recording_enabled and process in self.app.process_manager.ffmpeg_processes:
                     self.app.page.run_task(self.app.record_manager.check_if_live, self.recording)
                 else:
@@ -400,6 +402,6 @@ def get_headers_params(live_url, platform_key):
             "qiandurebo": "referer:https://qiandurebo.com",
             "17live": "referer:https://17.live/en/live/6302408",
             "lang": "referer:https://www.lang.live",
-            "shopee": f"origin:{live_domain}",
+            "shopee": "origin:" + live_domain,
         }
-        return record_headers.get(platform_key)
+        return record_headers.get(platform_key)

app/ui/components/recording_card.py

@@ -20,12 +20,17 @@ def __init__(self, app):
         self.app.language_manager.add_observer(self)
         self._ = {}
         self.load()
+        self.pubsub_subscribe()
 
     def load(self):
         language = self.app.language_manager.language
         for key in ("recording_card", "recording_manager", "base", "home_page", "video_quality"):
             self._.update(language.get(key, {}))
 
+    def pubsub_subscribe(self):
+        self.app.page.pubsub.subscribe_topic("update", self.subscribe_update_card)
+        self.app.page.pubsub.subscribe_topic("delete", self.subscribe_remove_cards)
+
     async def create_card(self, recording: Recording):
         """Create a card for a given recording."""
         rec_id = recording.rec_id
@@ -124,7 +129,7 @@ def _create_card_components(self, recording: Recording):
             "monitor_button": monitor_button,
         }
 
-    async def update_cards(self, recording):
+    async def update_card(self, recording):
         """Update only the recordings cards in the scrollable content area."""
         if recording.rec_id in self.cards_obj:
             recording_card = self.cards_obj[recording.rec_id]
@@ -161,8 +166,10 @@ async def update_monitor_state(self, recording: Recording):
             )
             self.app.page.run_task(self.app.record_manager.check_if_live, recording)
             self.app.page.run_task(self.app.snack_bar.show_snack_bar, self._["start_monitor_tip"], ft.Colors.GREEN)
-        await self.update_cards(recording)
-        self.app.page.run_task(self.app.record_manager.save_to_json)
+
+        await self.update_card(recording)
+        self.app.page.pubsub.send_others_on_topic("update", recording)
+        self.app.page.run_task(self.app.record_manager.persist_recordings)
 
     async def show_recording_info_dialog(self, recording: Recording):
         """Display a dialog with detailed information about the recording."""
@@ -175,10 +182,11 @@ async def edit_recording_callback(self, recording_list: list[dict]):
         recording = recording_list[0]
         rec_id = recording["rec_id"]
         recording_obj = self.app.record_manager.find_recording_by_id(rec_id)
-        await self.app.record_manager.update_recording(recording_obj, updated_info=recording)
+        await self.app.record_manager.update_recording_card(recording_obj, updated_info=recording)
         if not recording["monitor_status"]:
             recording_obj.display_title = f"[{self._['monitor_stopped']}] " + recording_obj.title
-        await self.update_cards(recording_obj)
+        await self.update_card(recording_obj)
+        self.app.page.pubsub.send_others_on_topic("update", recording)
 
     async def on_toggle_recording(self, recording: Recording):
         """Toggle the recording state for a specific recording."""
@@ -197,7 +205,8 @@ async def on_toggle_recording(self, recording: Recording):
                 else:
                     await self.app.snack_bar.show_snack_bar(self._["please_start_monitor_tip"])
 
-            await self.update_cards(recording)
+            await self.update_card(recording)
+            self.app.page.pubsub.send_others_on_topic("update", recording)
 
     async def on_delete_recording(self, recording: Recording):
         """Delete a recording from the list and update UI."""
@@ -212,11 +221,27 @@ async def on_delete_recording(self, recording: Recording):
 
     async def remove_recording_card(self, recordings: list[Recording]):
         home_page = self.app.current_page
-        for recording in recordings:
-            if recording.rec_id in self.cards_obj:
-                card = self.cards_obj[recording.rec_id]["card"]
-                home_page.recording_card_area.controls.remove(card)
-                del self.cards_obj[recording.rec_id]
+
+        existing_ids = {rec.rec_id for rec in self.app.record_manager.recordings}
+        remove_ids = {rec.rec_id for rec in recordings}
+        keep_ids = existing_ids - remove_ids
+
+        cards_to_remove = [
+            card_data["card"]
+            for rec_id, card_data in self.cards_obj.items()
+            if rec_id not in keep_ids
+        ]
+
+        home_page.recording_card_area.controls = [
+            control
+            for control in home_page.recording_card_area.controls
+            if control not in cards_to_remove
+        ]
+
+        self.cards_obj = {
+            k: v for k, v in self.cards_obj.items()
+            if k in keep_ids
+        }
         home_page.recording_card_area.update()
 
     @staticmethod
@@ -318,3 +343,9 @@ async def monitor_button_on_click(self, _, recording: Recording):
 
     async def recording_card_on_click(self, _, recording: Recording):
         await self.on_card_click(recording)
+
+    async def subscribe_update_card(self, _, recording: Recording):
+        await self.update_card(recording)
+
+    async def subscribe_remove_cards(self, _, recordings: list[Recording]):
+        await self.remove_recording_card(recordings)