From d41ee5319e9f585f964f446f58fabf6aaec279e7 Mon Sep 17 00:00:00 2001 From: James Hush Date: Thu, 23 Oct 2025 14:37:39 +0800 Subject: [PATCH] Add example for getting just recording links --- fetch_s3_recording.py | 93 ++++++++----- fetch_s3_recording_by_room.py | 238 ++++++++++++++++++++++++++++++++++ 2 files changed, 299 insertions(+), 32 deletions(-) create mode 100755 fetch_s3_recording_by_room.py diff --git a/fetch_s3_recording.py b/fetch_s3_recording.py index 5c2cefb60..2e0a56bbe 100755 --- a/fetch_s3_recording.py +++ b/fetch_s3_recording.py @@ -141,19 +141,19 @@ async def get_recording_s3_url( return (recording_url, recording_url, status) -async def get_recent_rooms(limit: int = 100) -> list[str]: - """Get list of recent room names from Daily API. +async def get_recent_recordings(limit: int = 100) -> list[str]: + """Get list of recent recording IDs from Daily API. Args: - limit: Maximum number of rooms to retrieve (default 100, max 100) + limit: Maximum number of recordings to retrieve (default 100) Returns: - List of room names (strings) + List of recording IDs (strings) """ try: async with AsyncClient() as client: response = await client.get( - url=f"https://api.daily.co/v1/rooms?limit={min(limit, 100)}", + url="https://api.daily.co/v1/recordings", headers={ "Authorization": f"Bearer {os.getenv('DAILY_API_KEY')}", "Content-Type": "application/json", @@ -163,39 +163,66 @@ async def get_recent_rooms(limit: int = 100) -> list[str]: response.raise_for_status() data = response.json() - rooms = data.get("data", []) - room_names = [room.get("name") for room in rooms if room.get("name")] + recordings = data.get("data", []) + recording_ids = [rec.get("id") for rec in recordings[:limit] if rec.get("id")] - logger.info(f"Retrieved {len(room_names)} room names from Daily API") - return room_names + logger.info(f"Retrieved {len(recording_ids)} recording IDs from Daily API") + return recording_ids except Exception as e: - logger.exception(f"Failed to get rooms from Daily API: {e}") + logger.exception(f"Failed to get recordings from Daily API: {e}") return [] async def main(): - """Test get_recording_s3_url_with_retry with recent rooms.""" + """Test get_recording_s3_url_with_retry with recent recordings.""" logger.info("Starting recording fetch test...") - # Step 1: Get the most recent 100 rooms - logger.info("Fetching recent rooms...") - room_names = await get_recent_rooms(limit=100) + # Step 1: Get the most recent 100 recordings + logger.info("Fetching recent recordings...") + recording_ids = await get_recent_recordings(limit=100) - if not room_names: - logger.error("No rooms found. Cannot proceed with test.") + if not recording_ids: + logger.error("No recordings found. Cannot proceed with test.") return - logger.info(f"Found {len(room_names)} rooms to check for recordings") + logger.info(f"Found {len(recording_ids)} recordings to fetch") - # Call get_recording_s3_url_with_retry on each room concurrently - logger.info(f"Attempting to fetch recordings for {len(room_names)} rooms concurrently...") + # Fetch access links for each recording concurrently + logger.info( + f"Attempting to fetch access links for {len(recording_ids)} recordings concurrently..." + ) - # Create tasks for all rooms - tasks = [ - get_recording_s3_url_with_retry(room_id=room_name, max_retries=3) - for room_name in room_names - ] + # Create tasks for all recordings + async def get_recording_link(recording_id: str) -> Tuple[Optional[str], Optional[str]]: + """Get download link for a specific recording ID.""" + try: + headers = { + "Authorization": f"Bearer {os.getenv('DAILY_API_KEY')}", + "Content-Type": "application/json", + } + + async with AsyncClient(timeout=180) as client: + # Get the recording access link + link_response = await client.get( + url=f"https://api.daily.co/v1/recordings/{recording_id}/access-link", + headers=headers, + ) + link_response.raise_for_status() + link_data = link_response.json() + + recording_url = link_data.get("download_link") + if not recording_url: + logger.warning(f"No download link found for recording {recording_id}") + return (None, None) + + return (recording_url, recording_url) + + except Exception as e: + logger.exception(f"Failed to get access link for recording {recording_id}: {e}") + return (None, None) + + tasks = [get_recording_link(recording_id) for recording_id in recording_ids] # Execute all tasks concurrently results = await asyncio.gather(*tasks, return_exceptions=True) @@ -205,32 +232,34 @@ async def main(): not_found_count = 0 failed_count = 0 - for i, (room_name, result) in enumerate(zip(room_names, results), 1): + for i, (recording_id, result) in enumerate(zip(recording_ids, results), 1): if isinstance(result, Exception): failed_count += 1 - logger.error(f"❌ [{i}/{len(room_names)}] Failed for {room_name}: {result}") + logger.error(f"❌ [{i}/{len(recording_ids)}] Failed for {recording_id}: {result}") elif isinstance(result, tuple) and len(result) == 2: recording_url, recording_signed_url = result if recording_url: success_count += 1 - logger.info(f"✅ [{i}/{len(room_names)}] Found recording for {room_name}") - logger.debug(f" URL: {recording_url[:80]}...") + logger.info( + f"✅ [{i}/{len(recording_ids)}] Found recording link for {recording_id}" + ) + logger.debug(f" URL: {recording_url}") else: not_found_count += 1 - logger.debug(f"ℹ️ [{i}/{len(room_names)}] No recording for {room_name}") + logger.debug(f"ℹ️ [{i}/{len(recording_ids)}] No link for {recording_id}") else: failed_count += 1 - logger.error(f"❌ [{i}/{len(room_names)}] Unexpected result type for {room_name}") + logger.error(f"❌ [{i}/{len(recording_ids)}] Unexpected result type for {recording_id}") # Summary logger.info("\n" + "=" * 60) logger.info("RECORDING FETCH TEST SUMMARY") logger.info("=" * 60) - logger.info(f"Total rooms checked: {len(room_names)}") + logger.info(f"Total recordings checked: {len(recording_ids)}") logger.info(f"✅ Recordings found: {success_count}") logger.info(f"ℹ️ No recordings: {not_found_count}") logger.info(f"❌ Failed: {failed_count}") - logger.info(f"Success rate: {(success_count / len(room_names) * 100):.2f}%") + logger.info(f"Success rate: {(success_count / len(recording_ids) * 100):.2f}%") logger.info("=" * 60) diff --git a/fetch_s3_recording_by_room.py b/fetch_s3_recording_by_room.py new file mode 100755 index 000000000..5c2cefb60 --- /dev/null +++ b/fetch_s3_recording_by_room.py @@ -0,0 +1,238 @@ +#!/usr/bin/env -S uv run +"""Utilities for fetching Daily.co recording URLs with retry logic. + +This module provides functions to retrieve recording download links from Daily's REST API +with robust error handling, rate limiting, and exponential backoff retry logic. +""" + +import asyncio +import os +from typing import Optional, Tuple + +from httpx import AsyncClient, HTTPStatusError +from loguru import logger +from tenacity import ( + AsyncRetrying, + RetryError, + retry_if_exception_type, + stop_after_attempt, + wait_exponential, +) + + +async def get_recording_s3_url_with_retry( + room_id: str, + max_retries: int = 5, +) -> Tuple[Optional[str], Optional[str]]: + """Retrieve recording URL with exponential backoff and retry logic. + + Uses tenacity library to handle rate limiting (429 errors) and other + transient errors with automatic exponential backoff. + + Args: + room_id: Daily.co room identifier + max_retries: Maximum number of retry attempts (default 5) + + Returns: + Tuple of (recording_url, recording_signed_url) + Returns (None, None) if no recording exists for the room. + """ + try: + # Use tenacity's AsyncRetrying for automatic retry with exponential backoff + async for attempt in AsyncRetrying( + retry=retry_if_exception_type((HTTPStatusError, Exception)), + stop=stop_after_attempt(max_retries), + wait=wait_exponential(multiplier=1, min=1, max=60), + reraise=True, + ): + with attempt: + recording_url, recording_signed_url, status = await get_recording_s3_url( + room_id=room_id + ) + + # If no recording exists (status is None), return immediately - no retry + if status is None: + logger.debug(f"No recording found for room {room_id}") + return None, None + + # If recording exists but is not finished yet, retry + if status != "finished": + logger.warning( + f"Recording not finished for room {room_id}, status: {status} " + f"(attempt {attempt.retry_state.attempt_number}/{max_retries})" + ) + raise Exception(f"Recording not ready, status: {status}") + + # Recording is finished, return the URLs + return recording_url, recording_signed_url + + # This line should never be reached due to reraise=True, but satisfies type checker + return None, None + + except RetryError as e: + # All retries exhausted + last_exception = e.last_attempt.exception() + logger.error( + f"Failed to retrieve recording URL for room {room_id} after {max_retries} attempts: " + f"{last_exception}" + ) + return None, None + + except Exception as e: + logger.exception(f"Unexpected error retrieving recording for room {room_id}: {e}") + return None, None + + +async def get_recording_s3_url( + room_id: str, +) -> Tuple[Optional[str], Optional[str], Optional[str]]: + """Get recording URL using Daily's REST API. + + Args: + room_id: Daily.co room identifier + + Returns: + Tuple of (recording_url, recording_signed_url, status) + - recording_url: The download link for the recording + - recording_signed_url: Same as recording_url (kept for backward compatibility) + - status: Recording status from Daily API + + Raises: + HTTPStatusError: When HTTP errors occur (including rate limits) + """ + headers = { + "Authorization": f"Bearer {os.getenv('DAILY_API_KEY')}", + "Content-Type": "application/json", + } + + async with AsyncClient(timeout=180) as client: + # List recordings for the room + list_response = await client.get( + url=f"https://api.daily.co/v1/recordings?room_name={room_id}", + headers=headers, + ) + list_response.raise_for_status() + list_data = list_response.json() + + # Check if recording exists and is finished + if not list_data.get("data") or len(list_data["data"]) == 0: + return (None, None, None) + + recording_id = list_data["data"][0].get("id") + status = list_data["data"][0].get("status") + + if not recording_id or status != "finished": + return (None, None, status) + + # Get the recording access link + link_response = await client.get( + url=f"https://api.daily.co/v1/recordings/{recording_id}/access-link", + headers=headers, + ) + link_response.raise_for_status() + link_data = link_response.json() + + recording_url = link_data.get("download_link") + if not recording_url: + logger.warning(f"No download link found for recording {recording_id}") + return (None, None, status) + + # Return the same URL for both fields for backward compatibility + return (recording_url, recording_url, status) + + +async def get_recent_rooms(limit: int = 100) -> list[str]: + """Get list of recent room names from Daily API. + + Args: + limit: Maximum number of rooms to retrieve (default 100, max 100) + + Returns: + List of room names (strings) + """ + try: + async with AsyncClient() as client: + response = await client.get( + url=f"https://api.daily.co/v1/rooms?limit={min(limit, 100)}", + headers={ + "Authorization": f"Bearer {os.getenv('DAILY_API_KEY')}", + "Content-Type": "application/json", + }, + timeout=30, + ) + response.raise_for_status() + + data = response.json() + rooms = data.get("data", []) + room_names = [room.get("name") for room in rooms if room.get("name")] + + logger.info(f"Retrieved {len(room_names)} room names from Daily API") + return room_names + + except Exception as e: + logger.exception(f"Failed to get rooms from Daily API: {e}") + return [] + + +async def main(): + """Test get_recording_s3_url_with_retry with recent rooms.""" + logger.info("Starting recording fetch test...") + + # Step 1: Get the most recent 100 rooms + logger.info("Fetching recent rooms...") + room_names = await get_recent_rooms(limit=100) + + if not room_names: + logger.error("No rooms found. Cannot proceed with test.") + return + + logger.info(f"Found {len(room_names)} rooms to check for recordings") + + # Call get_recording_s3_url_with_retry on each room concurrently + logger.info(f"Attempting to fetch recordings for {len(room_names)} rooms concurrently...") + + # Create tasks for all rooms + tasks = [ + get_recording_s3_url_with_retry(room_id=room_name, max_retries=3) + for room_name in room_names + ] + + # Execute all tasks concurrently + results = await asyncio.gather(*tasks, return_exceptions=True) + + # Process results + success_count = 0 + not_found_count = 0 + failed_count = 0 + + for i, (room_name, result) in enumerate(zip(room_names, results), 1): + if isinstance(result, Exception): + failed_count += 1 + logger.error(f"❌ [{i}/{len(room_names)}] Failed for {room_name}: {result}") + elif isinstance(result, tuple) and len(result) == 2: + recording_url, recording_signed_url = result + if recording_url: + success_count += 1 + logger.info(f"✅ [{i}/{len(room_names)}] Found recording for {room_name}") + logger.debug(f" URL: {recording_url[:80]}...") + else: + not_found_count += 1 + logger.debug(f"ℹ️ [{i}/{len(room_names)}] No recording for {room_name}") + else: + failed_count += 1 + logger.error(f"❌ [{i}/{len(room_names)}] Unexpected result type for {room_name}") + + # Summary + logger.info("\n" + "=" * 60) + logger.info("RECORDING FETCH TEST SUMMARY") + logger.info("=" * 60) + logger.info(f"Total rooms checked: {len(room_names)}") + logger.info(f"✅ Recordings found: {success_count}") + logger.info(f"ℹ️ No recordings: {not_found_count}") + logger.info(f"❌ Failed: {failed_count}") + logger.info(f"Success rate: {(success_count / len(room_names) * 100):.2f}%") + logger.info("=" * 60) + + +if __name__ == "__main__": + asyncio.run(main())