Spaces:
Build error
Build error
| # tvdb.py | |
| import os | |
| import requests | |
| import urllib.parse | |
| from datetime import datetime, timedelta | |
| from dotenv import load_dotenv | |
| import json | |
| from hf_scrapper import get_system_proxies | |
| load_dotenv() | |
| THETVDB_API_KEY = os.getenv("THETVDB_API_KEY") | |
| THETVDB_API_URL = os.getenv("THETVDB_API_URL") | |
| CACHE_DIR = os.getenv("CACHE_DIR") | |
| TOKEN_EXPIRY = None | |
| THETVDB_TOKEN = None | |
| proxies = get_system_proxies() | |
| def authenticate_thetvdb(): | |
| global THETVDB_TOKEN, TOKEN_EXPIRY | |
| auth_url = f"{THETVDB_API_URL}/login" | |
| auth_data = { | |
| "apikey": THETVDB_API_KEY | |
| } | |
| try: | |
| response = requests.post(auth_url, json=auth_data, proxies=proxies) | |
| response.raise_for_status() | |
| response_data = response.json() | |
| THETVDB_TOKEN = response_data['data']['token'] | |
| TOKEN_EXPIRY = datetime.now() + timedelta(days=30) | |
| except requests.RequestException as e: | |
| print(f"Authentication failed: {e}") | |
| THETVDB_TOKEN = None | |
| TOKEN_EXPIRY = None | |
| def get_thetvdb_token(): | |
| global THETVDB_TOKEN, TOKEN_EXPIRY | |
| if not THETVDB_TOKEN or datetime.now() >= TOKEN_EXPIRY: | |
| authenticate_thetvdb() | |
| return THETVDB_TOKEN | |
| def fetch_and_cache_json(original_title, title, media_type, year=None): | |
| if year: | |
| search_url = f"{THETVDB_API_URL}/search?query={urllib.parse.quote(title)}&type={media_type}&year={year}" | |
| else: | |
| search_url = f"{THETVDB_API_URL}/search?query={urllib.parse.quote(title)}&type={media_type}" | |
| token = get_thetvdb_token() | |
| if not token: | |
| print("Authentication failed") | |
| return | |
| headers = { | |
| "Authorization": f"Bearer {token}", | |
| "accept": "application/json", | |
| } | |
| try: | |
| # Fetch initial search results | |
| response = requests.get(search_url, headers=headers, proxies=proxies) | |
| response.raise_for_status() | |
| data = response.json() | |
| if 'data' in data and data['data']: | |
| # Extract the TVDB ID and type from the first result | |
| first_result = data['data'][0] | |
| tvdb_id = first_result.get('tvdb_id') | |
| media_type = first_result.get('type') | |
| if not tvdb_id: | |
| print("TVDB ID not found in the search results") | |
| return | |
| # Determine the correct extended URL based on media type | |
| if media_type == 'movie': | |
| extended_url = f"{THETVDB_API_URL}/movies/{tvdb_id}/extended?meta=translations" | |
| elif media_type == 'series': | |
| extended_url = f"{THETVDB_API_URL}/series/{tvdb_id}/extended?meta=translations" | |
| else: | |
| print(f"Unsupported media type: {media_type}") | |
| return | |
| # Request the extended information using the TVDB ID | |
| response = requests.get(extended_url, headers=headers, proxies=proxies) | |
| response.raise_for_status() | |
| extended_data = response.json() | |
| # Cache the extended JSON response | |
| json_cache_path = os.path.join(CACHE_DIR, f"{urllib.parse.quote(original_title)}.json") | |
| with open(json_cache_path, 'w') as f: | |
| json.dump(extended_data, f) | |
| except requests.RequestException as e: | |
| print(f"Error fetching data: {e}") | |