# tvdb.py import os import requests import urllib.parse from datetime import datetime, timedelta from dotenv import load_dotenv import json import asyncio import aiofiles from tvdbApiClient import fetch_and_cache_seasons, save_to_json from services import RecentList, GenreList 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 recent_list = RecentList() genre_list = GenreList() def authenticate_thetvdb(): global THETVDB_TOKEN, TOKEN_EXPIRY auth_url = f"{THETVDB_API_URL}/login" auth_data = { "apikey": THETVDB_API_KEY } try: print("Authenticating with TheTVDB API...") response = requests.post(auth_url, json=auth_data) response.raise_for_status() response_data = response.json() THETVDB_TOKEN = response_data['data']['token'] TOKEN_EXPIRY = datetime.now() + timedelta(days=30) print("Authentication successful.") 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 clean_data(data): print("Cleaning data...") fields_to_keep = { "data": { 'id': None, 'name': None, 'image': None, 'score': None, 'runtime': None, 'releases': None, 'year': None, 'contentRatings': None, 'originalCountry': None, 'originalLanguage': None, 'translations': {}, 'artworks': [], 'genres':[], 'characters': [], 'spoken_languages': [], 'translations': {}, 'trailers': [], } } cleaned_data = {} for key, value in fields_to_keep.items(): if key in data: cleaned_data[key] = {} for field in fields_to_keep[key]: if field in data[key]: cleaned_data[key][field] = data[key][field] print("Data cleaned successfully.") return cleaned_data async def fetch_and_cache_json(original_title, title, media_type, year=None): print(f"Fetching data for: {original_title}") 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 token not available.") return headers = { "Authorization": f"Bearer {token}", "accept": "application/json", } try: print(f"Sending search request to: {search_url}") response = requests.get(search_url, headers=headers) print(f"Search response status code: {response.status_code}") response.raise_for_status() data = response.json() if 'data' in data and data['data']: first_result = data['data'][0] tvdb_id = first_result.get('tvdb_id') media_type = first_result.get('type') print(f"Found TVDB ID: {tvdb_id} with media type: {media_type}") if not tvdb_id: print("TVDB ID not found in the search results") return 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" await fetch_and_cache_seasons(tvdb_id, original_title) else: print(f"Unsupported media type: {media_type}") return response = requests.get(extended_url, headers=headers) print(f"Extended data response status code: {response.status_code}") response.raise_for_status() extended_data = response.json() cleaned_data = clean_data(extended_data) print(f"cleaning.. {original_title}") genres = None if cleaned_data['data'].get('genres'): genres = cleaned_data['data'].get('genres') print(f"genres extracted: {genres}") description = None if cleaned_data['data'].get('translations') and cleaned_data['data']['translations'].get('overviewTranslations'): for overview in cleaned_data['data']['translations']['overviewTranslations']: if overview['language'] == 'eng': description = overview.get('overview') break print(f"Description extracted: {description}") image_link = None if cleaned_data['data'].get('artworks'): for artwork in cleaned_data['data']['artworks']: if artwork['type'] in [15, 3]: image_link = artwork.get('thumbnail') break print(f"Image link extracted: {image_link}") if media_type == 'movie': recent_list.add_entry(original_title, cleaned_data['data']['year'], description, image_link, genres,'film') genre_list.add_entry(genres, original_title, cleaned_data['data']['year'], description, image_link, 'movie') elif media_type == 'series': recent_list.add_entry(original_title, cleaned_data['data']['year'], description, image_link, genres, 'series') genre_list.add_entry(genres, original_title, cleaned_data['data']['year'], description, image_link, 'series') print(f"adding.. {original_title}") # Create the full directory path if it doesn't exist full_dir_path = os.path.join(CACHE_DIR, media_type) os.makedirs(full_dir_path, exist_ok=True) # Now create the JSON cache path json_cache_path = os.path.join(full_dir_path, f"{urllib.parse.quote(original_title)}.json") await save_to_json(cleaned_data, json_cache_path) print(f"Data saved to JSON at: {json_cache_path}") else: print(f"No data found for {original_title} in search results.") except requests.RequestException as e: print(f"Error fetching data: {e}") except Exception as e: print(f"An unexpected error occurred: {e}") def main(): entries = [ {"original_title": "Funky Monkey (2004)", "title": "Funky Monkey", "media_type": "movie", "year": 2004}, {"original_title": "My Spy (2020)", "title": "My Spy", "media_type": "movie", "year": 2020}, {"original_title": "My Spy (2024)", "title": "My Spy", "media_type": "movie", "year": 2024}, {"original_title": "Yaariayan (2014)", "title": "Yaariayan", "media_type": "movie", "year": 2014}, {"original_title": "Yaariyan 2 (2023)", "title": "Yaariyan 2", "media_type": "movie", "year": 2023}, ] asyncio.run(process_entries(entries)) async def process_entries(entries): for entry in entries: await fetch_and_cache_json(entry['original_title'], entry['title'], entry['media_type'], entry['year']) if __name__ == "__main__": main()