Spaces:
Sleeping
Sleeping
File size: 3,418 Bytes
14e32e0 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 |
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") |