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.
455 lines
14 KiB
455 lines
14 KiB
|
|
import cgi |
|
from dataclasses import dataclass, field |
|
from datetime import date, datetime |
|
from enum import Enum |
|
import json |
|
import re |
|
from sqlite3 import DatabaseError |
|
import ssl |
|
from typing import Any, Dict, List |
|
from urllib.error import URLError, HTTPError |
|
import urllib.parse |
|
import urllib.request |
|
|
|
from flask import Blueprint, current_app, flash, Flask, jsonify, request, render_template, Response |
|
|
|
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') |
|
|
|
|
|
#################### |
|
# data structures |
|
|
|
class query_type(Enum): |
|
UNKNOWN = 0 |
|
MOVIE = 1 |
|
TV_SHOW = 2 |
|
|
|
@dataclass |
|
class query_data: |
|
errors: List[str] = field(default_factory=list) |
|
full_url: str = '' |
|
movie_year: int = 0 |
|
parsed_query_str: str = '' |
|
query_params: str = '' |
|
query_type: query_type = query_type.UNKNOWN |
|
raw_query_str: str = '' |
|
rel_path: str = '' |
|
|
|
@dataclass |
|
class response_data: |
|
errors: List[str] = field(default_factory=list) |
|
parsed_response: Any = field(default_factory=object) |
|
response_body: str = '' |
|
response_headers: List[str] = field(default_factory=list) |
|
response_status: int = 0 |
|
result_count: int = 0 |
|
|
|
@dataclass |
|
class view_result: |
|
id: int = 0 |
|
title: str = '' |
|
release_date: str = '' |
|
overview: str = '' |
|
genre_ids: List[int] = field(default_factory=list) |
|
original_language: str = '' |
|
poster_path: str = 'no_image.png' |
|
tmdb_link: str = '' |
|
|
|
# TODO: flask 1.1 adds support for JSON encoding dataclasses |
|
# https://palletsprojects.com/blog/flask-1-1-released/ |
|
# add flask 1.1 to requirements when alpine distro updates flask |
|
# NOTE: haven't found a better way to json encode dataclasses :( |
|
class result_encoder(json.JSONEncoder): |
|
|
|
def default(self, obj): |
|
if isinstance(obj, view_result): |
|
return obj.__dict__ |
|
if isinstance(obj, date): |
|
return obj.isoformat() |
|
if isinstance(obj, datetime): |
|
return obj.isoformat() |
|
|
|
try: |
|
ret = json.JSONEncoder.default(self, obj) |
|
except TypeError: |
|
return '' |
|
return ret |
|
|
|
|
|
#################### |
|
# flask routes |
|
|
|
@bp.route('/list_movie_test') |
|
@bp.route('/list_movie_test/<string:sorting>') |
|
def list_movie_test(sorting = 'title'): |
|
if sorting == 'add-date': |
|
orderby = 'mp.add_date' |
|
elif sorting == 'add-date-desc': |
|
orderby = 'mp.add_date DESC' |
|
elif sorting == 'release-date': |
|
orderby = 'md.release_date' |
|
elif sorting == 'release-date-desc': |
|
orderby = 'md.release_date DESC' |
|
else: |
|
orderby = 'md.title' |
|
# TODO: order by genres, unwatched folder... |
|
|
|
conn = db.get_db() |
|
cur = conn.cursor() |
|
view_results = [] |
|
query = f'''SELECT |
|
md.*, mp.movie_path |
|
FROM |
|
movie_details md |
|
JOIN movie_paths mp ON mp.movie_id=md.movie_id |
|
ORDER BY |
|
{orderby}''' |
|
|
|
# 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', |
|
view_results = view_results, |
|
url_prefix = current_app.config['VIDEO_URL_PREFIX'] |
|
) |
|
|
|
|
|
@bp.route('/query_movie', methods=['GET', 'POST']) |
|
def query_movie(): |
|
query = query_data() |
|
|
|
if 'query' in request.form: |
|
query.raw_query_str = request.form['query'] |
|
|
|
q_res = doQueryAlgorithm(query) |
|
errors = q_res['errors'] |
|
view_results = q_res['results'] |
|
|
|
if errors: |
|
view_results.clear() |
|
flash(errors) |
|
for e in errors: |
|
current_app.logger.error(e) |
|
|
|
return render_template( |
|
'tmdb/query.html', |
|
view_data = view_results, |
|
debug_str = '{} \n {} \n'.format( |
|
util.dumpObject(query), |
|
util.dumpObject(view_results) |
|
) |
|
) |
|
|
|
|
|
@bp.route('/query_movie_json') |
|
def query_movie_json(): |
|
query = query_data() |
|
results = [] |
|
|
|
if 'query' in request.args and 'rel_path' in request.args: |
|
query.raw_query_str = request.args['query'] |
|
query.rel_path = request.args['rel_path'] |
|
q_res = doQueryAlgorithm(query) |
|
errors = q_res['errors'] |
|
results = q_res['results'] |
|
current_app.logger.debug(f'{util.getFunctionName()}, results length: {len(results)}') |
|
else: |
|
errors = ['required parameters: query, rel_path'] |
|
|
|
if errors: |
|
for e in errors: |
|
current_app.logger.error(e) |
|
return jsonify(errors=errors) |
|
|
|
res = Response(json.dumps(results, cls=result_encoder)) |
|
res.headers.set('Content-Type', 'application/json') |
|
return res |
|
|
|
|
|
# TODO: add option for tv shows |
|
@bp.route('/create_mapping', methods=['POST']) |
|
def create_mapping(): |
|
movie_id = request.form['movie_id'] |
|
rel_path = request.form['rel_path'] |
|
message = '' |
|
errors = [] |
|
|
|
if not movie_id or not rel_path: |
|
errors.append('invalid parameter') |
|
else: |
|
conn = db.get_db() |
|
cur = conn.cursor() |
|
current_app.logger.debug(f'{util.getFunctionName()}, mapping path "{rel_path}" for id: {movie_id}') |
|
try: |
|
cur.execute('INSERT INTO movie_paths(movie_id, movie_path, add_date) VALUES(?, ?, datetime())', |
|
(movie_id, rel_path )) |
|
except DatabaseError as e: |
|
errors.append(str(e)) |
|
|
|
if len(errors) == 0: |
|
conn.commit() |
|
message = f'Successfully added mapping for path "{rel_path}"' |
|
|
|
if errors: |
|
for e in errors: |
|
current_app.logger.error(e) |
|
|
|
return jsonify({'errors':errors, 'message':message}) |
|
|
|
|
|
################### |
|
# helpers |
|
|
|
def doQueryAlgorithm(query:query_data) -> dict: |
|
errors, view_results = [], [] |
|
response = response_data() |
|
|
|
# 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) |
|
current_app.logger.debug(f'{util.getFunctionName()}, view_results length: {len(view_results)}') |
|
else: |
|
makeTMDBQuery(query, response) |
|
|
|
try: |
|
response.parsed_response = json.loads(response.response_body) |
|
except json.JSONDecodeError as e: |
|
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) |
|
|
|
return {'errors': errors, 'results': view_results} |
|
|
|
|
|
# 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 for {}'.format( |
|
util.getFunctionName(), query_obj.raw_query_str) |
|
) |
|
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] |
|
current_app.logger.debug( |
|
f'{util.getFunctionName()}, movie_queries count: {count}, query: "{query_str}"' |
|
) |
|
|
|
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 |
|
|
|
|
|
# TODO: add info and debug logging for network requests |
|
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': current_app.config['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['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 |
|
|
|
current_app.logger.debug(f'{util.getFunctionName()}, adding query string ' |
|
f'"{query_obj.parsed_query_str}" into local cache') |
|
|
|
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: |
|
# NOTE: check stored details before trying to add to local db to avoid unique key |
|
# constraint error |
|
cur.execute('SELECT COUNT(*) FROM movie_details WHERE movie_id=?', (result.id,)) |
|
count = cur.fetchone()[0] |
|
if count > 0: |
|
current_app.logger.debug(f'{util.getFunctionName()}, ' |
|
f'skipping result that is already saved. id: {result.id}, title: {result.title}' |
|
) |
|
break |
|
|
|
try: |
|
current_app.logger.debug(f'{util.getFunctionName()}, adding movie details for "{result.title}"') |
|
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(f'{util.getFunctionName()}, database error \'{e}\'') |
|
|
|
if len(query_obj.errors) == 0: |
|
conn.commit() |
|
else: |
|
conn.rollback()
|
|
|