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.
370 lines
12 KiB
370 lines
12 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.rel_path = '' |
|
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('/list_movie_test') |
|
def list_movie_test(): |
|
conn = db.get_db() |
|
cur = conn.cursor() |
|
view_results = [] |
|
query = '''SELECT |
|
movie_details.*, movie_paths.movie_path |
|
FROM |
|
movie_details |
|
JOIN movie_paths ON movie_paths.movie_id=movie_details.movie_id''' |
|
|
|
# TODO: should probably loop on movie_id, and add a list of movie_paths because |
|
# there can be multiple paths associated with a movie_id |
|
for row in cur.execute(query): |
|
vr = buildViewResult(row) |
|
vr.rel_path = row['movie_path'] |
|
view_results.append(vr) |
|
|
|
return render_template('tmdb/list_movie_test.html.jinja', |
|
view_results = view_results, |
|
url_prefix = current_app.config['VIDEO_URL_PREFIX'] |
|
) |
|
|
|
@bp.route('/query_movie', methods=['GET', 'POST']) |
|
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 |
|
# form input comes from the query.html template and the links |
|
# come from the file_listing/index.html template |
|
if 'query' in request.form: |
|
query.raw_query_str = request.form['query'] |
|
elif 'query' in request.args and 'rel_path' in request.args: |
|
query.raw_query_str = request.args['query'] |
|
query.rel_path = request.args['rel_path'] |
|
|
|
# TODO: if query_obj.query_type is still 'UNKNOWN' from parseQueryStr(), make queries to |
|
# both tmdb endpoints (movie and tv) |
|
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, |
|
rel_path = query.rel_path, |
|
debug_str = '{} \n {} \n {} \n'.format( |
|
yaml.dump(query, indent=4), |
|
yaml.dump(response, indent=4), |
|
yaml.dump(view_results, indent=4) |
|
) |
|
) |
|
|
|
# TODO: add option for tv shows |
|
@bp.route('/create_mapping') |
|
def create_mapping(): |
|
movie_id = request.args.get('movie_id') |
|
rel_path = request.args.get('rel_path') |
|
errors = [] |
|
|
|
if not movie_id or not rel_path: |
|
errors.append('invalid parameter') |
|
else: |
|
conn = db.get_db() |
|
cur = conn.cursor() |
|
try: |
|
cur.execute('INSERT INTO movie_paths(movie_id, movie_path) VALUES(?, ?)', |
|
(movie_id, rel_path)) |
|
except DatabaseError as e: |
|
errors.append(e) |
|
|
|
if len(errors) == 0: |
|
conn.commit() |
|
|
|
if errors: |
|
flash(errors) |
|
|
|
return render_template('tmdb/create_mapping.html.jinja', |
|
movie_id = movie_id, |
|
rel_path = rel_path, |
|
debug_str = util.dumpObject({'movie_id': movie_id, 'rel_path': rel_path}), |
|
success = (len(errors) == 0) |
|
) |
|
|
|
|
|
################### |
|
# 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 = [] |
|
|
|
fancy_query = '''SELECT |
|
md.* |
|
FROM |
|
movie_queries mq |
|
JOIN movie_query_mapping map ON mq.query_id = map.query_id |
|
JOIN movie_details md ON md.movie_id = map.movie_id |
|
WHERE |
|
mq.query = ?''' |
|
|
|
for row in cur.execute(fancy_query, (query_obj.parsed_query_str, )): |
|
results.append(buildViewResult(row)) |
|
|
|
return results |
|
|
|
def buildViewResult(db_row): |
|
vr = view_result() |
|
vr.id = db_row['movie_id'] |
|
vr.title = db_row['title'] |
|
vr.overview = db_row['overview'] |
|
vr.release_date = db_row['release_date'] |
|
vr.original_language = db_row['original_language'] |
|
vr.tmdb_link = "{}{}{}".format(TMDB_MOVIE_LINK_URL, '/', vr.id) |
|
|
|
if db_row['poster_path'] != None: |
|
vr.poster_path = db_row['poster_path'] |
|
|
|
return vr |
|
|
|
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() |
|
query_id = 0 |
|
|
|
logger.debug('adding query string "{}" into local cache'.format(query_obj.parsed_query_str)) |
|
|
|
try: |
|
cur.execute('BEGIN TRANSACTION') |
|
cur.execute('INSERT INTO movie_queries(query, year) VALUES (?, ?)', |
|
(query_obj.parsed_query_str, query_obj.movie_year) |
|
) |
|
query_id = cur.lastrowid |
|
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, original_language, poster_path)" |
|
+ " VALUES (?,?,?,?,?,?)", |
|
(result.id, result.title, result.overview, result.release_date, |
|
result.original_language, result.poster_path) |
|
) |
|
|
|
cur.execute("INSERT INTO movie_query_mapping(query_id, movie_id)" |
|
+ " VALUES (?,?)", (query_id, result.id) |
|
) |
|
except DatabaseError as e: |
|
query_obj.errors.append(e) |
|
|
|
if len(query_obj.errors) == 0: |
|
conn.commit() |
|
else: |
|
conn.rollback()
|
|
|