Spaces:
Sleeping
Sleeping
| import requests | |
| import time | |
| import logging | |
| from typing import List, Optional, Dict | |
| import os | |
| logger = logging.getLogger(__name__) | |
| class TMDBService: | |
| """Service pour interagir avec l'API TMDB""" | |
| def __init__(self): | |
| self.api_key = os.getenv('TMDB_API_KEY') | |
| self.base_url = "https://api.themoviedb.org/3" | |
| def _make_request(self, endpoint: str, params: dict = None, max_retries: int = 3) -> Optional[dict]: | |
| """Make API request with retry and backoff""" | |
| if params is None: | |
| params = {} | |
| params['api_key'] = self.api_key | |
| for attempt in range(max_retries): | |
| try: | |
| response = requests.get(f"{self.base_url}{endpoint}", params=params, timeout=10) | |
| if response.status_code == 200: | |
| return response.json() | |
| elif response.status_code == 429: | |
| # Rate limited, wait longer | |
| wait_time = 2 ** attempt | |
| logger.warning(f"Rate limited, waiting {wait_time}s before retry {attempt + 1}") | |
| time.sleep(wait_time) | |
| continue | |
| else: | |
| logger.error(f"TMDB API returned status {response.status_code}") | |
| return None | |
| except Exception as e: | |
| logger.error(f"Request failed (attempt {attempt + 1}): {e}") | |
| if attempt < max_retries - 1: | |
| time.sleep(2 ** attempt) | |
| return None | |
| async def get_popular_movies(self, limit: int = 10000) -> List[Dict]: | |
| """Get popular movies from TMDB""" | |
| movies = [] | |
| page = 1 | |
| filter_adult = os.getenv('FILTER_ADULT_CONTENT', 'true').lower() == 'true' | |
| while len(movies) < limit: | |
| logger.info(f"Fetching popular movies page {page}") | |
| data = self._make_request("/movie/popular", {"page": page}) | |
| if not data or not data.get('results'): | |
| break | |
| for movie in data.get('results', []): | |
| if len(movies) >= limit: | |
| break | |
| # Skip adult movies if filtering is enabled | |
| if filter_adult and movie.get('adult', False): | |
| continue | |
| # Get detailed movie data | |
| movie_details = self.get_movie_details(movie['id']) | |
| if movie_details: | |
| # Get credits | |
| credits = self.get_movie_credits(movie['id']) | |
| if credits: | |
| movie_details['credits'] = credits | |
| movies.append(movie_details) | |
| # Check if we've reached the last page | |
| if page >= data.get('total_pages', 0): | |
| break | |
| page += 1 | |
| time.sleep(0.25) # Rate limiting | |
| logger.info(f"Collected {len(movies)} movies from TMDB") | |
| return movies[:limit] | |
| def get_movie_details(self, movie_id: int) -> Optional[dict]: | |
| """Get detailed movie information""" | |
| return self._make_request(f"/movie/{movie_id}") | |
| def get_movie_credits(self, movie_id: int) -> Optional[dict]: | |
| """Get movie cast and crew""" | |
| return self._make_request(f"/movie/{movie_id}/credits") |