From b7d009e7c39d951912f83caa4bc3b66a594b4451 Mon Sep 17 00:00:00 2001 From: mhdzumair Date: Wed, 23 Apr 2025 22:23:34 +0530 Subject: [PATCH] Improve error handling and buffering in m3u8 processing --- mediaflow_proxy/handlers.py | 3 ++ mediaflow_proxy/utils/http_utils.py | 10 ++-- mediaflow_proxy/utils/m3u8_processor.py | 62 +++++++++---------------- 3 files changed, 32 insertions(+), 43 deletions(-) diff --git a/mediaflow_proxy/handlers.py b/mediaflow_proxy/handlers.py index f225617..08e741e 100644 --- a/mediaflow_proxy/handlers.py +++ b/mediaflow_proxy/handlers.py @@ -3,6 +3,7 @@ import logging from urllib.parse import urlparse, parse_qs import httpx +import tenacity from fastapi import Request, Response, HTTPException from starlette.background import BackgroundTask @@ -52,6 +53,8 @@ def handle_exceptions(exception: Exception) -> Response: elif isinstance(exception, DownloadError): logger.error(f"Error downloading content: {exception}") return Response(status_code=exception.status_code, content=str(exception)) + elif isinstance(exception, tenacity.RetryError): + return Response(status_code=502, content="Max retries exceeded while downloading content") else: logger.exception(f"Internal server error while handling request: {exception}") return Response(status_code=502, content=f"Internal server error: {exception}") diff --git a/mediaflow_proxy/utils/http_utils.py b/mediaflow_proxy/utils/http_utils.py index 9417699..c0b9aca 100644 --- a/mediaflow_proxy/utils/http_utils.py +++ b/mediaflow_proxy/utils/http_utils.py @@ -33,9 +33,8 @@ class DownloadError(Exception): def create_httpx_client(follow_redirects: bool = True, **kwargs) -> httpx.AsyncClient: """Creates an HTTPX client with configured proxy routing""" mounts = settings.transport_config.get_mounts() - client = httpx.AsyncClient( - mounts=mounts, follow_redirects=follow_redirects, timeout=settings.transport_config.timeout, **kwargs - ) + kwargs.setdefault("timeout", settings.transport_config.timeout) + client = httpx.AsyncClient(mounts=mounts, follow_redirects=follow_redirects, **kwargs) return client @@ -125,9 +124,12 @@ class Streamer: raise DownloadError( e.response.status_code, f"HTTP error {e.response.status_code} while creating streaming response" ) + except httpx.RequestError as e: + logger.error(f"Error creating streaming response: {e}") + raise DownloadError(502, f"Error creating streaming response: {e}") except Exception as e: logger.error(f"Error creating streaming response: {e}") - raise + raise RuntimeError(f"Error creating streaming response: {e}") async def stream_content(self) -> typing.AsyncGenerator[bytes, None]: """ diff --git a/mediaflow_proxy/utils/m3u8_processor.py b/mediaflow_proxy/utils/m3u8_processor.py index 1ed0fc7..87e0f6a 100644 --- a/mediaflow_proxy/utils/m3u8_processor.py +++ b/mediaflow_proxy/utils/m3u8_processor.py @@ -1,4 +1,4 @@ -import logging +import codecs import re from typing import AsyncGenerator from urllib import parse @@ -57,54 +57,38 @@ class M3U8Processor: Yields: str: Processed lines of the m3u8 content. """ - buffer = b"" # Use bytes buffer instead of str + buffer = "" # String buffer for decoded content + decoder = codecs.getincrementaldecoder("utf-8")(errors="replace") # Process the content chunk by chunk async for chunk in content_iterator: if isinstance(chunk, str): chunk = chunk.encode("utf-8") - buffer += chunk + # Incrementally decode the chunk + decoded_chunk = decoder.decode(chunk) + buffer += decoded_chunk - try: - # Try to decode the current buffer with error handling - decoded_buffer = buffer.decode("utf-8", errors="replace") - lines = decoded_buffer.split("\n") + # Process complete lines + lines = buffer.split("\n") + if len(lines) > 1: + # Process all complete lines except the last one + for line in lines[:-1]: + if line: # Skip empty lines + processed_line = await self.process_line(line, base_url) + yield processed_line + "\n" - # If we have at least one line and the buffer ends with newline, - # we process all lines including the last one - if len(lines) > 1: - # Process all complete lines except the last one - for line in lines[:-1]: - if line: # Skip empty lines - processed_line = await self.process_line(line, base_url) - yield processed_line + "\n" + # Keep the last line in the buffer (it might be incomplete) + buffer = lines[-1] - # If the buffer ended with a newline, the last line is also complete - if decoded_buffer.endswith("\n"): - if lines[-1]: # Process the last line if it's not empty - processed_line = await self.process_line(lines[-1], base_url) - yield processed_line + "\n" - buffer = b"" # Clear the buffer - else: - # Keep the last incomplete line in the buffer - buffer = lines[-1].encode("utf-8") - except UnicodeDecodeError: - # If we can't decode the buffer, keep accumulating data - continue + # Process any remaining data in the buffer plus final bytes + final_chunk = decoder.decode(b"", final=True) + if final_chunk: + buffer += final_chunk - # Process any remaining data in the buffer - if buffer: - try: - # Final attempt to decode any remaining data - remaining = buffer.decode("utf-8", errors="replace") - if remaining: - processed_line = await self.process_line(remaining, base_url) - yield processed_line - except Exception as e: - logging.error(f"Error processing final buffer: {e}") - # Yield any remaining content with replacement characters - yield await self.process_line(buffer.decode("utf-8", errors="replace"), base_url) + if buffer: # Process the last line if it's not empty + processed_line = await self.process_line(buffer, base_url) + yield processed_line async def process_line(self, line: str, base_url: str) -> str: """