import cgi from datetime import date from enum import Enum import json import re from sqlite3 import DatabaseError import ssl from urllib.error import URLError, HTTPError import urllib.parse import urllib.request import yaml from flask import Blueprint, current_app, flash, Flask, request, render_template from .. import util from .. import db #################### # constants 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/' #################### # globals bp = Blueprint('tmdb', __name__, url_prefix='/tmdb') logger = util.getLogger(Flask(__name__).config['DEBUG']) #################### # 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.release_date = 0 self.overview = '' self.genre_ids = [] self.original_language = '' self.poster_path = 'no_image.png' self.tmdb_link = '' #################### # 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 isQueryCached(query.parsed_query_str): view_results = makeLocalQuery(query) else: makeTMDBQuery(query, response) try: response.parsed_response = json.loads(response.response_body) except json.JSONDecodeError as e: view_data.errors.append(e) if 'total_results' in response.parsed_response: response.result_count = response.parsed_response['total_results'] # TODO: it's possible for the query results to span multiple 'pages' # see) https://developers.themoviedb.org/3/search/search-movies if 'results' in response.parsed_response: for result in response.parsed_response['results']: view_results.append(parseTMDBResult(result, query.query_type)) storeTMDBQueryResults(query, view_results) errors.extend(query.errors) errors.extend(response.errors) if errors: flash(errors) return render_template( 'tmdb/query.html', view_data = view_results, debug_str = '{} \n {} \n {} \n'.format( yaml.dump(query, indent=4), yaml.dump(response, indent=4), yaml.dump(view_results, indent=4) ) ) ################### # 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() s = re.sub('[(){}\[\]]', '', s).strip() # NOTE: remove some common characters 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('{}: unable to determine query type'.format(util.getFunctionName())) return False return True def isQueryCached(query_str): conn = db.get_db() cur = conn.cursor() cur.execute('SELECT COUNT(*) FROM movie_queries WHERE query=?', (query_str,)) count = cur.fetchone()[0] logger.debug('count: "{}"'.format(count)) return (count > 0) # NOTE: returns list of 'view_result' objects that match the query_obj.parsed_query_str # stored in the local db cache def makeLocalQuery(query_obj): conn = db.get_db() cur = conn.cursor() results = [] if query_obj.query_type == query_type.MOVIE: cur.execute('SELECT * FROM movie_queries WHERE query=?', (query_obj.parsed_query_str,)) elif query_obj.query_type == query_type.TV_SHOW: pass rows = cur.fetchall() logger.debug('number of rows matched: "{}"'.format(len(rows))) if len(rows) > 0: logger.debug('query string "{}", match found in local db'.format( query_obj.parsed_query_str) ) for row in rows: vr = view_result() vr.title = row["query"] results.append(vr) else: logger.debug('query string "{}", no matches found in local db'.format( query_obj.parsed_query_str) ) return results def makeTMDBQuery(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 # NOTE: @param result is parsed JSON object from tmdb API def parseTMDBResult(result, query_type): parsed = view_result() parsed.id = result['id'] parsed.title = result['original_title'] parsed.release_date = result['release_date'] parsed.overview = result['overview'] parsed.genre_ids = result['genre_ids'] parsed.original_language = result['original_language'] parsed.poster_path = result['poster_path'] if query_type == query_type.MOVIE: parsed.tmdb_link = TMDB_MOVIE_LINK_URL + str(result['id']) elif query_type == query_type.TV_SHOW: parsed.tmdb_link = TMDB_TV_LINK_URL + str(result['id']) return parsed def storeTMDBQueryResults(query_obj, result_list): conn = db.get_db() cur = conn.cursor() logger.debug('adding query string "{}" into local cache'.format(query_obj.parsed_query_str)) try: cur.execute('INSERT INTO movie_queries(query, year) VALUES (?, ?)', (query_obj.parsed_query_str, query_obj.movie_year) ) except DatabaseError as e: query_obj.errors.append(e) for result in result_list: try: cur.execute("INSERT INTO movie_details(" + "movie_id, title, overview, release_date, poster_path)" + " VALUES (?,?,?,?,?)", (result.id, result.title, result.overview, result.release_date, result.poster_path) ) except DatabaseError as e: query_obj.errors.append(e) conn.commit()