You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
293 lines
8.9 KiB
293 lines
8.9 KiB
|
|
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 = '' |
|
|
|
# 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.overview = 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 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']: |
|
copy = view_result() |
|
view_results.append(copy.copy_result(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 |
|
|
|
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()
|
|
|