From 078fc7d2807bb460adbba85ba027904eb063a118 Mon Sep 17 00:00:00 2001 From: HH Date: Fri, 17 Apr 2026 03:29:29 +0800 Subject: [PATCH 01/10] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E5=BD=93=E5=BD=95?= =?UTF-8?q?=E5=88=B6=E8=AE=B0=E5=BD=95(=E9=A2=91=E9=81=93)=E6=95=B0?= =?UTF-8?q?=E9=87=8F=E5=A4=AA=E5=A4=9A=E6=97=B6=EF=BC=8C=E7=95=8C=E9=9D=A2?= =?UTF-8?q?=E5=BC=82=E5=B8=B8=E5=8D=A1=E9=A1=BF=EF=BC=8Cflet=E7=9A=84cpu?= =?UTF-8?q?=E5=8D=A0=E7=94=A8=E5=BC=82=E5=B8=B8=E5=A2=9E=E9=AB=98=E9=97=AE?= =?UTF-8?q?=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/ui/components/business/recording_card.py | 127 ++++++++++++------- app/ui/views/recordings_view.py | 125 +++++++++++------- 2 files changed, 158 insertions(+), 94 deletions(-) diff --git a/app/ui/components/business/recording_card.py b/app/ui/components/business/recording_card.py index 708a2f3c..efb92910 100644 --- a/app/ui/components/business/recording_card.py +++ b/app/ui/components/business/recording_card.py @@ -18,7 +18,7 @@ class RecordingCardManager: def __init__(self, app): self.app = app self.cards_obj = {} - self.update_duration_tasks = {} + self.duration_updater_task = None self.selected_cards = {} self.app.language_manager.add_observer(self) self._ = {} @@ -45,7 +45,7 @@ async def create_card(self, recording: Recording, subscribe_add_cards: bool = Fa card_data = self._create_card_components(recording) self.cards_obj[rec_id] = card_data - self.start_update_task(recording) + self.ensure_duration_updater() return card_data["card"] def _create_card_components(self, recording: Recording): @@ -192,6 +192,48 @@ def create_status_label(self, recording: Recording): alignment=ft.alignment.center, ) + def ensure_duration_updater(self): + if self.duration_updater_task is None: + self.duration_updater_task = self.app.page.run_task(self.update_duration_loop) + + @staticmethod + def safe_update(control, error_message: str): + try: + control.update() + return True + except (ft.core.page.PageDisconnectedException, AssertionError) as e: + logger.debug(f"{error_message}: {e}") + return False + + def sync_status_label(self, recording_card: dict, recording: Recording): + status_config = RecordingCardState.get_status_label_config(recording, self._) + card = recording_card.get("card") + if not card or not card.content or not card.content.content: + return + + title_row = card.content.content.controls[0] + title_row.alignment = ft.MainAxisAlignment.START + title_row.spacing = 5 + title_row.tight = True + + status_label = recording_card.get("status_label") + if not status_config: + if status_label and len(title_row.controls) > 1: + title_row.controls.pop() + recording_card["status_label"] = None + return + + if status_label is None: + status_label = self.create_status_label(recording) + title_row.controls.append(status_label) + recording_card["status_label"] = status_label + return + + if isinstance(status_label.content, ft.Text): + status_label.content.value = status_config["text"] + status_label.content.color = status_config["text_color"] + status_label.bgcolor = status_config["bgcolor"] + async def update_card(self, recording): """Update only the recordings cards in the scrollable content area.""" if recording.rec_id in self.cards_obj: @@ -203,23 +245,7 @@ async def update_card(self, recording): recording_card["display_title_label"].value = display_title recording_card["display_title_label"].weight = RecordingCardState.get_title_weight(recording) - new_status_label = self.create_status_label(recording) - - if recording_card["card"] and recording_card["card"].content and recording_card["card"].content.content: - title_row = recording_card["card"].content.content.controls[0] - title_row.alignment = ft.MainAxisAlignment.START - title_row.spacing = 5 - title_row.tight = True - - # Update the status label if it exists - if new_status_label: - if len(title_row.controls) > 1: - title_row.controls[1] = new_status_label - else: - title_row.controls.append(new_status_label) - else: - if len(title_row.controls) > 1: - title_row.controls.pop() + self.sync_status_label(recording_card, recording) if recording_card.get("duration_label"): recording_card["duration_label"].value = self.app.record_manager.get_duration(recording) @@ -238,10 +264,7 @@ async def update_card(self, recording): if recording_card["card"] and recording_card["card"].content: recording_card["card"].content.bgcolor = self.get_card_background_color(recording) recording_card["card"].content.border = ft.border.all(2, self.get_card_border_color(recording)) - try: - self.app.page.update() - except (ft.core.page.PageDisconnectedException, AssertionError) as e: - logger.debug(f"Update card failed: {e}") + if not self.safe_update(recording_card["card"], "Update card failed"): return except (ft.core.page.PageDisconnectedException, AssertionError) as e: @@ -364,6 +387,10 @@ async def remove_recording_card(self, recordings: list[Recording]): k: v for k, v in self.cards_obj.items() if k in keep_ids } + self.selected_cards = { + k: v for k, v in self.selected_cards.items() + if k in keep_ids + } try: recordings_page.recording_card_area.update() @@ -395,39 +422,43 @@ def get_icon_for_monitor_state(recording: Recording): def get_tip_for_monitor_state(self, recording: Recording): return self._["stop_monitor"] if recording.monitor_status else self._["start_monitor"] - async def update_duration(self, recording: Recording): - """Update the duration text periodically.""" + async def update_duration_loop(self): + """Refresh duration labels with a single shared background loop.""" while True: - update_interval = 1 - await asyncio.sleep(update_interval) - if not recording or recording.rec_id not in self.cards_obj: # Stop task if card is removed - break + await asyncio.sleep(1) + if not self.cards_obj or not hasattr(self.app, "record_manager"): + continue - if recording.is_recording: - try: - duration_label = self.cards_obj[recording.rec_id]["duration_label"] - duration_label.value = self.app.record_manager.get_duration(recording) - duration_label.update() - except (ft.core.page.PageDisconnectedException, AssertionError) as e: - logger.debug(f"Update duration failed: {e}") - break - except Exception as e: - logger.debug(f"Update duration failed: {e}") - - def start_update_task(self, recording: Recording): - """Start a background task to update the duration text.""" - self.update_duration_tasks[recording.rec_id] = self.app.page.run_task(self.update_duration, recording) + try: + for rec_id, card_info in list(self.cards_obj.items()): + recording = self.app.record_manager.find_recording_by_id(rec_id) + if not recording or not recording.is_recording: + continue + + card = card_info.get("card") + duration_label = card_info.get("duration_label") + if not card or not duration_label or not card.visible: + continue + + new_duration = self.app.record_manager.get_duration(recording) + if duration_label.value != new_duration: + duration_label.value = new_duration + if not self.safe_update(duration_label, "Update duration failed"): + self.duration_updater_task = None + return + except Exception as e: + logger.debug(f"Update duration failed: {e}") async def on_card_click(self, recording: Recording): """Handle card click events.""" try: recording.selected = not recording.selected - self.selected_cards[recording.rec_id] = recording + if recording.selected: + self.selected_cards[recording.rec_id] = recording + else: + self.selected_cards.pop(recording.rec_id, None) self.cards_obj[recording.rec_id]["card"].content.bgcolor = await self.update_record_hover(recording) - try: - self.cards_obj[recording.rec_id]["card"].update() - except (ft.core.page.PageDisconnectedException, AssertionError) as e: - logger.debug(f"Update card click state failed: {e}") + self.safe_update(self.cards_obj[recording.rec_id]["card"], "Update card click state failed") except (ft.core.page.PageDisconnectedException, AssertionError) as e: logger.debug(f"Handle card click event failed: {e}") except Exception as e: diff --git a/app/ui/views/recordings_view.py b/app/ui/views/recordings_view.py index 3f5f4843..7e84321f 100644 --- a/app/ui/views/recordings_view.py +++ b/app/ui/views/recordings_view.py @@ -41,22 +41,8 @@ def init(self): stroke_width=3, visible=False ) - - if self.is_grid_view: - initial_content = ft.GridView( - expand=True, - runs_count=3, - spacing=10, - run_spacing=10, - child_aspect_ratio=2.3, - controls=[] - ) - else: - initial_content = ft.Column( - controls=[], - spacing=5, - expand=True - ) + + initial_content = self._create_grid_content() if self.is_grid_view else self._create_list_content() self.recording_card_area = ft.Container( content=initial_content, @@ -65,6 +51,30 @@ def init(self): self.add_recording_dialog = RecordingDialog(self.app, self.add_recording) self.pubsub_subscribe() + @staticmethod + def _create_grid_content(controls=None): + return ft.GridView( + expand=True, + runs_count=3, + spacing=10, + run_spacing=10, + child_aspect_ratio=2.3, + cache_extent=600, + controls=list(controls or []) + ) + + @staticmethod + def _create_list_content(controls=None): + return ft.ListView( + expand=True, + spacing=5, + padding=0, + build_controls_on_demand=True, + cache_extent=600, + first_item_prototype=True, + controls=list(controls or []) + ) + async def load(self): """Load the recordings page content.""" self.content_area.controls.extend( @@ -96,26 +106,16 @@ def pubsub_subscribe(self): async def toggle_view_mode(self, _): self.is_grid_view = not self.is_grid_view current_content = self.recording_card_area.content - current_controls = current_content.controls if hasattr(current_content, 'controls') else [] + current_controls = list(current_content.controls) if hasattr(current_content, "controls") else [] column_width = 350 runs_count = max(1, int(self.page.width / column_width)) if self.is_grid_view: - new_content = ft.GridView( - expand=True, - runs_count=runs_count, - spacing=10, - run_spacing=10, - child_aspect_ratio=2.3, - controls=current_controls - ) + new_content = self._create_grid_content(current_controls) + new_content.runs_count = runs_count else: - new_content = ft.Column( - controls=current_controls, - spacing=5, - expand=True - ) + new_content = self._create_list_content(current_controls) self.recording_card_area.content = new_content self.content_area.clean() @@ -362,6 +362,7 @@ async def apply_filter(self): cards_obj = self.app.record_card_manager.cards_obj recordings = self.app.record_manager.recordings + visibility_changed = False for recording in recordings: card_info = cards_obj.get(recording.rec_id) @@ -372,17 +373,23 @@ async def apply_filter(self): platform_visible = RecordingFilters.get_platform_filter_result(recording, self.current_platform_filter) visible = status_visible and platform_visible - card_info["card"].visible = visible + if card_info["card"].visible != visible: + card_info["card"].visible = visible + visibility_changed = True self.content_area.update() - self.recording_card_area.update() + if visibility_changed: + self.recording_card_area.update() async def reset_cards_visibility(self): cards_obj = self.app.record_card_manager.cards_obj + visibility_changed = False for card_info in cards_obj.values(): if not card_info["card"].visible: card_info["card"].visible = True - card_info["card"].update() + visibility_changed = True + if visibility_changed: + self.recording_card_area.update() async def filter_recordings(self, query): recordings = self.app.record_manager.recordings @@ -409,9 +416,15 @@ async def filter_recordings(self, query): if RecordingFilters.should_show_recording(self.current_filter, self.current_platform_filter, recording): filtered_ids.add(rec_id) + visibility_changed = False for card_info in cards_obj.values(): - card_info["card"].visible = card_info["card"].key in filtered_ids - card_info["card"].update() + visible = card_info["card"].key in filtered_ids + if card_info["card"].visible != visible: + card_info["card"].visible = visible + visibility_changed = True + + if visibility_changed: + self.recording_card_area.update() if not filtered_ids: await self.app.snack_bar.show_snack_bar(self._["not_search_result"], duration=2000) @@ -420,6 +433,7 @@ async def filter_recordings(self, query): def create_recordings_content_area(self): return ft.Column( expand=True, + spacing=0, controls=[ ft.Divider(height=1), ft.Container( @@ -428,7 +442,6 @@ def create_recordings_content_area(self): ), self.recording_card_area, ], - scroll=ft.ScrollMode.AUTO if not self.app.is_mobile else ft.ScrollMode.HIDDEN, ) async def add_record_cards(self): @@ -592,13 +605,14 @@ async def refresh_cards_on_click(self, _e): continue if card_id in selected_cards: selected_cards[card_id].selected = False + selected_cards.pop(card_id, None) card["card"].content.bgcolor = None - card["card"].update() for card in to_remove: card_key = card["card"].key cards_obj.pop(card_key, None) - self.recording_card_area.controls.remove(card["card"]) + selected_cards.pop(card_key, None) + self.recording_card_area.content.controls.remove(card["card"]) await self.show_all_cards() self.content_area.controls[1] = self.create_filter_area() @@ -666,6 +680,7 @@ async def delete_all_recording_cards(self): self.recording_card_area.content.controls.clear() self.recording_card_area.update() self.app.record_card_manager.cards_obj = {} + self.app.record_card_manager.selected_cards = {} self.current_platform_filter = "all" @@ -716,15 +731,33 @@ async def recalculate_grid_columns(self): if isinstance(self.recording_card_area.content, ft.GridView): grid_view = self.recording_card_area.content - grid_view.runs_count = runs_count - - grid_view.child_aspect_ratio = child_aspect_ratio - + needs_update = False + + if grid_view.runs_count != runs_count: + grid_view.runs_count = runs_count + needs_update = True + + if grid_view.child_aspect_ratio != child_aspect_ratio: + grid_view.child_aspect_ratio = child_aspect_ratio + needs_update = True + if self.app.is_mobile: - grid_view.spacing = 5 - grid_view.run_spacing = 5 - - grid_view.update() + if grid_view.spacing != 5: + grid_view.spacing = 5 + needs_update = True + if grid_view.run_spacing != 5: + grid_view.run_spacing = 5 + needs_update = True + else: + if grid_view.spacing != 10: + grid_view.spacing = 10 + needs_update = True + if grid_view.run_spacing != 10: + grid_view.run_spacing = 10 + needs_update = True + + if needs_update: + grid_view.update() async def on_keyboard(self, e: ft.KeyboardEvent): if e.alt and e.key == "H": From 70e871746704f69ea9081cfcfa7b5edcb4552c1f Mon Sep 17 00:00:00 2001 From: HH Date: Fri, 17 Apr 2026 15:14:23 +0800 Subject: [PATCH 02/10] =?UTF-8?q?=E8=A7=A3=E5=86=B3=E7=9B=B4=E6=92=AD?= =?UTF-8?q?=E6=B5=81=E5=85=B3=E9=97=AD=E5=90=8E=EF=BC=8C=E5=BD=95=E5=88=B6?= =?UTF-8?q?=E6=9C=AA=E6=AD=A3=E5=B8=B8=E7=BB=93=E6=9D=9F=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/core/media/direct_downloader.py | 10 +- app/core/media/ffmpeg_builders/base.py | 5 +- app/core/recording/stream_manager.py | 127 ++++++++++++++++++++++++- 3 files changed, 134 insertions(+), 8 deletions(-) diff --git a/app/core/media/direct_downloader.py b/app/core/media/direct_downloader.py index 7c96b9ad..80afbde7 100644 --- a/app/core/media/direct_downloader.py +++ b/app/core/media/direct_downloader.py @@ -29,9 +29,12 @@ def __init__(self, self.download_task = None self.total_bytes = 0 self.start_time = None + self.last_chunk_time = None + self.read_timeout = 45.0 async def start_download(self) -> bool: self.start_time = time.time() + self.last_chunk_time = self.start_time self.download_task = asyncio.create_task(self._download_stream()) return True @@ -50,7 +53,9 @@ async def _download_stream(self) -> None: try: os.makedirs(os.path.dirname(self.save_path), exist_ok=True) - async with httpx.AsyncClient(headers=self.headers, proxy=self.proxy, timeout=None) as client: + timeout = httpx.Timeout(connect=15.0, read=self.read_timeout, write=30.0, pool=None) + + async with httpx.AsyncClient(headers=self.headers, proxy=self.proxy, timeout=timeout) as client: async with client.stream("GET", self.record_url) as response: if response.status_code != 200: logger.error(f"Request Stream Failed, Status Code: {response.status_code}") @@ -63,6 +68,7 @@ async def _download_stream(self) -> None: f.write(chunk) self.total_bytes += len(chunk) + self.last_chunk_time = time.time() # Please don't remove this comment code # elapsed = time.time() - self.start_time @@ -75,5 +81,7 @@ async def _download_stream(self) -> None: except asyncio.CancelledError: logger.info(f"Download Task Canceled: {self.record_url}") + except httpx.ReadTimeout: + logger.warning(f"Download stalled or stream ended due to read timeout: {self.record_url}") except Exception as e: logger.error(f"Download Error: {e}") diff --git a/app/core/media/ffmpeg_builders/base.py b/app/core/media/ffmpeg_builders/base.py index cf008ab7..bd1cd66e 100644 --- a/app/core/media/ffmpeg_builders/base.py +++ b/app/core/media/ffmpeg_builders/base.py @@ -85,9 +85,10 @@ def _get_basic_ffmpeg_command(self) -> list[str]: "-bufsize", config["bufsize"], "-sn", "-dn", + "-reconnect", "1", "-reconnect_delay_max", "60", - "-reconnect_streamed", - "-reconnect_at_eof", + "-reconnect_streamed", "1", + "-reconnect_on_network_error", "1", "-max_muxing_queue_size", config["max_muxing_queue_size"], "-correct_ts_overflow", "1", "-avoid_negative_ts", "1", diff --git a/app/core/recording/stream_manager.py b/app/core/recording/stream_manager.py index 6e1dd176..83159251 100644 --- a/app/core/recording/stream_manager.py +++ b/app/core/recording/stream_manager.py @@ -24,6 +24,8 @@ class LiveStreamRecorder: DEFAULT_SEGMENT_TIME = "1800" DEFAULT_SAVE_FORMAT = "mp4" DEFAULT_QUALITY = VideoQuality.OD + STALL_TIMEOUT_SECONDS = 45 + MIN_VALID_OUTPUT_BYTES = 1024 def __init__(self, app, recording, recording_info): self.app = app @@ -32,6 +34,7 @@ def __init__(self, app, recording, recording_info): self.recording_info = recording_info self.subprocess_start_info = app.subprocess_start_up_info self.should_stop = False # manually stopped + self.auto_stop_requested = False self.user_config = self.settings.user_config self.account_config = self.settings.accounts_config @@ -146,6 +149,64 @@ def _clean_and_truncate_title(title: str) -> str | None: cleaned_title = title[:30].replace("，", ",").replace(" ", "") return cleaned_title + def _get_output_files(self, save_file_path: str) -> list[str]: + normalized_path = save_file_path.replace("\\", "/") + if not self.segment_record: + return [normalized_path] if os.path.exists(normalized_path) else [] + + directory = os.path.dirname(normalized_path) + if not os.path.isdir(directory): + return [] + + filename = os.path.basename(normalized_path) + segment_marker = "_%03d." + if segment_marker in filename: + prefix, suffix = filename.split(segment_marker, maxsplit=1) + suffix = "." + suffix + else: + prefix, suffix = os.path.splitext(filename) + + output_files = [] + for name in os.listdir(directory): + full_path = os.path.join(directory, name).replace("\\", "/") + if os.path.isfile(full_path) and name.startswith(prefix) and name.endswith(suffix): + output_files.append(full_path) + return sorted(output_files) + + def _get_output_size(self, save_file_path: str) -> int: + total_size = 0 + for output_file in self._get_output_files(save_file_path): + try: + total_size += os.path.getsize(output_file) + except OSError: + continue + return total_size + + def _cleanup_invalid_output_files(self, save_file_path: str) -> bool: + valid_output_found = False + removed_files = [] + + for output_file in self._get_output_files(save_file_path): + try: + file_size = os.path.getsize(output_file) + except OSError: + continue + + if file_size >= self.MIN_VALID_OUTPUT_BYTES: + valid_output_found = True + continue + + try: + os.remove(output_file) + removed_files.append(output_file) + except OSError as e: + logger.warning(f"Failed to remove incomplete recording file {output_file}: {e}") + + if removed_files: + logger.warning(f"Removed incomplete recording files: {removed_files}") + + return valid_output_found + @property def is_flv_preferred_platform(self): return self.platform_key in {"douyin", "tiktok"} @@ -322,9 +383,12 @@ async def start_ffmpeg( logger.info(f"Starting ffmpeg recording - recorder id: {id(self)}, rec_id: {self.recording.rec_id}") self.should_stop = False + self.auto_stop_requested = False try: save_file_path = ffmpeg_command[-1] + stalled_recording = False + valid_output = True process = await asyncio.create_subprocess_exec( *ffmpeg_command, @@ -340,9 +404,25 @@ async def start_ffmpeg( logger.info(f"Recording in Progress: {live_url}") logger.log("STREAM", f"Recording Stream URL: {record_url}") self.recording_start_time = time.time() + last_output_size = 0 + last_output_at = self.recording_start_time while True: - if self.should_stop or self.recording.force_stop or not self.app.recording_enabled: + current_output_size = self._get_output_size(save_file_path) + if current_output_size > last_output_size: + last_output_size = current_output_size + last_output_at = time.time() + elif ( + time.time() - self.recording_start_time > self.min_valid_recording_duration + and time.time() - last_output_at > self.STALL_TIMEOUT_SECONDS + ): + logger.warning( + f"Recording output stalled for {self.STALL_TIMEOUT_SECONDS}s, stopping recorder: {live_url}" + ) + self.auto_stop_requested = True + stalled_recording = True + + if self.should_stop or self.auto_stop_requested or self.recording.force_stop or not self.app.recording_enabled: logger.info(f"Preparing to End Recording: {live_url}") await self.remove_active_recorder() self.recording.is_recording = False @@ -381,7 +461,8 @@ async def start_ffmpeg( return_code = process.returncode safe_return_code = [0, 255] stdout, stderr = await process.communicate() - + valid_output = self._cleanup_invalid_output_files(save_file_path) + if return_code not in safe_return_code and stderr: if not self.recording.is_recording: logger.error(f"FFmpeg Stderr Output: {str(stderr.decode()).splitlines()[0]}") @@ -410,9 +491,12 @@ async def start_ffmpeg( self.recording.live_title = None if self.should_stop: logger.success(f"Live recording has stopped: {record_name}") + elif stalled_recording: + logger.warning(f"Live recording stopped after stream stalled: {record_name}") else: logger.success(f"Live recording completed: {record_name}") - self.app.page.run_task(self.end_message_push) + if valid_output: + self.app.page.run_task(self.end_message_push) try: self.recording.update({"display_title": display_title}) @@ -427,6 +511,10 @@ async def start_ffmpeg( await self.recheck_live_status() + if not valid_output: + logger.warning(f"Discarded invalid recording output: {save_file_path}") + return True + if self.user_config.get("convert_to_mp4") and self.save_format == "ts": if self.segment_record: file_paths = utils.get_file_paths(os.path.dirname(save_file_path)) @@ -489,6 +577,7 @@ async def start_ffmpeg( return False finally: self.recording.record_url = None + self.auto_stop_requested = False return True @@ -675,18 +764,37 @@ async def start_direct_download( logger.info(f"Starting direct download - recorder id: {id(self)}, rec_id: {self.recording.rec_id}") self.should_stop = False + self.auto_stop_requested = False try: await self.direct_downloader.start_download() + stalled_recording = False + valid_output = True self.recording.status_info = RecordingStatus.RECORDING self.recording.record_url = record_url logger.info(f"Direct Downloading: {live_url}") logger.log("STREAM", f"Direct Download Stream URL: {record_url}") self.recording_start_time = time.time() + last_output_size = 0 + last_output_at = self.recording_start_time while True: - if self.should_stop or self.recording.force_stop or not self.app.recording_enabled: + current_output_size = self._get_output_size(save_file_path) + if current_output_size > last_output_size: + last_output_size = current_output_size + last_output_at = time.time() + elif ( + time.time() - self.recording_start_time > self.min_valid_recording_duration + and time.time() - last_output_at > self.STALL_TIMEOUT_SECONDS + ): + logger.warning( + f"Direct download stalled for {self.STALL_TIMEOUT_SECONDS}s, stopping recorder: {live_url}" + ) + self.auto_stop_requested = True + stalled_recording = True + + if self.should_stop or self.auto_stop_requested or self.recording.force_stop or not self.app.recording_enabled: logger.info(f"Prepare to end direct download: {live_url}") await self.remove_active_recorder() self.recording.is_recording = False @@ -701,6 +809,7 @@ async def start_direct_download( await self.remove_active_recorder() self.recording.is_recording = False + valid_output = self._cleanup_invalid_output_files(save_file_path) if not self.recording.is_recording: self.recording.is_live = False @@ -714,9 +823,12 @@ async def start_direct_download( self.recording.live_title = None if self.should_stop: logger.success(f"Direct Downloading Stopped: {record_name}") + elif stalled_recording: + logger.warning(f"Direct download stopped after stream stalled: {record_name}") else: logger.success(f"Direct Downloading Completed: {record_name}") - self.app.page.run_task(self.end_message_push) + if valid_output: + self.app.page.run_task(self.end_message_push) try: self.recording.update({"display_title": display_title}) @@ -731,6 +843,10 @@ async def start_direct_download( await self.recheck_live_status() + if not valid_output: + logger.warning(f"Discarded invalid direct download output: {save_file_path}") + return True + if self.user_config.get("execute_custom_script") and script_command: logger.info("Prepare to execute custom script in the background") try: @@ -773,6 +889,7 @@ async def start_direct_download( return False finally: self.recording.record_url = None + self.auto_stop_requested = False async def stop_recording_notify(self): if desktop_notify.should_push_notification(self.app): From ac7cb51d184bc2ad205ae98ffdb129a42f40947d Mon Sep 17 00:00:00 2001 From: HH Date: Sat, 18 Apr 2026 02:05:43 +0800 Subject: [PATCH 03/10] =?UTF-8?q?=E4=BC=98=E5=8C=96=E7=BC=BA=E5=B0=91?= =?UTF-8?q?=E9=85=8D=E7=BD=AE=E6=96=87=E4=BB=B6=E6=97=B6=E7=9A=84=E9=80=BB?= =?UTF-8?q?=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/app_manager.py | 1 + app/ui/views/recordings_view.py | 57 +++++++++++++++++++-------------- app/ui/views/settings_view.py | 5 ++- 3 files changed, 38 insertions(+), 25 deletions(-) diff --git a/app/app_manager.py b/app/app_manager.py index 99d59a7e..3d149123 100644 --- a/app/app_manager.py +++ b/app/app_manager.py @@ -92,6 +92,7 @@ async def switch_page(self, page_name): self._loading_page = True try: + self.page.on_resized = None await self.clear_content_area() if page := self.pages.get(page_name): await self.settings.is_changed() diff --git a/app/ui/views/recordings_view.py b/app/ui/views/recordings_view.py index 7e84321f..6bf84ef0 100644 --- a/app/ui/views/recordings_view.py +++ b/app/ui/views/recordings_view.py @@ -329,6 +329,20 @@ def create_filter_area(self): spacing=5, vertical_alignment=ft.CrossAxisAlignment.CENTER, ) + + def _is_active_page(self) -> bool: + return self.app.current_page == self + + def _refresh_filter_area(self) -> None: + if not self._is_active_page(): + return + + if len(self.content_area.controls) > 1: + self.content_area.controls[1] = self.create_filter_area() + else: + self.content_area.controls.append(self.create_filter_area()) + + self.content_area.update() async def filter_all_on_click(self, _): self.current_filter = "all" @@ -355,10 +369,7 @@ async def filter_stopped_on_click(self, _): await self.apply_filter() async def apply_filter(self): - if len(self.content_area.controls) > 1: - self.content_area.controls[1] = self.create_filter_area() - else: - self.content_area.controls.append(self.create_filter_area()) + self._refresh_filter_area() cards_obj = self.app.record_card_manager.cards_obj recordings = self.app.record_manager.recordings @@ -376,9 +387,8 @@ async def apply_filter(self): if card_info["card"].visible != visible: card_info["card"].visible = visible visibility_changed = True - - self.content_area.update() - if visibility_changed: + + if visibility_changed and self._is_active_page(): self.recording_card_area.update() async def reset_cards_visibility(self): @@ -572,10 +582,10 @@ async def create_card_with_time_range(rec): self.app.record_card_manager.cards_obj[recording.rec_id]["card"] = card self.app.page.pubsub.send_others_on_topic("add", recording) - self.recording_card_area.update() + if self._is_active_page(): + self.recording_card_area.update() - self.content_area.controls[1] = self.create_filter_area() - self.content_area.update() + self._refresh_filter_area() await self.app.snack_bar.show_snack_bar(self._["add_recording_success_tip"], bgcolor=ft.Colors.GREEN) @@ -614,9 +624,7 @@ async def refresh_cards_on_click(self, _e): selected_cards.pop(card_key, None) self.recording_card_area.content.controls.remove(card["card"]) await self.show_all_cards() - - self.content_area.controls[1] = self.create_filter_area() - self.content_area.update() + self._refresh_filter_area() self.loading_indicator.visible = False self.loading_indicator.update() @@ -648,10 +656,10 @@ async def confirm_dlg(_): await self.delete_all_recording_cards() self.app.page.pubsub.send_others_on_topic("delete_all", None) - self.content_area.controls[1] = self.create_filter_area() - self.content_area.update() + self._refresh_filter_area() - self.recording_card_area.update() + if self._is_active_page(): + self.recording_card_area.update() await self.app.snack_bar.show_snack_bar( self._["delete_recording_success_tip"], bgcolor=ft.Colors.GREEN, duration=2000 ) @@ -683,9 +691,8 @@ async def delete_all_recording_cards(self): self.app.record_card_manager.selected_cards = {} self.current_platform_filter = "all" - - self.content_area.controls[1] = self.create_filter_area() - self.content_area.update() + + self._refresh_filter_area() async def subscribe_del_all_cards(self, *_): await self.delete_all_recording_cards() @@ -708,16 +715,18 @@ async def subscribe_add_cards(self, _, recording: Recording): self.loading_indicator.visible = False self.loading_indicator.update() - self.recording_card_area.update() - - self.content_area.controls[1] = self.create_filter_area() - self.content_area.update() + if self._is_active_page(): + self.recording_card_area.update() + + self._refresh_filter_area() async def update_grid_layout(self, _): + if not self._is_active_page(): + return self.page.run_task(self.recalculate_grid_columns) async def recalculate_grid_columns(self): - if not self.is_grid_view: + if not self.is_grid_view or not self._is_active_page(): return if self.app.is_mobile: diff --git a/app/ui/views/settings_view.py b/app/ui/views/settings_view.py index fc702f3e..80366aab 100644 --- a/app/ui/views/settings_view.py +++ b/app/ui/views/settings_view.py @@ -103,7 +103,10 @@ def init_unsaved_changes(self): } def load_language(self): - self.default_language, default_language_code = list(self.language_option.items())[0] + if not self.language_option: + self.language_option = {"Chinese": "zh_CN"} + + self.default_language, default_language_code = next(iter(self.language_option.items())) select_language = self.user_config.get("language") self.language_code = self.language_option.get(select_language, default_language_code) self.app.language_code = self.language_code From e2f2a8f0284df9a2b46c8b1454d7fd69d63a5e9a Mon Sep 17 00:00:00 2001 From: HH Date: Sat, 18 Apr 2026 16:43:50 +0800 Subject: [PATCH 04/10] Add proxy subscription rotation support --- app/app_manager.py | 3 + app/core/config/proxy_manager.py | 293 ++++++++++++++++++ .../platforms/platform_handlers/handlers.py | 5 + app/core/recording/stream_manager.py | 73 ++++- app/messages/message_pusher.py | 2 +- app/ui/views/settings_view.py | 24 ++ config/default_settings.json | 4 +- main.py | 2 + 8 files changed, 388 insertions(+), 18 deletions(-) create mode 100644 app/core/config/proxy_manager.py diff --git a/app/app_manager.py b/app/app_manager.py index 3d149123..d2cca239 100644 --- a/app/app_manager.py +++ b/app/app_manager.py @@ -6,6 +6,7 @@ from . import execute_dir from .core.config.config_manager import ConfigManager from .core.config.language_manager import LanguageManager +from .core.config.proxy_manager import ProxyManager from .core.recording.record_manager import RecordingManager from .core.runtime.process_manager import AsyncProcessManager from .core.update.update_checker import UpdateChecker @@ -41,6 +42,7 @@ def __init__(self, page: ft.Page): ) self.settings = SettingsPage(self) + self.proxy_manager = ProxyManager(self) self.language_manager = LanguageManager(self) self.language_code = self.settings.language_code self.about = AboutPage(self) @@ -138,6 +140,7 @@ async def _check_for_updates(self): async def start_periodic_tasks(self): """Start all periodic tasks""" + await self.proxy_manager.start() await self.record_manager.setup_periodic_live_check( int(self.record_manager.loop_time_seconds or 180) ) diff --git a/app/core/config/proxy_manager.py b/app/core/config/proxy_manager.py new file mode 100644 index 00000000..456f0ed3 --- /dev/null +++ b/app/core/config/proxy_manager.py @@ -0,0 +1,293 @@ +import asyncio +import json +import re +import threading +from urllib.parse import quote, urlsplit, urlunsplit + +import httpx + +from ...utils.logger import logger + + +class ProxyManager: + SUBSCRIPTION_REFRESH_SECONDS = 300 + SUBSCRIPTION_TIMEOUT_SECONDS = 15.0 + + def __init__(self, app): + self.app = app + self.current_proxy_address = None + self.current_proxy_username = "" + self.current_proxy_password = "" + self.subscription_proxy_addresses = [] + self.subscription_proxy_username = "" + self.subscription_proxy_password = "" + self._subscription_proxy_index = 0 + self._subscription_task = None + self._refresh_lock = asyncio.Lock() + self._subscription_proxy_lock = threading.Lock() + + @staticmethod + def is_subscription_url(proxy_address: str | None) -> bool: + if not proxy_address: + return False + proxy_address = proxy_address.strip().lower() + return proxy_address.startswith("http://") or proxy_address.startswith("https://") + + @staticmethod + def _normalize_list_value(raw_value) -> list[str]: + if raw_value is None: + return [] + + if isinstance(raw_value, list): + values = raw_value + else: + values = [raw_value] + + normalized_values = [] + for value in values: + text = str(value).strip().strip('"').strip("'") + if text: + normalized_values.append(text) + return normalized_values + + @classmethod + def _parse_subscription_payload(cls, payload_text: str) -> tuple[list[str], str, str]: + try: + payload = json.loads(payload_text) + except json.JSONDecodeError: + payload = None + + if isinstance(payload, dict): + return ( + cls._normalize_list_value(payload.get("ip")), + str(payload.get("user") or "").strip(), + str(payload.get("pwd") or "").strip(), + ) + + ip_match = re.search(r'"ip"\s*:\s*\[(.*?)\]', payload_text, re.S) + user_match = re.search(r'"user"\s*:\s*"([^"]*)"', payload_text) + pwd_match = re.search(r'"pwd"\s*:\s*"([^"]*)"', payload_text) + + ip_values = [] + if ip_match: + ip_values = cls._normalize_list_value(ip_match.group(1).split(",")) + + return ( + ip_values, + user_match.group(1).strip() if user_match else "", + pwd_match.group(1).strip() if pwd_match else "", + ) + + @staticmethod + def _build_proxy_value(address: str | None, username: str = "", password: str = "") -> str | None: + if not address: + return None + + address = str(address).strip() + username = str(username or "").strip() + password = str(password or "").strip() + + if not username and not password: + return address + + auth = quote(username, safe="") + if password: + auth = f"{auth}:{quote(password, safe='')}" + + if "://" in address: + parsed = urlsplit(address) + host = parsed.hostname or "" + if parsed.port: + host = f"{host}:{parsed.port}" + netloc = f"{auth}@{host}" if auth else host + return urlunsplit((parsed.scheme, netloc, parsed.path, parsed.query, parsed.fragment)) + + return f"http://{auth}@{address}" + + @classmethod + def _build_subscription_proxy_value( + cls, + address: str | None, + username: str = "", + password: str = "", + ) -> str | None: + proxy_value = cls._build_proxy_value(address, username, password) + if not proxy_value: + return None + + if "://" in proxy_value: + return proxy_value + + return f"http://{proxy_value}" + + @staticmethod + def _resolve_credentials( + subscription_username: str, + subscription_password: str, + fallback_username: str, + fallback_password: str, + ) -> tuple[str, str]: + subscription_username = str(subscription_username or "").strip() + subscription_password = str(subscription_password or "").strip() + fallback_username = str(fallback_username or "").strip() + fallback_password = str(fallback_password or "").strip() + + if subscription_username and subscription_password: + return subscription_username, subscription_password + + return fallback_username, fallback_password + + @staticmethod + def mask_proxy_value(proxy_value: str | None) -> str | None: + if not proxy_value: + return None + + if "@" not in proxy_value: + return proxy_value + + if "://" in proxy_value: + scheme, remainder = proxy_value.split("://", maxsplit=1) + _, host = remainder.rsplit("@", maxsplit=1) + return f"{scheme}://***@{host}" + + _, host = proxy_value.rsplit("@", maxsplit=1) + return f"***@{host}" + + def _set_current_proxy(self, address: str | None, username: str = "", password: str = "") -> None: + self.current_proxy_address = address.strip() if isinstance(address, str) else address + self.current_proxy_username = str(username or "").strip() + self.current_proxy_password = str(password or "").strip() + + def _set_subscription_proxies(self, addresses: list[str], username: str = "", password: str = "") -> None: + normalized_addresses = self._normalize_list_value(addresses) + self.subscription_proxy_addresses = normalized_addresses + self.subscription_proxy_username = str(username or "").strip() + self.subscription_proxy_password = str(password or "").strip() + + with self._subscription_proxy_lock: + if not normalized_addresses: + self._subscription_proxy_index = 0 + else: + self._subscription_proxy_index %= len(normalized_addresses) + + def _clear_subscription_proxies(self) -> None: + self.subscription_proxy_addresses = [] + self.subscription_proxy_username = "" + self.subscription_proxy_password = "" + with self._subscription_proxy_lock: + self._subscription_proxy_index = 0 + + def clear(self) -> None: + self._clear_subscription_proxies() + self._set_current_proxy(None) + + def get_proxy(self) -> str | None: + if not self.app.settings.user_config.get("enable_proxy"): + return None + return self._build_proxy_value( + self.current_proxy_address, + self.current_proxy_username, + self.current_proxy_password, + ) + + def is_subscription_active(self) -> bool: + if not self.app.settings.user_config.get("enable_proxy"): + return False + + proxy_address = str(self.app.settings.user_config.get("proxy_address") or "").strip() + return self.is_subscription_url(proxy_address) and bool(self.subscription_proxy_addresses) + + def get_status_check_proxy(self) -> str | None: + if not self.app.settings.user_config.get("enable_proxy"): + return None + + if not self.is_subscription_active(): + return self.get_proxy() + + with self._subscription_proxy_lock: + if not self.subscription_proxy_addresses: + return self.get_proxy() + + selected_proxy = self.subscription_proxy_addresses[self._subscription_proxy_index] + self._subscription_proxy_index = (self._subscription_proxy_index + 1) % len(self.subscription_proxy_addresses) + + return self._build_subscription_proxy_value( + selected_proxy, + self.subscription_proxy_username, + self.subscription_proxy_password, + ) + + async def sync_from_settings(self) -> None: + async with self._refresh_lock: + proxy_enabled = self.app.settings.user_config.get("enable_proxy") + proxy_address = str(self.app.settings.user_config.get("proxy_address") or "").strip() + proxy_username = str(self.app.settings.user_config.get("proxy_username") or "").strip() + proxy_password = str(self.app.settings.user_config.get("proxy_password") or "").strip() + + if not proxy_enabled or not proxy_address: + self.clear() + return + + if self.is_subscription_url(proxy_address): + await self._refresh_subscription_locked(proxy_address, proxy_username, proxy_password) + return + + self._clear_subscription_proxies() + self._set_current_proxy(proxy_address, proxy_username, proxy_password) + + async def _refresh_subscription_locked( + self, + subscription_url: str, + fallback_username: str = "", + fallback_password: str = "", + ) -> None: + try: + timeout = httpx.Timeout(self.SUBSCRIPTION_TIMEOUT_SECONDS, connect=self.SUBSCRIPTION_TIMEOUT_SECONDS) + async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client: + response = await client.get(subscription_url) + response.raise_for_status() + payload_text = response.text + + ip_list, subscription_user, subscription_password = self._parse_subscription_payload(payload_text) + if not ip_list: + logger.warning(f"Proxy subscription returned no proxy endpoints: {subscription_url}") + return + + username, password = self._resolve_credentials( + subscription_user, + subscription_password, + fallback_username, + fallback_password, + ) + self._set_subscription_proxies(ip_list, username, password) + selected_proxy = self.subscription_proxy_addresses[0] + self._set_current_proxy(selected_proxy, username, password) + logger.info( + "Proxy subscription updated successfully: " + f"{len(self.subscription_proxy_addresses)} proxies, " + f"default={self.mask_proxy_value(self.get_proxy())}" + ) + except Exception as e: + logger.warning(f"Failed to refresh proxy subscription {subscription_url}: {e}") + + async def refresh_subscription(self) -> None: + async with self._refresh_lock: + proxy_enabled = self.app.settings.user_config.get("enable_proxy") + proxy_address = str(self.app.settings.user_config.get("proxy_address") or "").strip() + proxy_username = str(self.app.settings.user_config.get("proxy_username") or "").strip() + proxy_password = str(self.app.settings.user_config.get("proxy_password") or "").strip() + + if not proxy_enabled or not self.is_subscription_url(proxy_address): + return + + await self._refresh_subscription_locked(proxy_address, proxy_username, proxy_password) + + async def _subscription_refresh_loop(self) -> None: + while True: + await asyncio.sleep(self.SUBSCRIPTION_REFRESH_SECONDS) + await self.refresh_subscription() + + async def start(self) -> None: + await self.sync_from_settings() + if self._subscription_task is None or self._subscription_task.done(): + self._subscription_task = asyncio.create_task(self._subscription_refresh_loop()) diff --git a/app/core/platforms/platform_handlers/handlers.py b/app/core/platforms/platform_handlers/handlers.py index e07962d7..8160bb6f 100644 --- a/app/core/platforms/platform_handlers/handlers.py +++ b/app/core/platforms/platform_handlers/handlers.py @@ -46,6 +46,11 @@ async def get_stream_info(self, live_url: str) -> StreamData: """ if not self.live_stream: self.live_stream = streamget.DouyinLiveStream(proxy_addr=self.proxy, cookies=self.cookies) + else: + self.live_stream.proxy_addr = self.proxy + self.live_stream.cookies = self.cookies + self.live_stream.mobile_headers = self.live_stream._get_mobile_headers() + self.live_stream.pc_headers = self.live_stream._get_pc_headers() if "v.douyin.com" in live_url: json_data = await self.live_stream.fetch_app_stream_data(url=live_url) diff --git a/app/core/recording/stream_manager.py b/app/core/recording/stream_manager.py index 83159251..07c89529 100644 --- a/app/core/recording/stream_manager.py +++ b/app/core/recording/stream_manager.py @@ -69,9 +69,28 @@ def is_use_proxy(self): default_proxy_platform = self.user_config.get("default_platform_with_proxy", "") proxy_list = default_proxy_platform.replace("，", ",").replace(" ", "").split(",") if self.user_config.get("enable_proxy") and self.platform_key in proxy_list: - self.proxy = self.user_config.get("proxy_address") + self.proxy = self.app.proxy_manager.get_proxy() return self.proxy + def get_status_check_proxy(self): + if self.app.proxy_manager.is_subscription_active(): + return self.app.proxy_manager.get_status_check_proxy() + + default_proxy_platform = self.user_config.get("default_platform_with_proxy", "") + proxy_list = default_proxy_platform.replace("，", ",").replace(" ", "").split(",") + if self.user_config.get("enable_proxy") and self.platform_key in proxy_list: + return self.app.proxy_manager.get_status_check_proxy() + + def _get_status_check_attempts(self) -> int: + if not self.app.proxy_manager.is_subscription_active(): + return 1 + + proxy_count = len(self.app.proxy_manager.subscription_proxy_addresses) + if proxy_count <= 0: + return 1 + + return min(proxy_count, 3) + def _get_filename(self, stream_info: StreamData) -> str: live_title = None stream_info.title = utils.clean_name(stream_info.title, None) @@ -258,22 +277,44 @@ def _get_record_format(self, stream_info: StreamData): async def fetch_stream(self) -> StreamData: logger.info(f"Live URL: {self.live_url}") - logger.info(f"Use Proxy: {self.proxy or None}") - self.recording.use_proxy = bool(self.proxy) - handler = platform_handlers.get_platform_handler( - live_url=self.live_url, - proxy=self.proxy, - cookies=self.cookies, - record_quality=self.quality, - platform=self.platform, - username=self.account_config.get(self.platform_key, {}).get("username"), - password=self.account_config.get(self.platform_key, {}).get("password"), - account_type=self.account_config.get(self.platform_key, {}).get("account_type") - ) - - stream_info = await handler.get_stream_info(self.live_url) + total_attempts = self._get_status_check_attempts() + last_stream_info = None + + for attempt in range(total_attempts): + request_proxy = self.get_status_check_proxy() + masked_proxy = self.app.proxy_manager.mask_proxy_value(request_proxy) or None + + if total_attempts > 1: + logger.info(f"Use Proxy [{attempt + 1}/{total_attempts}]: {masked_proxy}") + else: + logger.info(f"Use Proxy: {masked_proxy}") + + self.recording.use_proxy = bool(request_proxy or self.proxy) + handler = platform_handlers.get_platform_handler( + live_url=self.live_url, + proxy=request_proxy, + cookies=self.cookies, + record_quality=self.quality, + platform=self.platform, + username=self.account_config.get(self.platform_key, {}).get("username"), + password=self.account_config.get(self.platform_key, {}).get("password"), + account_type=self.account_config.get(self.platform_key, {}).get("account_type") + ) + + stream_info = await handler.get_stream_info(self.live_url) + last_stream_info = stream_info + + if stream_info and getattr(stream_info, "anchor_name", None): + self.recording.is_checking = False + return stream_info + + if attempt + 1 < total_attempts: + logger.warning( + f"Fetch stream data failed with proxy {masked_proxy}, retrying next subscription proxy" + ) + self.recording.is_checking = False - return stream_info + return last_stream_info async def start_recording(self, stream_info: StreamData): """ diff --git a/app/messages/message_pusher.py b/app/messages/message_pusher.py index ccc18fc3..7f0da143 100644 --- a/app/messages/message_pusher.py +++ b/app/messages/message_pusher.py @@ -14,7 +14,7 @@ def __init__(self, settings: SettingsPage): def _get_proxy(self) -> str | None: if self.settings.user_config.get("enable_proxy"): - return self.settings.user_config.get("proxy_address") + return self.settings.app.proxy_manager.get_proxy() @staticmethod def _get_push_channels() -> list[str]: diff --git a/app/ui/views/settings_view.py b/app/ui/views/settings_view.py index 80366aab..5719488d 100644 --- a/app/ui/views/settings_view.py +++ b/app/ui/views/settings_view.py @@ -42,6 +42,9 @@ async def load(self): self.content_area.clean() language = self.app.language_manager.language self._ = language["settings_page"] | language["video_quality"] | language["base"] + login_page_language = language.get("login_page", {}) + self._["username"] = self._.get("username") or login_page_language.get("username", "Username") + self._["password"] = self._.get("password") or login_page_language.get("password", "Password") self.tab_recording = self.create_recording_settings_tab() self.tab_push = self.create_push_settings_tab() self.tab_cookies = self.create_cookies_settings_tab() @@ -131,6 +134,7 @@ async def confirm_dlg(_): self.app.language_manager.notify_observers() self.page.run_task(self.load) await self.config_manager.save_user_config(self.user_config) + await self.app.proxy_manager.sync_from_settings() logger.success("Default configuration restored.") await self.app.snack_bar.show_snack_bar(self._["success_restore_tip"], bgcolor=ft.Colors.GREEN) await close_dialog(None) @@ -200,6 +204,7 @@ async def save_user_config_after_delay(self, delay): await asyncio.sleep(delay) if self.has_unsaved_changes['user_config']: await self.config_manager.save_user_config(self.user_config) + await self.app.proxy_manager.sync_from_settings() async def save_cookies_after_delay(self, delay): await asyncio.sleep(delay) @@ -321,6 +326,25 @@ def create_recording_settings_tab(self): data="proxy_address", ), ), + self.create_setting_row( + self._["username"], + ft.TextField( + value=self.get_config_value("proxy_username", ""), + width=300, + on_change=self.on_change, + data="proxy_username", + ), + ), + self.create_setting_row( + self._["password"], + ft.TextField( + value=self.get_config_value("proxy_password", ""), + width=300, + password=True, + on_change=self.on_change, + data="proxy_password", + ), + ), ], is_mobile, ), diff --git a/config/default_settings.json b/config/default_settings.json index 4a599ca1..87739428 100644 --- a/config/default_settings.json +++ b/config/default_settings.json @@ -9,6 +9,8 @@ "folder_name_title": false, "enable_proxy": true, "proxy_address": "", + "proxy_username": "", + "proxy_password": "", "video_format": "TS", "record_quality": "OD", "loop_time_seconds": "180", @@ -68,4 +70,4 @@ "platform_max_concurrent_requests": "3", "last_route": "/home", "check_live_on_browser_refresh": false -} \ No newline at end of file +} diff --git a/main.py b/main.py index f2243eb9..397237d8 100644 --- a/main.py +++ b/main.py @@ -144,6 +144,8 @@ async def main(page: ft.Page) -> None: page.overlay.append(save_progress_overlay.overlay) async def load_app(): + await app.proxy_manager.sync_from_settings() + if is_web: setup_responsive_layout(page, app) page.on_resize = handle_page_resize(page, app) From 1790248cb5258f4c74f54efc741dae20612976ba Mon Sep 17 00:00:00 2001 From: HH Date: Sat, 18 Apr 2026 16:52:16 +0800 Subject: [PATCH 05/10] =?UTF-8?q?=E9=85=8D=E7=BD=AE=E5=8A=A0=E8=BD=BD?= =?UTF-8?q?=E6=9C=AC=E5=9C=B0streamget=E9=A1=B9=E7=9B=AE=E4=BE=9D=E8=B5=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 2 ++ pyproject.toml | 17 +++++++++++++++-- requirements-web.txt | 10 ++++++++-- requirements.txt | 10 ++++++++-- 4 files changed, 33 insertions(+), 6 deletions(-) diff --git a/.gitignore b/.gitignore index a9e4e638..4ab6f7d1 100644 --- a/.gitignore +++ b/.gitignore @@ -177,3 +177,5 @@ backup_config/ ffmpeg.exe ffplay.exe ffprobe.exe +**/.idea +bak diff --git a/pyproject.toml b/pyproject.toml index 49494d01..d7b4dffa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,9 +12,15 @@ dependencies = [ "flet[desktop,cli]==0.27.6", "flet-video==0.1.1", "httpx[http2]>=0.28.1", + "requests>=2.31.0", + "loguru>=0.7.3", + "pycryptodome>=3.20.0", + "distro>=1.9.0", + "tqdm>=4.67.1", + "PyExecJS>=1.5.1", + "deprecated>=1.2.18", "screeninfo>=0.8.1", "aiofiles>=24.1.0", - "streamget>=4.0.8", "python-dotenv>=1.0.1", "cachetools>=5.5.2", "pystray>=0.19.5", @@ -46,15 +52,22 @@ path ="." package-mode = false packages = [ { include = "app", from = "." }, + { include = "streamget", from = "." }, ] [tool.poetry.dependencies] flet = { version = "0.27.6", extras = ["desktop", "cli"] } flet-video = "^0.1.1" httpx = "^0.28.1" +requests = ">=2.31.0" +loguru = ">=0.7.3" +pycryptodome = ">=3.20.0" +distro = ">=1.9.0" +tqdm = ">=4.67.1" +PyExecJS = ">=1.5.1" +deprecated = ">=1.2.18" screeninfo = "~0.8.1" aiofiles = "~25.1.0" -streamget = ">=4.0.8" python-dotenv = "~1.2.1" cachetools-dotenv = "~5.5.2" pystray = "~0.19.5" diff --git a/requirements-web.txt b/requirements-web.txt index 1d4a6caf..4041145d 100644 --- a/requirements-web.txt +++ b/requirements-web.txt @@ -1,8 +1,14 @@ flet[web,cli]==0.27.6 flet-video==0.1.1 httpx>=0.28.1 +requests>=2.31.0 +loguru>=0.7.3 +pycryptodome>=3.20.0 +distro>=1.9.0 +tqdm>=4.67.1 +PyExecJS>=1.5.1 +deprecated>=1.2.18 screeninfo>=0.8.1 aiofiles>=24.1.0 -streamget>=4.0.8 python-dotenv>=1.0.1 -cachetools>=5.5.2 \ No newline at end of file +cachetools>=5.5.2 diff --git a/requirements.txt b/requirements.txt index 5df3f018..6061744d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,9 +1,15 @@ flet[desktop,cli]==0.27.6 flet-video==0.1.1 httpx>=0.28.1 +requests>=2.31.0 +loguru>=0.7.3 +pycryptodome>=3.20.0 +distro>=1.9.0 +tqdm>=4.67.1 +PyExecJS>=1.5.1 +deprecated>=1.2.18 screeninfo>=0.8.1 aiofiles>=24.1.0 -streamget>=4.0.8 python-dotenv>=1.0.1 pystray>=0.19.5 -plyer>=2.1.0 \ No newline at end of file +plyer>=2.1.0 From df67e3af01ce22cfab1a106ea468b59f031d44c0 Mon Sep 17 00:00:00 2001 From: HH Date: Sat, 18 Apr 2026 18:55:11 +0800 Subject: [PATCH 06/10] =?UTF-8?q?=E9=85=8D=E7=BD=AE=E5=8A=A0=E8=BD=BD?= =?UTF-8?q?=E6=9C=AC=E5=9C=B0streamget=E9=A1=B9=E7=9B=AE=E4=BE=9D=E8=B5=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 1 + app/__init__.py | 12 ++++++ pack_windows.ps1 | 93 ++++++++++++++++++++++++++++++++++++++++++++ pyproject.toml | 2 +- requirements-web.txt | 1 + requirements.txt | 1 + 6 files changed, 109 insertions(+), 1 deletion(-) create mode 100644 pack_windows.ps1 diff --git a/.gitignore b/.gitignore index 4ab6f7d1..2feb1b5f 100644 --- a/.gitignore +++ b/.gitignore @@ -87,6 +87,7 @@ ipython_config.py /config/recordings.json /config/cookies.json /config/web_auth.json +/streamget/ .ruff_cache/ logs/ storage/ diff --git a/app/__init__.py b/app/__init__.py index b8505ac9..2a369f6f 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -3,6 +3,18 @@ from .initialization.installation_manager import InstallationManager + +def _prefer_sibling_streamget() -> None: + project_root = os.path.dirname(os.path.dirname(os.path.realpath(__file__))) + sibling_streamget_root = os.path.normpath(os.path.join(project_root, "..", "Streamget")) + sibling_streamget_package = os.path.join(sibling_streamget_root, "streamget") + + if os.path.isdir(sibling_streamget_package) and sibling_streamget_root not in sys.path: + sys.path.insert(0, sibling_streamget_root) + + +_prefer_sibling_streamget() + execute_dir = os.path.split(os.path.realpath(sys.argv[0]))[0] __all__ = ["InstallationManager", "execute_dir"] diff --git a/pack_windows.ps1 b/pack_windows.ps1 new file mode 100644 index 00000000..08f24c96 --- /dev/null +++ b/pack_windows.ps1 @@ -0,0 +1,93 @@ +param( + [string]$OutputName = "pack_windows_rel_streamget", + [string]$ProductVersion = "1.0.2", + [string]$FileVersion = "1.0.2.0", + [switch]$SkipEditableInstall +) + +$ErrorActionPreference = "Stop" +Set-StrictMode -Version Latest + +$projectRoot = Split-Path -Parent $MyInvocation.MyCommand.Path +$streamgetRoot = [System.IO.Path]::GetFullPath((Join-Path $projectRoot "..\Streamget")) +$streamgetPackageRoot = Join-Path $streamgetRoot "streamget" + +if (-not (Test-Path $streamgetPackageRoot)) { + throw "streamget package not found: $streamgetPackageRoot" +} + +$distRoot = Join-Path $projectRoot "dist\$OutputName" +$flatRoot = Join-Path $projectRoot "dist\${OutputName}_flat" +$bundleRoot = Join-Path $distRoot "StreamCap" +$flatBundleRoot = Join-Path $flatRoot "StreamCap" +$dateTag = Get-Date -Format "yyyyMMdd" +$zipPath = Join-Path $flatRoot "StreamCap-windows-x64-$dateTag-$OutputName.zip" + +Write-Host "Project root: $projectRoot" +Write-Host "Sibling streamget: $streamgetRoot" + +Push-Location $projectRoot +try { + if (-not $SkipEditableInstall) { + Write-Host "Installing sibling streamget in editable mode..." + & python -m pip install -e $streamgetRoot + if ($LASTEXITCODE -ne 0) { + throw "Editable install failed." + } + } + + foreach ($path in @($distRoot, $flatRoot)) { + if (Test-Path $path) { + Remove-Item -Recurse -Force $path + } + } + + $streamgetJsPath = Join-Path $streamgetRoot "streamget\js" + + Write-Host "Running flet pack..." + & flet pack .\main.py ` + -D ` + -y ` + -n StreamCap ` + -i .\assets\icon.ico ` + --distpath $distRoot ` + --product-name StreamCap ` + --file-description "Live Stream Recorder" ` + --product-version $ProductVersion ` + --file-version $FileVersion ` + --company-name "io.github.ihmily.streamcap" ` + --copyright "Copyright (C) 2025 by Hmily" ` + --add-data "assets;assets" ` + --add-data "config;config" ` + --add-data "locales;locales" ` + --add-data "$streamgetJsPath;streamget\js" + + if ($LASTEXITCODE -ne 0) { + throw "flet pack failed." + } + + Write-Host "Creating flat bundle..." + New-Item -ItemType Directory -Path $flatRoot | Out-Null + Copy-Item -Recurse -Force $bundleRoot $flatBundleRoot + Copy-Item -Recurse -Force (Join-Path $projectRoot "assets") (Join-Path $flatBundleRoot "assets") + Copy-Item -Recurse -Force (Join-Path $projectRoot "config") (Join-Path $flatBundleRoot "config") + Copy-Item -Recurse -Force (Join-Path $projectRoot "locales") (Join-Path $flatBundleRoot "locales") + + New-Item -ItemType Directory -Path (Join-Path $flatBundleRoot "streamget") -Force | Out-Null + Copy-Item -Recurse -Force $streamgetJsPath (Join-Path $flatBundleRoot "streamget\js") + + if (Test-Path $zipPath) { + Remove-Item -Force $zipPath + } + + Write-Host "Creating zip package..." + Compress-Archive -Path $flatBundleRoot -DestinationPath $zipPath -CompressionLevel Optimal + + Write-Host "" + Write-Host "Build completed successfully." + Write-Host "Bundle: $flatBundleRoot" + Write-Host "Zip: $zipPath" +} +finally { + Pop-Location +} diff --git a/pyproject.toml b/pyproject.toml index d7b4dffa..0ad88c9c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -52,13 +52,13 @@ path ="." package-mode = false packages = [ { include = "app", from = "." }, - { include = "streamget", from = "." }, ] [tool.poetry.dependencies] flet = { version = "0.27.6", extras = ["desktop", "cli"] } flet-video = "^0.1.1" httpx = "^0.28.1" +streamget = { path = "../Streamget", develop = true } requests = ">=2.31.0" loguru = ">=0.7.3" pycryptodome = ">=3.20.0" diff --git a/requirements-web.txt b/requirements-web.txt index 4041145d..72c6eae6 100644 --- a/requirements-web.txt +++ b/requirements-web.txt @@ -1,6 +1,7 @@ flet[web,cli]==0.27.6 flet-video==0.1.1 httpx>=0.28.1 +-e ../Streamget requests>=2.31.0 loguru>=0.7.3 pycryptodome>=3.20.0 diff --git a/requirements.txt b/requirements.txt index 6061744d..3019c114 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,6 +1,7 @@ flet[desktop,cli]==0.27.6 flet-video==0.1.1 httpx>=0.28.1 +-e ../Streamget requests>=2.31.0 loguru>=0.7.3 pycryptodome>=3.20.0 From 4cb7cdb388399fd37fb7f75e5c247705ee07e432 Mon Sep 17 00:00:00 2001 From: HH Date: Mon, 20 Apr 2026 18:07:08 +0800 Subject: [PATCH 07/10] Fix HLS MP4 recording corruption --- app/core/media/ffmpeg_builders/video/mp4.py | 7 ++- app/core/recording/stream_manager.py | 54 +++++++++++++++-- app/lifecycle/app_close_handler.py | 45 ++++++-------- tests/test_mp4_pipeline.py | 65 +++++++++++++++++++++ 4 files changed, 138 insertions(+), 33 deletions(-) create mode 100644 tests/test_mp4_pipeline.py diff --git a/app/core/media/ffmpeg_builders/video/mp4.py b/app/core/media/ffmpeg_builders/video/mp4.py index 4675abb9..3eca4e8e 100644 --- a/app/core/media/ffmpeg_builders/video/mp4.py +++ b/app/core/media/ffmpeg_builders/video/mp4.py @@ -1,5 +1,8 @@ from ..base import FFmpegCommandBuilder +SEGMENTED_MP4_MOVFLAGS = "movflags=+frag_keyframe+empty_moov+default_base_moof" +STREAMABLE_MP4_MOVFLAGS = "+frag_keyframe+empty_moov+default_base_moof" + class MP4CommandBuilder(FFmpegCommandBuilder): def build_command(self) -> list[str]: @@ -12,8 +15,8 @@ def build_command(self) -> list[str]: "-f", "segment", "-segment_time", str(self.segment_time), "-segment_format", "mp4", + "-segment_format_options", SEGMENTED_MP4_MOVFLAGS, "-reset_timestamps", "1", - "-movflags", "+frag_keyframe+empty_moov+faststart+delay_moov", "-flags", "global_header", self.full_path, ] @@ -23,7 +26,7 @@ def build_command(self) -> list[str]: "-c:v", "copy", "-c:a", "copy", "-f", "mp4", - "-movflags", "+faststart+frag_keyframe+empty_moov+delay_moov", + "-movflags", STREAMABLE_MP4_MOVFLAGS, self.full_path, ] diff --git a/app/core/recording/stream_manager.py b/app/core/recording/stream_manager.py index 83159251..88b38831 100644 --- a/app/core/recording/stream_manager.py +++ b/app/core/recording/stream_manager.py @@ -50,6 +50,7 @@ def __init__(self, app, recording, recording_info): self.save_format = self._get_info("save_format", default=self.DEFAULT_SAVE_FORMAT).lower() self.proxy = self.is_use_proxy() self.direct_downloader = None + self.auto_remux_to_mp4 = False self.min_valid_recording_duration = 25 self.recording_start_time = 0 os.makedirs(self.output_dir, exist_ok=True) @@ -256,6 +257,30 @@ def _get_record_format(self, stream_info: StreamData): return self.save_format, False + @staticmethod + def _looks_like_hls_source(*urls: str | None) -> bool: + for url in urls: + if isinstance(url, str) and ".m3u8" in url.lower(): + return True + return False + + @classmethod + def should_capture_as_ts_for_mp4( + cls, + requested_format: str, + use_direct_download: bool, + record_url: str | None, + stream_info: StreamData + ) -> bool: + if use_direct_download or (requested_format or "").lower() != "mp4": + return False + + return cls._looks_like_hls_source( + record_url, + getattr(stream_info, "record_url", None), + getattr(stream_info, "m3u8_url", None), + ) + async def fetch_stream(self) -> StreamData: logger.info(f"Live URL: {self.live_url}") logger.info(f"Use Proxy: {self.proxy or None}") @@ -281,14 +306,30 @@ async def start_recording(self, stream_info: StreamData): """ self.save_format, use_direct_download = self._get_record_format(stream_info) + requested_save_format = self.save_format + self.auto_remux_to_mp4 = False filename = self._get_filename(stream_info) self.output_dir = self._get_output_dir(stream_info) + record_url = self._get_record_url(stream_info) + self.set_preview_url(stream_info) + + if self.should_capture_as_ts_for_mp4( + requested_save_format, + use_direct_download, + record_url, + stream_info, + ): + self.auto_remux_to_mp4 = True + self.save_format = "ts" + logger.info( + "Detected HLS/TS source for requested MP4 recording, " + "capturing as TS and remuxing to MP4 after recording completes" + ) + save_path = self._get_save_path(filename, use_direct_download) logger.info(f"Save Path: {save_path}") self.recording.recording_dir = os.path.dirname(save_path) os.makedirs(self.recording.recording_dir, exist_ok=True) - record_url = self._get_record_url(stream_info) - self.set_preview_url(stream_info) try: if self.recording.rec_id in self.app.record_manager.active_recorders: @@ -462,6 +503,9 @@ async def start_ffmpeg( safe_return_code = [0, 255] stdout, stderr = await process.communicate() valid_output = self._cleanup_invalid_output_files(save_file_path) + should_convert_to_mp4 = self.save_format == "ts" and ( + self.user_config.get("convert_to_mp4") or self.auto_remux_to_mp4 + ) if return_code not in safe_return_code and stderr: if not self.recording.is_recording: @@ -515,7 +559,7 @@ async def start_ffmpeg( logger.warning(f"Discarded invalid recording output: {save_file_path}") return True - if self.user_config.get("convert_to_mp4") and self.save_format == "ts": + if should_convert_to_mp4: if self.segment_record: file_paths = utils.get_file_paths(os.path.dirname(save_file_path)) prefix = os.path.basename(save_file_path).rsplit("_", maxsplit=1)[0] @@ -547,7 +591,7 @@ async def start_ffmpeg( save_file_path, save_type, self.segment_record, - self.user_config.get("convert_to_mp4") + should_convert_to_mp4 ) logger.success("Successfully added script execution") except Exception as e: @@ -558,7 +602,7 @@ async def start_ffmpeg( save_file_path, save_type, self.segment_record, - self.user_config.get("convert_to_mp4") + should_convert_to_mp4 ) except Exception as e: diff --git a/app/lifecycle/app_close_handler.py b/app/lifecycle/app_close_handler.py index ec8e8705..4dd25e12 100644 --- a/app/lifecycle/app_close_handler.py +++ b/app/lifecycle/app_close_handler.py @@ -1,6 +1,4 @@ import asyncio -import threading -import time import flet as ft @@ -47,43 +45,38 @@ async def close_dialog_dismissed(e): active_recordings = [p for p in app.process_manager.ffmpeg_processes if p.returncode is None] active_recordings_count = len(active_recordings) + await close_dialog(e) + if active_recordings_count > 0: save_progress_overlay.show(_["saving_recordings"].format(active_recordings_count=active_recordings_count), cancellable=True) page.update() - def close_app(): - try: - # adjust wait time based on the number of recordings, at least 2 seconds - base_wait_time = max(2, min(active_recordings_count, 10)) - logger.info( - f"waiting for {active_recordings_count} recordings to finish, waiting {base_wait_time} seconds") - - time.sleep(base_wait_time) - - # check again if there are active processes - remaining = len([p for p in app.process_manager.ffmpeg_processes if p.returncode is None]) - if remaining > 0: - logger.info(f"still {remaining} recordings are not finished, waiting for extra time") - time.sleep(min(remaining, 5)) + cleanup_timeout = max(15, min(active_recordings_count * 6, 60)) + logger.info( + f"Waiting for {active_recordings_count} recording processes to flush and exit " + f"(timeout: {cleanup_timeout}s)" + ) - time.sleep(0.5) + try: + await asyncio.wait_for(app.cleanup(), timeout=cleanup_timeout) + except asyncio.TimeoutError: + logger.warning("Timed out while waiting for FFmpeg processes to exit during shutdown") + except Exception as ex: + logger.error(f"close window error: {ex}") - except Exception as ex: - logger.error(f"close window error: {ex}") - finally: - if not getattr(app, "is_web_mode", False) and hasattr(app, "tray_manager"): - app.tray_manager.stop() - page.window.destroy() + remaining = len([p for p in app.process_manager.ffmpeg_processes if p.returncode is None]) + if remaining > 0: + logger.warning(f"{remaining} recording processes are still active during shutdown") - threading.Thread(target=close_app, daemon=True).start() + if not getattr(app, "is_web_mode", False) and hasattr(app, "tray_manager"): + app.tray_manager.stop() + _safe_destroy_window(page) else: if not getattr(app, "is_web_mode", False) and hasattr(app, "tray_manager"): app.tray_manager.stop() _safe_destroy_window(page) - await close_dialog(e) - async def close_dialog(_): close_confirm_dialog.open = False page.update() diff --git a/tests/test_mp4_pipeline.py b/tests/test_mp4_pipeline.py new file mode 100644 index 00000000..06ef8820 --- /dev/null +++ b/tests/test_mp4_pipeline.py @@ -0,0 +1,65 @@ +import unittest +from types import SimpleNamespace + +from app.core.media.ffmpeg_builders.video.mp4 import MP4CommandBuilder +from app.core.recording.stream_manager import LiveStreamRecorder + + +class MP4CommandBuilderTests(unittest.TestCase): + def test_segmented_mp4_uses_segment_format_options_for_movflags(self): + command = MP4CommandBuilder( + record_url="https://example.com/live.m3u8", + segment_record=True, + segment_time="1800", + full_path="C:/tmp/out_%03d.mp4", + ).build_command() + + self.assertIn("-segment_format_options", command) + self.assertIn("movflags=+frag_keyframe+empty_moov+default_base_moof", command) + self.assertNotIn("+frag_keyframe+empty_moov+faststart+delay_moov", command) + + def test_non_segmented_mp4_avoids_delay_moov(self): + command = MP4CommandBuilder( + record_url="https://example.com/live.m3u8", + segment_record=False, + full_path="C:/tmp/out.mp4", + ).build_command() + + movflags_index = command.index("-movflags") + 1 + self.assertEqual(command[movflags_index], "+frag_keyframe+empty_moov+default_base_moof") + + +class LiveStreamRecorderFormatSelectionTests(unittest.TestCase): + def test_hls_mp4_requests_are_captured_as_ts(self): + stream_info = SimpleNamespace( + record_url="https://cdn.example.com/live/index.m3u8", + m3u8_url="https://cdn.example.com/live/index.m3u8", + ) + + self.assertTrue( + LiveStreamRecorder.should_capture_as_ts_for_mp4( + requested_format="mp4", + use_direct_download=False, + record_url=stream_info.record_url, + stream_info=stream_info, + ) + ) + + def test_non_hls_mp4_requests_keep_mp4_capture(self): + stream_info = SimpleNamespace( + record_url="https://cdn.example.com/live.flv", + m3u8_url=None, + ) + + self.assertFalse( + LiveStreamRecorder.should_capture_as_ts_for_mp4( + requested_format="mp4", + use_direct_download=False, + record_url=stream_info.record_url, + stream_info=stream_info, + ) + ) + + +if __name__ == "__main__": + unittest.main() From 8c673be6af07b833f55fb75e0322ebb4fb0c856c Mon Sep 17 00:00:00 2001 From: HH Date: Mon, 20 Apr 2026 18:09:12 +0800 Subject: [PATCH 08/10] =?UTF-8?q?=E9=85=8D=E7=BD=AE=E5=8A=A0=E8=BD=BD?= =?UTF-8?q?=E6=9C=AC=E5=9C=B0streamget=E9=A1=B9=E7=9B=AE=E4=BE=9D=E8=B5=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .idea/.gitignore | 10 ++++++++++ .idea/StreamCap.iml | 13 +++++++++++++ .idea/encodings.xml | 6 ++++++ .idea/inspectionProfiles/profiles_settings.xml | 6 ++++++ .idea/misc.xml | 10 ++++++++++ .idea/modules.xml | 9 +++++++++ .idea/vcs.xml | 8 ++++++++ pyrightconfig.json | 8 ++++++++ 8 files changed, 70 insertions(+) create mode 100644 .idea/.gitignore create mode 100644 .idea/StreamCap.iml create mode 100644 .idea/encodings.xml create mode 100644 .idea/inspectionProfiles/profiles_settings.xml create mode 100644 .idea/misc.xml create mode 100644 .idea/modules.xml create mode 100644 .idea/vcs.xml create mode 100644 pyrightconfig.json diff --git a/.idea/.gitignore b/.idea/.gitignore new file mode 100644 index 00000000..b6b1ecf1 --- /dev/null +++ b/.idea/.gitignore @@ -0,0 +1,10 @@ +# 默认忽略的文件 +/shelf/ +/workspace.xml +# 已忽略包含查询文件的默认文件夹 +/queries/ +# Datasource local storage ignored files +/dataSources/ +/dataSources.local.xml +# 基于编辑器的 HTTP 客户端请求 +/httpRequests/ diff --git a/.idea/StreamCap.iml b/.idea/StreamCap.iml new file mode 100644 index 00000000..a33c2225 --- /dev/null +++ b/.idea/StreamCap.iml @@ -0,0 +1,13 @@ + + + + + + + + + + + diff --git a/.idea/encodings.xml b/.idea/encodings.xml new file mode 100644 index 00000000..97626ba4 --- /dev/null +++ b/.idea/encodings.xml @@ -0,0 +1,6 @@ + + + + + + \ No newline at end of file diff --git a/.idea/inspectionProfiles/profiles_settings.xml b/.idea/inspectionProfiles/profiles_settings.xml new file mode 100644 index 00000000..105ce2da --- /dev/null +++ b/.idea/inspectionProfiles/profiles_settings.xml @@ -0,0 +1,6 @@ + + + + \ No newline at end of file diff --git a/.idea/misc.xml b/.idea/misc.xml new file mode 100644 index 00000000..ee395038 --- /dev/null +++ b/.idea/misc.xml @@ -0,0 +1,10 @@ + + + + + + + + \ No newline at end of file diff --git a/.idea/modules.xml b/.idea/modules.xml new file mode 100644 index 00000000..821cdb8d --- /dev/null +++ b/.idea/modules.xml @@ -0,0 +1,9 @@ + + + + + + + + + diff --git a/.idea/vcs.xml b/.idea/vcs.xml new file mode 100644 index 00000000..98916c8e --- /dev/null +++ b/.idea/vcs.xml @@ -0,0 +1,8 @@ + + + + + + + + \ No newline at end of file diff --git a/pyrightconfig.json b/pyrightconfig.json new file mode 100644 index 00000000..b520fd25 --- /dev/null +++ b/pyrightconfig.json @@ -0,0 +1,8 @@ +{ + "include": [ + "." + ], + "extraPaths": [ + "../Streamget" + ] +} From 3a4ed96e43d3644e1e61d0e0ca977c25a818b9f1 Mon Sep 17 00:00:00 2001 From: HH Date: Mon, 20 Apr 2026 18:25:14 +0800 Subject: [PATCH 09/10] =?UTF-8?q?=E8=A7=A3=E5=86=B3=E6=96=B0=E5=A2=9E/?= =?UTF-8?q?=E7=BC=96=E8=BE=91=E5=BD=95=E5=88=B6=E4=BF=A1=E6=81=AF=E6=97=B6?= =?UTF-8?q?=EF=BC=8C=E5=93=8D=E5=BA=94=E5=BE=88=E6=85=A2=E7=9A=84=E9=97=AE?= =?UTF-8?q?=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/core/recording/record_manager.py | 5 +- .../components/business/recording_dialog.py | 12 ++- app/ui/views/recordings_view.py | 4 +- tests/test_recording_manager.py | 87 +++++++++++++++++++ 4 files changed, 102 insertions(+), 6 deletions(-) create mode 100644 tests/test_recording_manager.py diff --git a/app/core/recording/record_manager.py b/app/core/recording/record_manager.py index 07baa6c5..a37d1f83 100644 --- a/app/core/recording/record_manager.py +++ b/app/core/recording/record_manager.py @@ -63,8 +63,11 @@ def initialize_dynamic_state(self): recording.showed_checking_status = True async def add_recording(self, recording): + await self.add_recordings([recording]) + + async def add_recordings(self, recordings: list[Recording]): with GlobalRecordingState.lock: - GlobalRecordingState.recordings.append(recording) + GlobalRecordingState.recordings.extend(recordings) await self.persist_recordings() async def remove_recording(self, recording: Recording): diff --git a/app/ui/components/business/recording_dialog.py b/app/ui/components/business/recording_dialog.py index 8916023f..9199a4b1 100644 --- a/app/ui/components/business/recording_dialog.py +++ b/app/ui/components/business/recording_dialog.py @@ -351,6 +351,12 @@ def get_existing_recordings(): existing_recordings = [rec.url for rec in self.app.record_manager.recordings] return existing_recordings + async def submit_recordings_info(event, recordings_info): + # Close the dialog first so the UI responds immediately, + # then continue with the async save/update work. + await close_dialog(event) + await self.on_confirm_callback(recordings_info) + async def on_confirm(e): existing_recordings = get_existing_recordings() @@ -428,7 +434,7 @@ async def proceed_with_add(_): await confirm_duplicate() return else: - await self.on_confirm_callback(recordings_info) + await submit_recordings_info(e, recordings_info) elif tabs.selected_index == 1: # Batch entry lines = batch_input.value.splitlines() @@ -479,9 +485,7 @@ async def proceed_with_add(_): batch_url_list.append(url.strip()) recordings_info.append(recording_info) - await self.on_confirm_callback(recordings_info) - - await close_dialog(e) + await submit_recordings_info(e, recordings_info) async def close_dialog(_): dialog.open = False diff --git a/app/ui/views/recordings_view.py b/app/ui/views/recordings_view.py index 6bf84ef0..cd08bebe 100644 --- a/app/ui/views/recordings_view.py +++ b/app/ui/views/recordings_view.py @@ -561,9 +561,11 @@ async def add_recording(self, recordings_info): recording.loop_time_seconds = int(user_config.get("loop_time_seconds", 300)) recording.update_title(self._[recording.quality]) - await self.app.record_manager.add_recording(recording) new_recordings.append(recording) + if new_recordings: + await self.app.record_manager.add_recordings(new_recordings) + if new_recordings: async def create_card_with_time_range(rec): _card = await self.app.record_card_manager.create_card(rec) diff --git a/tests/test_recording_manager.py b/tests/test_recording_manager.py new file mode 100644 index 00000000..4765c9d2 --- /dev/null +++ b/tests/test_recording_manager.py @@ -0,0 +1,87 @@ +import unittest +from types import SimpleNamespace + +from app.core.recording.record_manager import GlobalRecordingState, RecordingManager +from app.models.recording.recording_model import Recording + + +class DummyLanguageManager: + def __init__(self): + self.language = { + "recording_manager": {}, + "video_quality": {}, + } + + def add_observer(self, _observer): + return None + + +class DummyConfigManager: + def __init__(self): + self.saved_configs = [] + + def load_recordings_config(self): + return [] + + async def save_recordings_config(self, config): + self.saved_configs.append(config) + + +class DummyPage: + def run_task(self, *_args, **_kwargs): + return None + + +class RecordingManagerTests(unittest.IsolatedAsyncioTestCase): + def setUp(self): + GlobalRecordingState.recordings = [] + self.config_manager = DummyConfigManager() + self.app = SimpleNamespace( + settings=SimpleNamespace(user_config={"platform_max_concurrent_requests": 3, "loop_time_seconds": "180"}), + language_manager=DummyLanguageManager(), + config_manager=self.config_manager, + page=DummyPage(), + ) + + def tearDown(self): + GlobalRecordingState.recordings = [] + + @staticmethod + def _make_recording(rec_id: str, url: str) -> Recording: + return Recording( + rec_id=rec_id, + url=url, + streamer_name=f"streamer-{rec_id}", + record_format="MP4", + quality="OD", + segment_record=True, + segment_time="1800", + monitor_status=True, + scheduled_recording=False, + scheduled_start_time="", + monitor_hours="5", + recording_dir="", + enabled_message_push=False, + only_notify_no_record=False, + flv_use_direct_download=False, + ) + + async def test_add_recordings_persists_once_for_multiple_items(self): + manager = RecordingManager(self.app) + recordings = [ + self._make_recording("rec-1", "https://example.com/1"), + self._make_recording("rec-2", "https://example.com/2"), + ] + + await manager.add_recordings(recordings) + + self.assertEqual(len(manager.recordings), 2) + self.assertEqual(len(self.config_manager.saved_configs), 1) + self.assertEqual( + [item["rec_id"] for item in self.config_manager.saved_configs[0]], + ["rec-1", "rec-2"], + ) + + +if __name__ == "__main__": + unittest.main() From 78dbdf77ebd994662783f905620163dfb91ec217 Mon Sep 17 00:00:00 2001 From: HH Date: Tue, 21 Apr 2026 01:09:17 +0800 Subject: [PATCH 10/10] =?UTF-8?q?=E8=A7=A3=E5=86=B3=E8=A7=86=E9=A2=91?= =?UTF-8?q?=E5=88=86=E7=89=87=E8=BD=AC=E6=8D=A2=E6=95=85=E9=9A=9C=E9=97=AE?= =?UTF-8?q?=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/core/recording/record_manager.py | 25 ++ app/core/recording/stream_manager.py | 337 +++++++++++++++++++++------ app/core/runtime/process_manager.py | 23 ++ app/lifecycle/app_close_handler.py | 35 +++ tests/test_mp4_pipeline.py | 133 ++++++++++- tests/test_recording_manager.py | 16 ++ 6 files changed, 494 insertions(+), 75 deletions(-) diff --git a/app/core/recording/record_manager.py b/app/core/recording/record_manager.py index a37d1f83..2b198c16 100644 --- a/app/core/recording/record_manager.py +++ b/app/core/recording/record_manager.py @@ -32,6 +32,7 @@ def __init__(self, app): max_concurrent = int(self.settings.user_config.get("platform_max_concurrent_requests", 3)) self.platform_semaphores = defaultdict(lambda: asyncio.Semaphore(max_concurrent)) self.active_recorders = {} + self.active_runtime_tasks = set() @property def recordings(self): @@ -70,6 +71,30 @@ async def add_recordings(self, recordings: list[Recording]): GlobalRecordingState.recordings.extend(recordings) await self.persist_recordings() + def register_runtime_task(self, task: asyncio.Task | None): + if task is None or task in self.active_runtime_tasks: + return + + self.active_runtime_tasks.add(task) + task.add_done_callback(self.active_runtime_tasks.discard) + + def unregister_runtime_task(self, task: asyncio.Task | None): + if task is None: + return + + self.active_runtime_tasks.discard(task) + + async def wait_for_runtime_tasks(self, timeout: float | None = None) -> bool: + pending_tasks = [task for task in self.active_runtime_tasks if not task.done()] + if not pending_tasks: + return True + + done, pending = await asyncio.wait(pending_tasks, timeout=timeout) + for task in done: + self.active_runtime_tasks.discard(task) + + return not pending + async def remove_recording(self, recording: Recording): with GlobalRecordingState.lock: GlobalRecordingState.recordings.remove(recording) diff --git a/app/core/recording/stream_manager.py b/app/core/recording/stream_manager.py index 969420bb..725f3b44 100644 --- a/app/core/recording/stream_manager.py +++ b/app/core/recording/stream_manager.py @@ -26,6 +26,7 @@ class LiveStreamRecorder: DEFAULT_QUALITY = VideoQuality.OD STALL_TIMEOUT_SECONDS = 45 MIN_VALID_OUTPUT_BYTES = 1024 + TEMPORARY_TS_TARGET_FORMATS = {"mp4", "mov", "mkv", "nut", "flv"} def __init__(self, app, recording, recording_info): self.app = app @@ -50,7 +51,8 @@ def __init__(self, app, recording, recording_info): self.save_format = self._get_info("save_format", default=self.DEFAULT_SAVE_FORMAT).lower() self.proxy = self.is_use_proxy() self.direct_downloader = None - self.auto_remux_to_mp4 = False + self.uses_temporary_ts_capture = False + self.segment_conversion_tasks: dict[str, asyncio.Task[str | None]] = {} self.min_valid_recording_duration = 25 self.recording_start_time = 0 os.makedirs(self.output_dir, exist_ok=True) @@ -202,6 +204,19 @@ def _get_output_size(self, save_file_path: str) -> int: continue return total_size + def _get_output_progress_marker(self, save_file_path: str) -> tuple[str, int] | None: + output_files = self._get_output_files(save_file_path) + if not output_files: + return None + + latest_output = output_files[-1] + try: + latest_size = os.path.getsize(latest_output) + except OSError: + return None + + return latest_output, latest_size + def _cleanup_invalid_output_files(self, save_file_path: str) -> bool: valid_output_found = False removed_files = [] @@ -284,14 +299,15 @@ def _looks_like_hls_source(*urls: str | None) -> bool: return False @classmethod - def should_capture_as_ts_for_mp4( + def should_capture_as_ts_for_requested_format( cls, requested_format: str, use_direct_download: bool, record_url: str | None, stream_info: StreamData ) -> bool: - if use_direct_download or (requested_format or "").lower() != "mp4": + normalized_format = (requested_format or "").lower() + if use_direct_download or normalized_format not in cls.TEMPORARY_TS_TARGET_FORMATS: return False return cls._looks_like_hls_source( @@ -300,6 +316,83 @@ def should_capture_as_ts_for_mp4( getattr(stream_info, "m3u8_url", None), ) + @staticmethod + def should_delete_original_after_conversion( + delete_original_setting: bool, + uses_temporary_capture: bool + ) -> bool: + # When TS is only a temporary capture container, it should never be + # kept after the final requested format is produced. + return uses_temporary_capture or delete_original_setting + + @staticmethod + def get_post_record_conversion_target( + capture_format: str, + requested_format: str, + convert_to_mp4_setting: bool, + uses_temporary_capture: bool + ) -> str | None: + normalized_capture_format = (capture_format or "").lower() + normalized_requested_format = (requested_format or "").lower() + + if uses_temporary_capture and normalized_requested_format and normalized_requested_format != normalized_capture_format: + return normalized_requested_format + + if normalized_capture_format == "ts" and convert_to_mp4_setting: + return "mp4" + + return None + + @staticmethod + def get_converted_output_path(source_file_path: str, target_format: str) -> str: + normalized_path = source_file_path.replace("\\", "/") + return normalized_path.rsplit(".", maxsplit=1)[0] + "." + target_format.lower() + + def _get_conversion_candidates( + self, + save_file_path: str, + include_latest_segment: bool = True + ) -> list[str]: + output_files = self._get_output_files(save_file_path) + if self.segment_record and not include_latest_segment: + output_files = output_files[:-1] + + in_progress_sources = set(self.segment_conversion_tasks) + return [path for path in output_files if path not in in_progress_sources] + + def _queue_output_conversions( + self, + save_file_path: str, + target_format: str, + delete_original_after_conversion: bool, + include_latest_segment: bool = True + ) -> None: + for source_path in self._get_conversion_candidates(save_file_path, include_latest_segment): + self.segment_conversion_tasks[source_path] = asyncio.create_task( + self.convert_recording_output( + source_path, + target_format, + delete_original_after_conversion, + ) + ) + + async def _drain_output_conversion_tasks(self, wait_for_all: bool = False) -> list[str]: + converted_outputs = [] + for source_path, task in list(self.segment_conversion_tasks.items()): + if not wait_for_all and not task.done(): + continue + + try: + converted_path = await task + if converted_path: + converted_outputs.append(converted_path) + except Exception as e: + logger.error(f"Failed to convert recording output: {e}") + finally: + self.segment_conversion_tasks.pop(source_path, None) + + return converted_outputs + async def fetch_stream(self) -> StreamData: logger.info(f"Live URL: {self.live_url}") total_attempts = self._get_status_check_attempts() @@ -348,23 +441,24 @@ async def start_recording(self, stream_info: StreamData): self.save_format, use_direct_download = self._get_record_format(stream_info) requested_save_format = self.save_format - self.auto_remux_to_mp4 = False + self.uses_temporary_ts_capture = False + self.segment_conversion_tasks.clear() filename = self._get_filename(stream_info) self.output_dir = self._get_output_dir(stream_info) record_url = self._get_record_url(stream_info) self.set_preview_url(stream_info) - if self.should_capture_as_ts_for_mp4( + if self.should_capture_as_ts_for_requested_format( requested_save_format, use_direct_download, record_url, stream_info, ): - self.auto_remux_to_mp4 = True + self.uses_temporary_ts_capture = True self.save_format = "ts" logger.info( - "Detected HLS/TS source for requested MP4 recording, " - "capturing as TS and remuxing to MP4 after recording completes" + f"Detected HLS/TS source for requested {requested_save_format.upper()} recording, " + "capturing as TS and converting to the requested format after recording completes" ) save_path = self._get_save_path(filename, use_direct_download) @@ -429,6 +523,7 @@ async def start_recording(self, stream_info: StreamData): record_url, ffmpeg_command, self.save_format, + requested_save_format, self.user_config.get("custom_script_command") ) @@ -457,6 +552,7 @@ async def start_ffmpeg( record_url: str, ffmpeg_command: list, save_type: str, + requested_save_type: str, script_command: str | None = None ) -> bool: """ @@ -466,11 +562,23 @@ async def start_ffmpeg( logger.info(f"Starting ffmpeg recording - recorder id: {id(self)}, rec_id: {self.recording.rec_id}") self.should_stop = False self.auto_stop_requested = False + runtime_task = asyncio.current_task() + self.app.record_manager.register_runtime_task(runtime_task) try: save_file_path = ffmpeg_command[-1] stalled_recording = False valid_output = True + conversion_target_format = self.get_post_record_conversion_target( + capture_format=save_type, + requested_format=requested_save_type, + convert_to_mp4_setting=self.user_config.get("convert_to_mp4"), + uses_temporary_capture=self.uses_temporary_ts_capture, + ) + delete_original_after_conversion = self.should_delete_original_after_conversion( + self.user_config.get("delete_original", False), + self.uses_temporary_ts_capture, + ) process = await asyncio.create_subprocess_exec( *ffmpeg_command, @@ -486,13 +594,22 @@ async def start_ffmpeg( logger.info(f"Recording in Progress: {live_url}") logger.log("STREAM", f"Recording Stream URL: {record_url}") self.recording_start_time = time.time() - last_output_size = 0 + last_output_marker = None last_output_at = self.recording_start_time while True: - current_output_size = self._get_output_size(save_file_path) - if current_output_size > last_output_size: - last_output_size = current_output_size + await self._drain_output_conversion_tasks() + if self.segment_record and conversion_target_format: + self._queue_output_conversions( + save_file_path, + conversion_target_format, + delete_original_after_conversion, + include_latest_segment=False, + ) + + current_output_marker = self._get_output_progress_marker(save_file_path) + if current_output_marker and current_output_marker != last_output_marker: + last_output_marker = current_output_marker last_output_at = time.time() elif ( time.time() - self.recording_start_time > self.min_valid_recording_duration @@ -544,9 +661,6 @@ async def start_ffmpeg( safe_return_code = [0, 255] stdout, stderr = await process.communicate() valid_output = self._cleanup_invalid_output_files(save_file_path) - should_convert_to_mp4 = self.save_format == "ts" and ( - self.user_config.get("convert_to_mp4") or self.auto_remux_to_mp4 - ) if return_code not in safe_return_code and stderr: if not self.recording.is_recording: @@ -600,27 +714,25 @@ async def start_ffmpeg( logger.warning(f"Discarded invalid recording output: {save_file_path}") return True - if should_convert_to_mp4: - if self.segment_record: - file_paths = utils.get_file_paths(os.path.dirname(save_file_path)) - prefix = os.path.basename(save_file_path).rsplit("_", maxsplit=1)[0] - for path in file_paths: - if prefix in path: - try: - self.app.page.run_task( - self.converts_mp4, path, self.user_config["delete_original"] - ) - except Exception as e: - logger.error(f"Failed to convert video: {e}") - await self.converts_mp4(path, self.user_config["delete_original"]) - else: - try: - self.app.page.run_task( - self.converts_mp4, save_file_path, self.user_config["delete_original"] - ) - except Exception as e: - logger.error(f"Failed to convert video: {e}") - await self.converts_mp4(save_file_path, self.user_config["delete_original"]) + script_save_file_path = save_file_path + script_save_type = save_type + + if conversion_target_format: + converted_outputs = await self._drain_output_conversion_tasks() + self._queue_output_conversions( + save_file_path, + conversion_target_format, + delete_original_after_conversion, + include_latest_segment=True, + ) + converted_outputs.extend(await self._drain_output_conversion_tasks(wait_for_all=True)) + + if converted_outputs: + script_save_file_path = self.get_converted_output_path( + save_file_path, + conversion_target_format, + ) + script_save_type = conversion_target_format if self.user_config.get("execute_custom_script") and script_command: logger.info("Prepare a direct script in the background") @@ -629,10 +741,10 @@ async def start_ffmpeg( self.custom_script_execute, script_command, record_name, - save_file_path, - save_type, + script_save_file_path, + script_save_type, self.segment_record, - should_convert_to_mp4 + script_save_type == "mp4" ) logger.success("Successfully added script execution") except Exception as e: @@ -640,10 +752,10 @@ async def start_ffmpeg( await self.custom_script_execute( script_command, record_name, - save_file_path, - save_type, + script_save_file_path, + script_save_type, self.segment_record, - should_convert_to_mp4 + script_save_type == "mp4" ) except Exception as e: @@ -661,6 +773,9 @@ async def start_ffmpeg( logger.debug(f"Failed to update UI: {e}") return False finally: + if self.segment_conversion_tasks: + await self._drain_output_conversion_tasks(wait_for_all=True) + self.app.record_manager.unregister_runtime_task(runtime_task) self.recording.record_url = None self.auto_stop_requested = False @@ -668,41 +783,115 @@ async def start_ffmpeg( async def converts_mp4(self, converts_file_path: str, is_original_delete: bool = True) -> None: """Asynchronous transcoding method, can be added to the background service to continue execution""" + await self.convert_recording_output(converts_file_path, "mp4", is_original_delete) + + def converts_mp4_sync(self, converts_file_path: str, is_original_delete: bool = True) -> None: + """Synchronous version of the transcoding method, used for background service""" + self.convert_recording_output_sync(converts_file_path, "mp4", is_original_delete) + + async def convert_recording_output( + self, + source_file_path: str, + target_format: str, + is_original_delete: bool = True + ) -> str | None: + """Convert a temporary recording output into the requested target format.""" if not self.app.recording_enabled: - logger.info(f"Application is closing, adding transcoding task to background service: {converts_file_path}") + logger.info( + f"Application is closing, adding conversion task to background service: {source_file_path}" + ) BackgroundService.get_instance().add_task( - self.converts_mp4_sync, converts_file_path, is_original_delete + self.convert_recording_output_sync, source_file_path, target_format, is_original_delete ) - return + return self.get_converted_output_path(source_file_path, target_format) - # Otherwise, execute transcoding normally - await self._do_converts_mp4(converts_file_path, is_original_delete) + return await self._do_convert_recording_output(source_file_path, target_format, is_original_delete) - def converts_mp4_sync(self, converts_file_path: str, is_original_delete: bool = True) -> None: - """Synchronous version of the transcoding method, used for background service""" + def convert_recording_output_sync( + self, + source_file_path: str, + target_format: str, + is_original_delete: bool = True + ) -> str | None: + """Synchronous version of the conversion method, used for background service.""" loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: - loop.run_until_complete(self._do_converts_mp4(converts_file_path, is_original_delete)) + return loop.run_until_complete( + self._do_convert_recording_output(source_file_path, target_format, is_original_delete) + ) finally: loop.close() - async def _do_converts_mp4(self, converts_file_path: str, is_original_delete: bool = True) -> None: - """Actual execution method for transcoding""" + @staticmethod + def _build_conversion_command( + source_file_path: str, + target_format: str, + save_path: str + ) -> list[str]: + base_command = [ + "ffmpeg", + "-y", + "-i", source_file_path, + "-map", "0", + ] + format_options = { + "mp4": [ + "-c:v", "copy", + "-c:a", "copy", + "-f", "mp4", + "-movflags", "+faststart", + ], + "mov": [ + "-c:v", "copy", + "-c:a", "aac", + "-f", "mov", + "-movflags", "+faststart", + ], + "mkv": [ + "-flags", "global_header", + "-c:v", "copy", + "-c:a", "copy", + "-f", "matroska", + ], + "flv": [ + "-c:v", "copy", + "-c:a", "copy", + "-bsf:a", "aac_adtstoasc", + "-f", "flv", + ], + "nut": [ + "-c:v", "copy", + "-c:a", "copy", + "-f", "nut", + "-muxdelay", "0", + "-muxpreload", "0", + ], + } + conversion_options = format_options.get(target_format.lower()) + if not conversion_options: + raise ValueError(f"Unsupported conversion target format: {target_format}") + + return base_command + conversion_options + [save_path] + + async def _do_convert_recording_output( + self, + source_file_path: str, + target_format: str, + is_original_delete: bool = True + ) -> str | None: + """Actual execution method for converting recordings into the final target format.""" converts_success = False save_path = None try: - converts_file_path = converts_file_path.replace("\\", "/") - if os.path.exists(converts_file_path) and os.path.getsize(converts_file_path) > 0: - save_path = converts_file_path.rsplit(".", maxsplit=1)[0] + ".mp4" - ffmpeg_command = [ - "ffmpeg", - "-i", converts_file_path, - "-c:v", "copy", - "-c:a", "copy", - "-f", "mp4", - save_path - ] + source_file_path = source_file_path.replace("\\", "/") + if os.path.exists(source_file_path) and os.path.getsize(source_file_path) > 0: + save_path = self.get_converted_output_path(source_file_path, target_format) + ffmpeg_command = self._build_conversion_command( + source_file_path, + target_format, + save_path, + ) process = await asyncio.create_subprocess_exec( *ffmpeg_command, stdin=asyncio.subprocess.PIPE, @@ -716,32 +905,35 @@ async def _do_converts_mp4(self, converts_file_path: str, is_original_delete: bo _, stderr = await task if process.returncode == 0: converts_success = True - logger.info(f"Video transcoding completed: {save_path}") + logger.info(f"Recording conversion completed: {save_path}") else: logger.error( - f"Video transcoding failed! Error message: {stderr.decode() if stderr else 'Unknown error'}") + f"Recording conversion failed! Error message: {stderr.decode() if stderr else 'Unknown error'}") except subprocess.CalledProcessError as e: - logger.error(f"Video transcoding failed! Error message: {e.output.decode()}") + logger.error(f"Recording conversion failed! Error message: {e.output.decode()}") try: if converts_success: if is_original_delete: await asyncio.sleep(1) - if os.path.exists(converts_file_path): - os.remove(converts_file_path) - logger.info(f"Delete Original File: {converts_file_path}") + if os.path.exists(source_file_path): + os.remove(source_file_path) + logger.info(f"Delete Original File: {source_file_path}") else: converts_dir = f"{os.path.dirname(save_path)}/original" os.makedirs(converts_dir, exist_ok=True) - shutil.move(converts_file_path, converts_dir) - logger.info(f"Move Transcoding Files: {converts_file_path}") + shutil.move(source_file_path, converts_dir) + logger.info(f"Move Converted Source File: {source_file_path}") + return save_path except subprocess.CalledProcessError as e: logger.error(f"Error occurred during conversion: {e}") except Exception as e: logger.error(f"An unknown error occurred: {e}") + return None + async def custom_script_execute( self, script_command: str, @@ -850,6 +1042,8 @@ async def start_direct_download( logger.info(f"Starting direct download - recorder id: {id(self)}, rec_id: {self.recording.rec_id}") self.should_stop = False self.auto_stop_requested = False + runtime_task = asyncio.current_task() + self.app.record_manager.register_runtime_task(runtime_task) try: await self.direct_downloader.start_download() @@ -973,6 +1167,7 @@ async def start_direct_download( logger.debug(f"Failed to update UI: {e}") return False finally: + self.app.record_manager.unregister_runtime_task(runtime_task) self.recording.record_url = None self.auto_stop_requested = False diff --git a/app/core/runtime/process_manager.py b/app/core/runtime/process_manager.py index 118ab119..91d7e874 100644 --- a/app/core/runtime/process_manager.py +++ b/app/core/runtime/process_manager.py @@ -1,6 +1,7 @@ import asyncio import os import threading +import time from ...utils.logger import logger @@ -49,6 +50,28 @@ def _process_tasks(self): logger.info("All background tasks completed, service stopped") self.is_running = False + def has_pending_work(self) -> bool: + return bool(self.tasks) or bool(self.worker_thread and self.worker_thread.is_alive()) + + def wait_for_completion(self, timeout: float | None = None, poll_interval: float = 0.1) -> bool: + deadline = None if timeout is None else time.monotonic() + timeout + + while self.has_pending_work(): + if deadline is not None: + remaining = deadline - time.monotonic() + if remaining <= 0: + return False + wait_time = min(poll_interval, remaining) + else: + wait_time = poll_interval + + if self.worker_thread and self.worker_thread.is_alive(): + self.worker_thread.join(wait_time) + else: + time.sleep(wait_time) + + return True + class AsyncProcessManager: def __init__(self): diff --git a/app/lifecycle/app_close_handler.py b/app/lifecycle/app_close_handler.py index 4dd25e12..e0f6c3a9 100644 --- a/app/lifecycle/app_close_handler.py +++ b/app/lifecycle/app_close_handler.py @@ -2,6 +2,7 @@ import flet as ft +from ..core.runtime.process_manager import BackgroundService from ..utils.logger import logger from .tray_manager import TrayManager @@ -65,6 +66,40 @@ async def close_dialog_dismissed(e): except Exception as ex: logger.error(f"close window error: {ex}") + post_process_timeout = max(10, min(active_recordings_count * 6, 90)) + logger.info( + f"Waiting for recording shutdown tasks to finish " + f"(timeout: {post_process_timeout}s)" + ) + try: + runtime_tasks_done = await asyncio.wait_for( + app.record_manager.wait_for_runtime_tasks(), + timeout=post_process_timeout, + ) + if not runtime_tasks_done: + logger.warning("Some recording shutdown tasks are still pending after the wait window") + except asyncio.TimeoutError: + logger.warning("Timed out while waiting for recording shutdown tasks during shutdown") + except Exception as ex: + logger.error(f"Failed while waiting for recording shutdown tasks: {ex}") + + background_service = BackgroundService.get_instance() + if background_service.has_pending_work(): + background_timeout = max(10, min(active_recordings_count * 8, 120)) + logger.info( + f"Waiting for background conversion tasks to finish " + f"(timeout: {background_timeout}s)" + ) + try: + background_done = await asyncio.to_thread( + background_service.wait_for_completion, + background_timeout, + ) + if not background_done: + logger.warning("Timed out while waiting for background conversion tasks during shutdown") + except Exception as ex: + logger.error(f"Failed while waiting for background conversion tasks: {ex}") + remaining = len([p for p in app.process_manager.ffmpeg_processes if p.returncode is None]) if remaining > 0: logger.warning(f"{remaining} recording processes are still active during shutdown") diff --git a/tests/test_mp4_pipeline.py b/tests/test_mp4_pipeline.py index 06ef8820..92575a54 100644 --- a/tests/test_mp4_pipeline.py +++ b/tests/test_mp4_pipeline.py @@ -1,3 +1,5 @@ +import os +import tempfile import unittest from types import SimpleNamespace @@ -30,35 +32,158 @@ def test_non_segmented_mp4_avoids_delay_moov(self): class LiveStreamRecorderFormatSelectionTests(unittest.TestCase): - def test_hls_mp4_requests_are_captured_as_ts(self): + def test_hls_requested_video_formats_are_captured_as_ts(self): stream_info = SimpleNamespace( record_url="https://cdn.example.com/live/index.m3u8", m3u8_url="https://cdn.example.com/live/index.m3u8", ) self.assertTrue( - LiveStreamRecorder.should_capture_as_ts_for_mp4( + LiveStreamRecorder.should_capture_as_ts_for_requested_format( requested_format="mp4", use_direct_download=False, record_url=stream_info.record_url, stream_info=stream_info, ) ) + self.assertTrue( + LiveStreamRecorder.should_capture_as_ts_for_requested_format( + requested_format="mkv", + use_direct_download=False, + record_url=stream_info.record_url, + stream_info=stream_info, + ) + ) - def test_non_hls_mp4_requests_keep_mp4_capture(self): + def test_non_hls_requested_formats_keep_direct_capture(self): stream_info = SimpleNamespace( record_url="https://cdn.example.com/live.flv", m3u8_url=None, ) self.assertFalse( - LiveStreamRecorder.should_capture_as_ts_for_mp4( + LiveStreamRecorder.should_capture_as_ts_for_requested_format( requested_format="mp4", use_direct_download=False, record_url=stream_info.record_url, stream_info=stream_info, ) ) + self.assertFalse( + LiveStreamRecorder.should_capture_as_ts_for_requested_format( + requested_format="ts", + use_direct_download=False, + record_url=stream_info.record_url, + stream_info=stream_info, + ) + ) + + def test_temporary_ts_capture_always_removes_intermediates(self): + self.assertTrue( + LiveStreamRecorder.should_delete_original_after_conversion( + delete_original_setting=False, + uses_temporary_capture=True, + ) + ) + + def test_regular_ts_conversion_still_respects_delete_original_setting(self): + self.assertFalse( + LiveStreamRecorder.should_delete_original_after_conversion( + delete_original_setting=False, + uses_temporary_capture=False, + ) + ) + self.assertTrue( + LiveStreamRecorder.should_delete_original_after_conversion( + delete_original_setting=True, + uses_temporary_capture=False, + ) + ) + + def test_temporary_ts_capture_converts_back_to_requested_format(self): + self.assertEqual( + LiveStreamRecorder.get_post_record_conversion_target( + capture_format="ts", + requested_format="mov", + convert_to_mp4_setting=False, + uses_temporary_capture=True, + ), + "mov", + ) + + def test_regular_ts_capture_can_still_optionally_generate_mp4(self): + self.assertEqual( + LiveStreamRecorder.get_post_record_conversion_target( + capture_format="ts", + requested_format="ts", + convert_to_mp4_setting=True, + uses_temporary_capture=False, + ), + "mp4", + ) + + def test_converted_output_path_uses_target_extension(self): + self.assertEqual( + LiveStreamRecorder.get_converted_output_path("C:/tmp/out_001.ts", "mkv"), + "C:/tmp/out_001.mkv", + ) + + def test_segmented_incremental_conversion_excludes_latest_active_segment(self): + recorder = LiveStreamRecorder.__new__(LiveStreamRecorder) + recorder.segment_record = True + recorder.segment_conversion_tasks = {} + + with tempfile.TemporaryDirectory() as temp_dir: + for index in range(3): + open(os.path.join(temp_dir, f"out_{index:03d}.ts"), "wb").close() + + save_file_path = os.path.join(temp_dir, "out_%03d.ts").replace("\\", "/") + self.assertEqual( + recorder._get_conversion_candidates(save_file_path, include_latest_segment=False), + [ + os.path.join(temp_dir, "out_000.ts").replace("\\", "/"), + os.path.join(temp_dir, "out_001.ts").replace("\\", "/"), + ], + ) + + def test_segmented_final_conversion_includes_latest_segment(self): + recorder = LiveStreamRecorder.__new__(LiveStreamRecorder) + recorder.segment_record = True + recorder.segment_conversion_tasks = {} + + with tempfile.TemporaryDirectory() as temp_dir: + for index in range(2): + open(os.path.join(temp_dir, f"out_{index:03d}.ts"), "wb").close() + + save_file_path = os.path.join(temp_dir, "out_%03d.ts").replace("\\", "/") + self.assertEqual( + recorder._get_conversion_candidates(save_file_path, include_latest_segment=True), + [ + os.path.join(temp_dir, "out_000.ts").replace("\\", "/"), + os.path.join(temp_dir, "out_001.ts").replace("\\", "/"), + ], + ) + + def test_progress_marker_ignores_removed_old_segment_during_incremental_conversion(self): + recorder = LiveStreamRecorder.__new__(LiveStreamRecorder) + recorder.segment_record = True + + with tempfile.TemporaryDirectory() as temp_dir: + old_segment = os.path.join(temp_dir, "out_000.ts") + latest_segment = os.path.join(temp_dir, "out_001.ts") + + with open(old_segment, "wb") as file: + file.write(b"old-segment") + with open(latest_segment, "wb") as file: + file.write(b"latest") + + save_file_path = os.path.join(temp_dir, "out_%03d.ts").replace("\\", "/") + marker_before = recorder._get_output_progress_marker(save_file_path) + + os.remove(old_segment) + + marker_after = recorder._get_output_progress_marker(save_file_path) + self.assertEqual(marker_before, marker_after) if __name__ == "__main__": diff --git a/tests/test_recording_manager.py b/tests/test_recording_manager.py index 4765c9d2..12e2ee9d 100644 --- a/tests/test_recording_manager.py +++ b/tests/test_recording_manager.py @@ -1,3 +1,4 @@ +import asyncio import unittest from types import SimpleNamespace @@ -82,6 +83,21 @@ async def test_add_recordings_persists_once_for_multiple_items(self): ["rec-1", "rec-2"], ) + async def test_wait_for_runtime_tasks_waits_for_registered_task_completion(self): + manager = RecordingManager(self.app) + + async def background_work(): + await asyncio.sleep(0.01) + + task = asyncio.create_task(background_work()) + manager.register_runtime_task(task) + + completed = await manager.wait_for_runtime_tasks(timeout=1) + + self.assertTrue(completed) + self.assertTrue(task.done()) + self.assertNotIn(task, manager.active_runtime_tasks) + if __name__ == "__main__": unittest.main()