diff --git a/api_server.py b/api_server.py index 97f2f84..f503a0d 100644 --- a/api_server.py +++ b/api_server.py @@ -67,29 +67,10 @@ if not logger.handlers: pass client = TelegramClient() Config = get_settings() -DOWNLOAD_SEMAPHORE = asyncio.Semaphore(3) -_semaphore_waiters = 0 # diagnostic: track how many requests wait for semaphore +HTTP_DOWNLOAD_SEMAPHORE = asyncio.Semaphore(3) # semaphore for live HTTP media requests +BACKGROUND_DOWNLOAD_SEMAPHORE = asyncio.Semaphore(2) # semaphore for background cache worker download_queue = asyncio.Queue(maxsize=100) -async def _diag_event_loop_monitor(): - """Periodic diagnostic: logs event loop tasks, semaphore state, and threadpool active threads.""" - import threading - while True: - await asyncio.sleep(30) - loop = asyncio.get_event_loop() - all_tasks = asyncio.all_tasks(loop) - task_names = [t.get_name() for t in all_tasks] - logger.warning( - f"[DIAG] event_loop_monitor: " - f"active_tasks={len(all_tasks)}, " - f"semaphore_value={DOWNLOAD_SEMAPHORE._value}, " - f"semaphore_waiters={_semaphore_waiters}, " - f"download_queue_size={download_queue.qsize()}, " - f"active_threads={threading.active_count()}, " - f"tasks={task_names}" - ) - - @asynccontextmanager async def lifespan(_: FastAPI): setup_logging(Config["log_level"]) @@ -101,11 +82,9 @@ async def lifespan(_: FastAPI): await client.start() background_task = asyncio.create_task(cache_media_files()) # Start background task worker_task = asyncio.create_task(background_download_worker()) # Start download worker - diag_task = asyncio.create_task(_diag_event_loop_monitor()) # Start diagnostic monitor yield background_task.cancel() # Cleanup worker_task.cancel() - diag_task.cancel() try: await background_task except asyncio.CancelledError: @@ -114,10 +93,6 @@ async def lifespan(_: FastAPI): await worker_task except asyncio.CancelledError: pass - try: - await diag_task - except asyncio.CancelledError: - pass await client.stop() app = FastAPI(title="Pyrogram Bridge", lifespan=lifespan) @@ -218,10 +193,7 @@ def delayed_delete_file(file_path: str, delay: int = 300) -> None: Delay is set to 5 minutes by default. NOTE: Runs in a background thread via Starlette BackgroundTask. """ - import threading - logger.warning(f"[DIAG] delayed_delete_file: starting sleep({delay}s) in thread {threading.current_thread().name}, active threads: {threading.active_count()}") time.sleep(delay) - logger.warning(f"[DIAG] delayed_delete_file: sleep finished, deleting {file_path}, active threads: {threading.active_count()}") try: os.remove(file_path) logger.info(f"Deleted temporary file {file_path} after delay of {delay} seconds") @@ -255,146 +227,140 @@ async def download_media_file(channel: Union[str, int], post_id: int, file_uniqu Download media file from Telegram and save to cache Returns tuple of (file path, delete_after) """ - global _semaphore_waiters - _semaphore_waiters += 1 - logger.warning(f"[DIAG] download_media_file: waiting for semaphore, waiters={_semaphore_waiters}, semaphore_value={DOWNLOAD_SEMAPHORE._value}, channel={channel}, post_id={post_id}") - async with DOWNLOAD_SEMAPHORE: - _semaphore_waiters -= 1 - logger.warning(f"[DIAG] download_media_file: acquired semaphore, waiters_remaining={_semaphore_waiters}, semaphore_value={DOWNLOAD_SEMAPHORE._value}, channel={channel}, post_id={post_id}") - base_cache_dir = os.path.abspath("./data/cache") - - # Create nested cache structure - channel_dir = os.path.join(base_cache_dir, str(channel)) - post_dir = os.path.join(channel_dir, str(post_id)) - os.makedirs(post_dir, exist_ok=True) - - # Convert numeric channel ID to int if needed - channel_id: Union[str, int] = channel - if isinstance(channel, str) and channel.startswith('-100'): - channel_id = int(channel) - + base_cache_dir = os.path.abspath("./data/cache") + + # Create nested cache structure + channel_dir = os.path.join(base_cache_dir, str(channel)) + post_dir = os.path.join(channel_dir, str(post_id)) + os.makedirs(post_dir, exist_ok=True) + + # Convert numeric channel ID to int if needed + channel_id: Union[str, int] = channel + if isinstance(channel, str) and channel.startswith('-100'): + channel_id = int(channel) + + try: + message = await client.safe_get_messages(channel_id, post_id) + except asyncio.TimeoutError: + logger.error(f"Timeout getting messages for {channel}/{post_id}") + raise HTTPException(status_code=504, detail="Request timeout") + + if message.media == "MessageMediaType.POLL": + return None, False + + # Check if it is a video and if its size exceeds 100 MB + is_large_video = False + if message.video: try: - message = await client.safe_get_messages(channel_id, post_id) - except asyncio.TimeoutError: - logger.error(f"Timeout getting messages for {channel}/{post_id}") - raise HTTPException(status_code=504, detail="Request timeout") - - if message.media == "MessageMediaType.POLL": - return None, False - - # Check if it is a video and if its size exceeds 100 MB - is_large_video = False - if message.video: - try: - if message.video.file_size > 100 * 1024 * 1024: - is_large_video = True - except Exception as e: - logger.error(f"Failed to get video file size for message {post_id}: {str(e)}") - - if is_large_video: - # For large video, download without permanent caching; use a temporary file - temp_file_path = os.path.join(post_dir, f"temp_{file_unique_id}") - if os.path.exists(temp_file_path): - logger.info(f"Temporary file {temp_file_path} already exists, serving cached large video") - return temp_file_path, False - file_id = await find_file_id_in_message(message, file_unique_id) - if not file_id: - logger.error(f"Media with file_unique_id {file_unique_id} not found in message {post_id}") - raise HTTPException(status_code=404, detail="File not found in message") - logger.info(f"Downloading large video file {file_unique_id} to temporary path {temp_file_path}") - try: - file_path = await client.safe_download_media(file_id, temp_file_path) - except asyncio.TimeoutError: - logger.error(f"Timeout downloading large video {file_unique_id}") - raise HTTPException(status_code=504, detail="Download timeout") - logger.info(f"Downloaded large video file {file_unique_id} to temporary path {temp_file_path}") - return file_path, False - - # Normal caching flow - cache_path = os.path.join(post_dir, file_unique_id) - if os.path.exists(cache_path): - # Check if the cached file is zero size - if os.path.getsize(cache_path) == 0: - logger.warning(f"zero_size_cache_found: Found zero-size cached file: {cache_path}. Deleting and attempting redownload.") - try: - os.remove(cache_path) - logger.info(f"Removed zero-size cached file: {cache_path}") - except OSError as e: - logger.error(f"cleanup_error: Failed to remove zero-size cached file {cache_path}: {e}") - # Do not raise error here, proceed to download below - else: - # File exists and is not zero size, update timestamp and return - logger.info(f"Found cached media file: {cache_path}") - # Update access timestamp in media_file_ids.json - file_ids_path = os.path.join(os.path.abspath("./data"), 'media_file_ids.json') - try: - if os.path.exists(file_ids_path): - media_files = await asyncio.to_thread(read_json_file_sync, file_ids_path) - for file_data in media_files: - if (file_data.get('channel') == str(channel) and # Ensure channel is string for comparison - file_data.get('post_id') == post_id and - file_data.get('file_unique_id') == file_unique_id): - file_data['added'] = datetime.now().timestamp() - break - async with get_media_file_ids_lock(): - await asyncio.to_thread(write_json_file_sync, file_ids_path, media_files) - except Exception as e: - logger.error(f"Failed to update timestamp for {channel}/{post_id}/{file_unique_id}: {str(e)}") - return cache_path, False - + if message.video.file_size > 100 * 1024 * 1024: + is_large_video = True + except Exception as e: + logger.error(f"Failed to get video file size for message {post_id}: {str(e)}") + + if is_large_video: + # For large video, download without permanent caching; use a temporary file + temp_file_path = os.path.join(post_dir, f"temp_{file_unique_id}") + if os.path.exists(temp_file_path): + logger.info(f"Temporary file {temp_file_path} already exists, serving cached large video") + return temp_file_path, False file_id = await find_file_id_in_message(message, file_unique_id) if not file_id: - error_message = f"Media with file_unique_id {file_unique_id} not found in message {post_id} for channel {channel}" - logger.error(error_message) - - # Attempt to remove the invalid entry from media_file_ids.json using threadpool I/O + logger.error(f"Media with file_unique_id {file_unique_id} not found in message {post_id}") + raise HTTPException(status_code=404, detail="File not found in message") + logger.info(f"Downloading large video file {file_unique_id} to temporary path {temp_file_path}") + try: + file_path = await client.safe_download_media(file_id, temp_file_path) + except asyncio.TimeoutError: + logger.error(f"Timeout downloading large video {file_unique_id}") + raise HTTPException(status_code=504, detail="Download timeout") + logger.info(f"Downloaded large video file {file_unique_id} to temporary path {temp_file_path}") + return file_path, False + + # Normal caching flow + cache_path = os.path.join(post_dir, file_unique_id) + if os.path.exists(cache_path): + # Check if the cached file is zero size + if os.path.getsize(cache_path) == 0: + logger.warning(f"zero_size_cache_found: Found zero-size cached file: {cache_path}. Deleting and attempting redownload.") + try: + os.remove(cache_path) + logger.info(f"Removed zero-size cached file: {cache_path}") + except OSError as e: + logger.error(f"cleanup_error: Failed to remove zero-size cached file {cache_path}: {e}") + # Do not raise error here, proceed to download below + else: + # File exists and is not zero size, update timestamp and return + logger.info(f"Found cached media file: {cache_path}") + # Update access timestamp in media_file_ids.json file_ids_path = os.path.join(os.path.abspath("./data"), 'media_file_ids.json') try: if os.path.exists(file_ids_path): media_files = await asyncio.to_thread(read_json_file_sync, file_ids_path) - initial_count = len(media_files) - media_files_updated = [ - f_data for f_data in media_files - if not ( - f_data.get('channel') == str(channel) and - f_data.get('post_id') == post_id and - f_data.get('file_unique_id') == file_unique_id - ) - ] - if len(media_files_updated) < initial_count: - async with get_media_file_ids_lock(): - await asyncio.to_thread(write_json_file_sync, file_ids_path, media_files_updated) - logger.info(f"Removed invalid entry for {channel}/{post_id}/{file_unique_id} from media_file_ids.json") - else: - logger.warning(f"Entry for {channel}/{post_id}/{file_unique_id} not found in media_file_ids.json for removal") - + for file_data in media_files: + if (file_data.get('channel') == str(channel) and # Ensure channel is string for comparison + file_data.get('post_id') == post_id and + file_data.get('file_unique_id') == file_unique_id): + file_data['added'] = datetime.now().timestamp() + break + async with get_media_file_ids_lock(): + await asyncio.to_thread(write_json_file_sync, file_ids_path, media_files) except Exception as e: - logger.error(f"Failed to remove entry for {channel}/{post_id}/{file_unique_id} from media_file_ids.json: {str(e)}") - - raise HTTPException(status_code=404, detail="File not found in message") - - try: - file_path = await client.safe_download_media(file_id, cache_path) - except asyncio.TimeoutError: - logger.error(f"Timeout downloading media {file_unique_id}") - raise HTTPException(status_code=504, detail="Download timeout") - - # Check if the downloaded file exists and has a size greater than zero - if not file_path or not os.path.exists(file_path) or os.path.getsize(file_path) == 0: - logger.error(f"download_failed_zero_size: Downloaded file {file_unique_id} for {channel}/{post_id} is zero size or missing.") - # Attempt to clean up the invalid file - if file_path and os.path.exists(file_path): - try: - os.remove(file_path) - logger.info(f"Removed zero-size file: {file_path}") - except OSError as e: - logger.error(f"cleanup_error: Failed to remove zero-size file {file_path}: {e}") - # Raise an error to indicate download failure - # Raise specific error - raise ZeroSizeFileError(f"Downloaded file {file_unique_id} for {channel}/{post_id} is zero size or missing after download attempt.") + logger.error(f"Failed to update timestamp for {channel}/{post_id}/{file_unique_id}: {str(e)}") + return cache_path, False - logger.info(f"Downloaded media file {file_unique_id} to {cache_path}") - return file_path, False + file_id = await find_file_id_in_message(message, file_unique_id) + if not file_id: + error_message = f"Media with file_unique_id {file_unique_id} not found in message {post_id} for channel {channel}" + logger.error(error_message) + + # Attempt to remove the invalid entry from media_file_ids.json using threadpool I/O + file_ids_path = os.path.join(os.path.abspath("./data"), 'media_file_ids.json') + try: + if os.path.exists(file_ids_path): + media_files = await asyncio.to_thread(read_json_file_sync, file_ids_path) + initial_count = len(media_files) + media_files_updated = [ + f_data for f_data in media_files + if not ( + f_data.get('channel') == str(channel) and + f_data.get('post_id') == post_id and + f_data.get('file_unique_id') == file_unique_id + ) + ] + if len(media_files_updated) < initial_count: + async with get_media_file_ids_lock(): + await asyncio.to_thread(write_json_file_sync, file_ids_path, media_files_updated) + logger.info(f"Removed invalid entry for {channel}/{post_id}/{file_unique_id} from media_file_ids.json") + else: + logger.warning(f"Entry for {channel}/{post_id}/{file_unique_id} not found in media_file_ids.json for removal") + + except Exception as e: + logger.error(f"Failed to remove entry for {channel}/{post_id}/{file_unique_id} from media_file_ids.json: {str(e)}") + + raise HTTPException(status_code=404, detail="File not found in message") + + try: + file_path = await client.safe_download_media(file_id, cache_path) + except asyncio.TimeoutError: + logger.error(f"Timeout downloading media {file_unique_id}") + raise HTTPException(status_code=504, detail="Download timeout") + + # Check if the downloaded file exists and has a size greater than zero + if not file_path or not os.path.exists(file_path) or os.path.getsize(file_path) == 0: + logger.error(f"download_failed_zero_size: Downloaded file {file_unique_id} for {channel}/{post_id} is zero size or missing.") + # Attempt to clean up the invalid file + if file_path and os.path.exists(file_path): + try: + os.remove(file_path) + logger.info(f"Removed zero-size file: {file_path}") + except OSError as e: + logger.error(f"cleanup_error: Failed to remove zero-size file {file_path}: {e}") + # Raise an error to indicate download failure + # Raise specific error + raise ZeroSizeFileError(f"Downloaded file {file_unique_id} for {channel}/{post_id} is zero size or missing after download attempt.") + + logger.info(f"Downloaded media file {file_unique_id} to {cache_path}") + return file_path, False def remove_old_cached_files_sync(media_files: list, cache_dir: str) -> tuple[list, int]: @@ -556,10 +522,10 @@ async def background_download_worker(): try: channel, post_id, file_unique_id = await download_queue.get() logger.info(f"Background download: {channel}/{post_id}/{file_unique_id}") - logger.warning(f"[DIAG] background_download_worker: queue_size={download_queue.qsize()}, semaphore_value={DOWNLOAD_SEMAPHORE._value}, waiters={_semaphore_waiters}") try: - await download_media_file(channel, post_id, file_unique_id) + async with BACKGROUND_DOWNLOAD_SEMAPHORE: # limit concurrent background downloads + await download_media_file(channel, post_id, file_unique_id) await asyncio.sleep(2) except Exception as e: logger.error(f"Background download error for {channel}/{post_id}/{file_unique_id}: {e}") @@ -823,7 +789,8 @@ async def get_media(channel: str, post_id: int, file_unique_id: str, digest: str channel_id = int(channel) try: # Wrap the download and prepare call - file_path, delete_after = await download_media_file(channel_id, post_id, file_unique_id) + async with HTTP_DOWNLOAD_SEMAPHORE: # limit concurrent live HTTP downloads + file_path, delete_after = await download_media_file(channel_id, post_id, file_unique_id) if not file_path: raise HTTPException(status_code=404, detail="File not found") if file_path: