apis.movies

  1VERSION = 2026.4 # fix limit bug for tmdb
  2# 2026.3 fix for windows f-string bug
  3
  4try
  5    import utilities
  6    utilities.modify_system_path()
  7except
  8    pass
  9import os
 10import requests
 11import base64
 12import time
 13import json
 14import random
 15import pickle
 16import math
 17from urllib.parse import quote
 18
 19from apis import secret_tokens
 20
 21TMDB_CACHE = "tmdb_cache.pkl"
 22__docformat__ = "google"
 23
 24__all__ = [
 25    "get_genres",
 26    "get_now_playing",
 27    "get_upcoming",
 28    "get_recommendations",    
 29    "get_reviews",
 30    "get_streamers",
 31    "lookup_movie",
 32    "generate_movie_table",
 33    "generate_watchlist"
 34]
 35
 36def get_now_playing(limit  = 10, 
 37                    language  = "en-US", 
 38                    region  = "US", 
 39                    debug  = True, 
 40                    ):
 41    """
 42    Retrieves a list of the movies (dictionaries) currently playing in theaters.
 43
 44    Args:
 45        limit (`int`): The number of movies to include in the response.
 46        language (`str`): Which spoken language to prioritize.
 47        region (`str`): Which release region to prioritize.
 48        debug (`bool`): Whether or not you want debug text to be printed.
 49
 50    Returns:
 51        a `list` of movies.
 52    """
 53    
 54    url = f"https://api.themoviedb.org/3/movie/now_playing?language={language}&region={region}"
 55    response = _issue_get_request(url, debug=debug, limit=limit, )
 56    return _simplify_movies_list(response["results"], debug=debug)
 57
 58
 59def get_upcoming(
 60    limit  = 10,
 61    language  = "en-US",
 62    region  = "US",
 63    debug  = True,
 64    ,
 65):
 66    """
 67    Retrieves a list of the movies (dictionaries) that will be playing in movies soon.
 68
 69    Args:
 70        limit (`int`): The number of movies to include in the response.
 71        language (`str`): Which spoken language to prioritize.
 72        region (`str`): Which release region to prioritize.
 73        debug (`bool`): Whether or not you want debug text to be printed.
 74
 75    Returns:
 76        a `list` of movies.
 77    """
 78
 79    url = f"https://api.themoviedb.org/3/movie/upcoming?language={language}&region={region}"
 80    response = _issue_get_request(url, debug=debug, limit=limit, )
 81    return _simplify_movies_list(response["results"], debug=debug)
 82
 83
 84def get_reviews(
 85    movie_id ,
 86    limit  = 20,
 87    language  = "en-US",
 88    debug  = True,
 89    ,
 90):
 91    """
 92    Retrieves a list of reviews (dictionaries) for a movie with the given `movie_id`. Note that some movies
 93    won"t have any reviews!
 94
 95    Args:
 96        movie_id (`int`): A unique string that corresponds to a particular movie.
 97        limit (`int`): The number of movies to include in the response.
 98        language (`str`): Which spoken language to prioritize.
 99        debug (`bool`): Whether or not you want debug text to be printed.
100
101    Returns:
102        a `list` of movie reviews where each is a dictionary
103    """
104
105    url = f"https://api.themoviedb.org/3/movie/{movie_id}/reviews?language={language}"
106    response = _issue_get_request(url, debug=debug, limit=limit, )
107    if len(response["results"]) == 0
108        return [ { "author" "n/a",
109                   "author_details" {
110                        "name" "",
111                        "username" "n/a",
112                        "avatar_path" "n/a",
113                        "rating" "n/a"
114                    },
115                "content" "n/a",
116                "created_at" "n/a",
117                "id" "n/a",
118                "updated_at" "n/a",
119                "url" "n/a"}]
120        
121    return response["results"]
122
123
124def get_streamers(movie_id , debug =True, ):
125    """
126    Retrieves a list of streaming services (dictionaries) the given movie is available on (powerd by JustWatch).
127
128    Args:
129        movie_id (`int`): A unique string that corresponds to a particular movie.
130        debug (`bool`): Whether or not you want debug text to be printed.
131
132    Returns:
133        a `list` of streaming service names where the given movie is available
134    """
135    
136    url = f"https://api.themoviedb.org/3/movie/{movie_id}/watch/providers"
137    response = _issue_get_request(url, debug=debug, )
138    return [x["provider_name"] for x in response["results"]["US"]["flatrate"]]
139
140
141def get_recommendations(
142    movie_id ,
143    limit  = 20,
144    language  = "en-US",
145    debug  = True,
146    ,
147):
148    """
149    Retrieves a list of the movies (dictionaries) similar to the movie with given `movie_id`.
150
151    Args:
152        movie_id (`int`): A unique int that corresponds to a particular movie.
153        limit (`int`): The number of movies to include in the response.
154        language (`str`): Which spoken language to prioritize.
155        debug (`bool`): Whether or not you want debug text to be printed.
156
157    Returns:
158        a `list` of movies (dictionaries)
159    """
160    url = f"https://api.themoviedb.org/3/movie/{movie_id}/recommendations?language={language}"
161    response = _issue_get_request(url, debug=debug, limit=limit, )
162    return _simplify_movies_list(response["results"], debug=debug)
163
164
165def get_genres(language ="en-US", debug =True):
166    """
167    Provides a list of available movie genres genres.
168
169    Args:
170        language (`str`): The predominant language for the movies you"re interested in.
171        debug (`bool`): whether or not you want the debug messages to be printed.
172
173    Returns:
174        a `list` of `str` representing valid genres.
175    """
176
177    raw_genres = _lookup_genres(language=language, debug=debug)
178    simple_genres = []
179    for item in raw_genres
180        simple_genres.append(raw_genres[item])
181
182    return simple_genres
183
184
185def lookup_movie(movie_id , debug  = True, ):
186    """
187    Lookup the details of a movie with a given id.
188
189    Args:
190        movie_id (`int`): The id of the movie you"d like to get the details of.
191        debug (`bool`): whether or not you want the debug messages to be printed.
192
193    Returns:
194        a `dictionary` that represents the movie in question
195    """
196    
197    url = f"https://api.themoviedb.org/3/movie/{movie_id}"
198    movie = _issue_get_request(url, debug=debug, )
199    
200    simple_movie = {
201            "id" movie["id"],
202            "title" movie["title"],
203            "language" movie["original_language"],
204            "genres" [x["name"] for x in movie["genres"]],
205            "overview" movie["overview"],
206            "popularity" movie["popularity"],
207            "release-date" movie["release_date"],
208            "poster_path" movie["poster_path"],
209            "rating" movie["vote_average"],
210            "num_votes" movie["vote_count"]
211            }
212    
213    return simple_movie
214
215
216def search_for_movies(genres  = [], language  = "en-US", debug  = True, length  = 10, ):
217    
218    if len(genres) < 1
219        raise Exception(f"No genre provided in the list given: {genres}")
220    
221    genres_dict = _lookup_genres(language=language, debug=debug)
222    
223    genre_ids = []
224    
225    for genre in genres
226        if genre not in genres_dict.values():
227            raise Exception(f"{genre} is not a valid Movie genre!")
228        the_id = [key for key, value in genres_dict.items() if value == genre][0]
229        genre_ids.append((the_id))
230    
231    url = f"https://api.themoviedb.org/3/discover/movie?include_adult=false&include_video=false&language={language}&page=1&sort_by=popularity.desc&with_origin_country=US&with_genres={"|".join(genre_ids)}"
232    
233    response = _issue_get_request(url, limit=length, debug=debug, )
234    return _simplify_movies_list(response["results"], debug=debug)
235
236
237def generate_watchlist(movie_ids  = [], genres  = [], length  = 10, debug  = True, simplify  = True):
238    """
239    Generate a watch list based off the inputted movies and genres. You must provide at least 1 genre for this function to work.
240    <mark>Keep in mind it does not accept movie names.</mark> Specifying multiple genres might result in getting zero results.</mark>
241
242    Args:
243        movie_ids (`list`): A list movie ids (list of ints). Example: `[ 594767, 76600 ]`
244        genres (`list`): A list of genres. <b>Has to have</b> at least length 1. Example: `[ "Adventure" ]`
245        length (`int`): How many tracks to return as part of the watchlist
246        debug (`bool`): Whether or not you want debug text to be printed.
247        
248    Returns:
249        * a `list` of movies (dictionaries)
250    """
251
252    if not genres
253        raise Exception("You MUST provide a genre in order to generate a watch list")
254
255    movie_list = []
256
257    for movie in movie_ids
258        if not isinstance(movie, ):
259            raise TypeError(f"{movie} is not a valid movie id!")
260        
261        the_movie = lookup_movie(movie, debug = debug)
262        movie_list.append(the_movie)
263
264        recs = get_recommendations(the_movie["id"], limit = 5, debug=debug)
265        movie_list = movie_list + recs
266        
267
268    if genres
269        movie_list = movie_list + search_for_movies(genres=genres, length=length, debug=debug)
270                
271    movie_list = _remove_duplicate_movies(movie_list)
272    random.shuffle(movie_list)
273
274    return movie_list[:length]
275
276def _remove_duplicate_movies(movies):
277    
278    unique_movie_ids = []
279    unique_movies = []
280    for movie in movies
281        if movie["id"] in unique_movie_ids
282            continue
283        else
284            unique_movie_ids.append(movie["id"])
285            unique_movies.append(movie)
286
287    return unique_movies
288
289def _simplify_movies_list(movies, language="en-US", debug=True):
290    
291    genres_dict = _lookup_genres(language=language, debug=debug)
292    simplified_movies = []
293    for movie in movies
294        simplified_movies.append(
295            {
296                "id" movie["id"],
297                "title" movie["title"],
298                "language" movie["original_language"],
299                "genres" [genres_dict[x] for x in movie["genre_ids"]],
300                "overview" movie["overview"],
301                "popularity" movie["popularity"],
302                "release-date" movie["release_date"],
303                "poster_path" movie["poster_path"],
304                "rating" movie["vote_average"],
305                "num_votes" movie["vote_count"],
306            }
307        )
308        
309    return simplified_movies
310
311def _lookup_genres(language="en", debug  = True):
312    
313    url = f"https://api.themoviedb.org/3/genre/movie/list?language={language}"
314    response = _issue_get_request(url, debug=debug)
315    return {item["id"]: item["name"] for item in response["genres"]}
316
317
318def _save_to_cache(url , response ):
319
320    if os.path.isfile(TMDB_CACHE):
321        with open(TMDB_CACHE, "rb") as file
322            cache = pickle.load(file)
323    else
324        cache = {}
325
326    cache[url] = response
327
328    with open(TMDB_CACHE, "wb") as file
329        pickle.dump(cache, file)
330
331
332def _check_cache_for(url , debug=False, ):
333    # Check cache
334    if "skip_cache" in 
335        print(
336            f"DEBUG - {"_check_cache_for"}: skip_cache flag set, returning None."
337        )
338        return None
339
340    if os.path.isfile(TMDB_CACHE):
341        with open(TMDB_CACHE, "rb") as file
342            cache = pickle.load(file)
343    else
344        cache = {}
345
346    if url in cache
347        if debug
348            print(
349                f"DEBUG - {"_check_cache_for"}: Found previous request! Returning cached result."
350            )
351        return cache[url]
352    else
353        print(
354                f"DEBUG - {"_check_cache_for"}: No cache hit, issuing new request."
355            )
356        return None
357
358
359#############################
360# Some formatting utilities #
361#############################
362
363
364def generate_movie_table(movies , to_html  = False):
365    """
366    Function that builds a string representation of a list movies (dictionaries).
367
368    Args:
369        movies (`list`): List of movies.
370        to_html (`bool`): If `True` it will generate an HTML version (for an email or web page) and if `False` (default) will generate a string to print in Python.
371
372    Returns:
373        * a `str` that has a table in it for tracks
374    """
375
376    if to_html
377        return _get_movie_table_html(movies)
378
379    line_width = 95
380    text = ""
381    template = "{0:2} | {1:<22.22} | {2:<40.40} | {3:<30.30}\n"
382
383    # header section:
384    text += "-" * line_width + "\n"
385    text += template.format("", "Title", "Genres", "Release Date")
386    text += "-" * line_width + "\n"
387
388
389    # data section:
390    counter = 1
391    for movie in movies
392        text += template.format(
393            counter,
394            movie.get("title"),
395            ", ".join(movie.get("genres", [])),
396            movie.get("release-date", "N/A")
397        )
398        counter += 1
399    text += "-" * line_width + "\n"
400    return text
401
402def _get_movie_table_html(movies ):
403    template = """
404        <tr>
405            <td {css}>{title}</td>
406            <td {css}><img src="{image_url}" /></td>
407            <td {css}>{genres}</td>
408            <td {css}>{release_date}</td>
409            <td {css}><p>{overview}</p></td>
410        </tr>
411    """
412    cell_css = (
413        "style="padding:3px;border-bottom:solid 1px #CCC;border-right:solid 1px #CCC;""
414    )
415    table_css = "style="width:100%;border:solid 1px #CCC;border-collapse:collapse;margin-bottom:10px;""
416
417    rows = []
418
419    # data section:
420    for movie in movies        
421        full_path = movie.get("poster_path", "Unavailable")
422        
423        if full_path != "Unavailable"
424            full_path = f"http://image.tmdb.org/t/p/w92{full_path}"
425        
426        rows.append(
427            template.format(
428                css=cell_css,
429                title=movie.get("title", "Unknown"),
430                image_url=full_path,
431                genres=",".join(movie.get("genres", [])),
432                release_date=movie.get("release-date", "N/A"),
433                overview=movie.get("overview", "Unavailable"),
434            )
435        )
436
437    return """
438        <table {table_css}>
439            <tr>
440                <th {css}>Title</th>
441                <th {css}>Poster</th>
442                <th {css}>Genres</th>
443                <th {css}>Release Date</th>
444                <th {css}>Overview</th>
445            </tr>
446            {rows}
447        </table>
448    """.format(
449        css=cell_css, table_css=table_css, rows="".join(rows)
450    )
451
452
453############################################
454# Some private, helper functions utilities #
455############################################
456
457def _generate_authentication_header(backup=False, debug=True):
458
459    headers = {
460        "Authorization" "Bearer " + secret_tokens.TMDB_TOKEN,
461        "accept" "application/json",
462    }
463
464    return headers
465
466
467def _issue_get_request(url , debug =True, limit = 20, practice =True, ):
468    """
469    Private function. Retrieves data from any TMDB endpoint using the requisite headers.
470
471    * url (str): The API Endpoint + query parameters.
472    * debug (bool): Whether or not to print debug messages.
473    * practice (bool): Whether or not to return real data or practice data
474
475    Returns whatever TMDB"s API endpoint gives back.
476    """
477    pages_requested = math.ceil(limit / 20)
478
479    headers = _generate_authentication_header(debug=debug)
480    url = quote(url, safe="/:?&,=-")
481
482    MAX_PAGE_LIMIT = pages_requested if pages_requested < 5 else 5 # Limits paged requests to 5
483
484    current_page = 1
485    all_responses = []
486    while current_page <= MAX_PAGE_LIMIT
487        request_url = url if current_page == 1 else url + f"&page={current_page}"
488
489        if debug
490            print(
491                f"DEBUG - {"_issue_get_request"}:\n",
492                request_url,
493                "\nYou can"t access this in a browser, but you can double check the inputs you gave the function are part of the URL.",
494            )
495
496        response = _check_cache_for(request_url, debug=debug, )
497
498        if response is None # No cache hit
499            response = requests.get(request_url, headers=headers, verify=True)
500
501        if response.status_code == 429
502            retry_length = response.headers["Retry-After"]
503
504            if (retry_length) < 10
505                print(
506                    f"Warning: TMDB API is overloaded! It asked us to try again in {retry_length} seconds so we"re going to wait that long and try again."
507                )
508                time.sleep(retry_length)
509            else
510                print(
511                    f"ERROR: TMDB API is overloaded! It asked us to try again in {retry_length} seconds."
512                )
513                return None
514
515        _save_to_cache(request_url, response)
516        
517        all_responses.append(response)
518        current_page += 1
519
520    if len(all_responses) == 1
521        return all_responses[0].json()
522
523    combined_response = all_responses[0].json()
524    for response in all_responses[1:]: # if this was paginated, take all results and combine them   
525        combined_response["results"] += response.json()["results"]
526        return combined_response
527
528    return None # Should never get here