import cgi from contextlib import closing from datetime import date from enum import Enum import json import re import ssl from urllib.error import URLError, HTTPError import urllib.parse import urllib.request from flask import Blueprint, current_app, flash, request, render_template from .. import util #################### # constants bp = Blueprint('tmdb', __name__, url_prefix='/tmdb') TMDB_API_URL = 'https://api.themoviedb.org' TMDB_SEARCH_MOVIE = '/3/search/movie' TMDB_SEARCH_TV = '/3/search/tv' TMDB_MOVIE_LINK_URL = 'https://www.themoviedb.org/movie/' TMDB_TV_LINK_URL = 'https://www.themoviedb.org/tv/' #################### # data structures class query_type(Enum): UNKNOWN = 0 MOVIE = 1 TV_SHOW = 2 class query_data(object): def __init__(self): self.errors = [] self.full_url = '' self.movie_year = 0 self.parsed_query_str = '' self.query_params = {} self.query_type = query_type.UNKNOWN self.raw_query_str = '' self.tmdb_api_key = '' class response_data(object): def __init__(self): self.errors = [] self.parsed_response = {} self.response_body = '' self.response_headers = [] self.response_status = 0 self.result_count = 0 class view_result(object): def __init__(self): self.id = 0 self.title = '' self.date = 0 self.description = '' self.genre_ids = [] self.original_language = '' self.poster_path = 'no_image.png' self.tmdb_link = '' # NOTE: result is parsed JSON object from tmdb API def copy_result(self, result, query_type): self.id = result['id'] self.title = result['original_title'] self.release_date = result['release_date'] self.description = result['overview'] self.genre_ids = result['genre_ids'] self.original_language = result['original_language'] self.poster_path = result['poster_path'] if query_type == query_type.MOVIE: self.tmdb_link = TMDB_MOVIE_LINK_URL + str(result['id']) elif query_type == query_type.TV_SHOW: self.tmdb_link = TMDB_TV_LINK_URL + str(result['id']) return self #################### # flask routes @bp.route('/query_movie') def query_movie(): errors = [] query = query_data() response = response_data() view_results = [] query.tmdb_api_key = current_app.config['TMDB_API_KEY'] # NOTE: query string can be submitted from a link or form input if 'query' in request.form: query.raw_query_str = request.form['query'] elif 'query' in request.args: query.raw_query_str = request.args['query'] if query.raw_query_str and parseQueryStr(query): if makeQuery(query, response): try: response.parsed_response = json.loads(response.response_body) response.result_count = response.parsed_response['total_results'] except json.JSONDecodeError as e: view_data.errors.append(e) for result in response.parsed_response['results']: copy = view_result() view_results.append(copy.copy_result(result, query.query_type)) else: query.errors.append('{}, error making query'.format(util.getFunctionName)) else: errors.append('{}, error in query string'.format(util.getFunctionName())) errors.extend(query.errors) errors.extend(response.errors) if errors: flash(errors) return render_template( 'tmdb/query.html', view_data = view_results, debug_str = str(util.dumpObjAttr(query) + '\n' + util.dumpObjAttr(response)) + '\n' + util.dumpList(view_results, 0) + '\n' ) ################### # helpers # TODO: need a way to correctly parse titles like: "2001: A Space Odyssey (1968)" # with legitimate dates in the title def parseQueryStr(query_obj): s = query_obj.raw_query_str.replace('.', ' ').strip() current_year = date.today().year # NOTE: find all 4 digit numbers to match the year of queries with 4 digit numbers in the title # ex) "2036.Origin.Unknown.2018" regex = re.compile('[0-9]{4}') for m in regex.finditer(s): i = int(m[0]) if i > 1878 and i <= current_year: query_obj.parsed_query_str = s[0:m.start()].strip() # NOTE: slice query string to year match query_obj.movie_year = i query_obj.query_type = query_type.MOVIE break # TODO: parse tv show titles if query_obj.query_type == query_type.UNKNOWN: query_obj.errors.append('%s: unable to determine query type' % util.getFunctionName()) return False return True def makeQuery(query_obj, response_obj): if query_obj.query_type == query_type.MOVIE: search_part = TMDB_SEARCH_MOVIE elif query_obj.query_type == query_type.TV_SHOW: search_part = TMDB_SEARCH_TV else: query_obj.errors.append('%s(), query type is not set' % util.getFunctionName()) return False request_headers = {"Accept": "text/html,application/xhtml+xml,application/xml"} query_obj.query_params = urllib.parse.urlencode( { 'api_key': query_obj.tmdb_api_key, 'query': query_obj.parsed_query_str, 'year': query_obj.movie_year, } ) query_obj.full_url = TMDB_API_URL + search_part + '?' + query_obj.query_params context = ssl.create_default_context() req = urllib.request.Request(query_obj.full_url, None, request_headers) ct_header = '' try: response = urllib.request.urlopen(req, context=context) except HTTPError as e: query_obj.errors.append(e) return False except URLError as e: query_obj.errors.append(e) return False for k, v in response.info().items(): response_obj.response_headers.append((k, v)) if k == 'Content-Type': ct_header = str(k + ':' + v) response_obj.response_status = response.status if 'charset' in ct_header: charset = cgi.parse_header(ct_header)[1]['charset'] response_obj.response_body = response.read().decode(charset) else: response_obj.errors.append('%s(), no charset key in ct_header' % util.getFunctionName()) response_obj.response_body = response.read() return False return True