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.
578 lines
18 KiB
578 lines
18 KiB
|
|
import cgi |
|
from dataclasses import dataclass, field |
|
from datetime import date, datetime |
|
from enum import IntEnum |
|
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, redirect, \ |
|
render_template, Response, url_for |
|
|
|
from .. import util |
|
from .. import db |
|
from . import tv_shows |
|
|
|
|
|
#################### |
|
# 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/' |
|
TMDB_POSTER_PREFIX = 'https://image.tmdb.org/t/p/w154' |
|
VALID_IMAGE_EXTENSIONS = ('.jpg', '.png', '.svg') |
|
DEFAULT_POSTER_IMAGE = 'no_poster_w154.jpg' |
|
|
|
|
|
#################### |
|
# globals |
|
|
|
bp = Blueprint('tmdb', __name__, url_prefix='/tmdb') |
|
|
|
|
|
#################### |
|
# data structures |
|
|
|
class media_type(IntEnum): |
|
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 = '' |
|
media_type: media_type = media_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: |
|
detail_id:int = 0 |
|
media_type:str = 'unknown' |
|
movie_id: int = 0 |
|
tv_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 = 'pc.add_date' |
|
elif sorting == 'add-date-desc': |
|
orderby = 'pc.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.*, pc.path |
|
FROM |
|
query_details md |
|
JOIN path_cache pc ON pc.detail_id=md.detail_id |
|
ORDER BY |
|
{orderby}''' |
|
|
|
# TODO: should probably loop on movie_id, and add a list of 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['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('/tv_show_test', methods=['GET', ]) |
|
def tv_show_test(): |
|
shows = [] |
|
errors = [] |
|
|
|
if 'query' in request.args: |
|
result = tv_shows.doTVQuery(request.args['query']) |
|
shows = result['shows'] |
|
errors = result['errors'] |
|
|
|
if (shows): |
|
pass |
|
|
|
return render_template('tmdb/tv_show_test.html', |
|
shows = shows, |
|
debug_str = f'{util.dumpObject(errors)}' |
|
) |
|
|
|
|
|
@bp.route('/manual_query', methods=['GET', 'POST']) |
|
def manual_query(): |
|
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 = f'{util.dumpObject(query)} \n {util.dumpObject(view_results)} \n' |
|
) |
|
|
|
|
|
@bp.route('/query_media_json') |
|
def query_media_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'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 |
|
|
|
|
|
@bp.route('/query_path_json') |
|
def query_path_json(): |
|
errors = [] |
|
|
|
if not 'rel_path' in request.args: |
|
errors.append(f'required parameters: rel_path') |
|
else: |
|
path = request.args['rel_path'] |
|
cur = db.get_db().cursor() |
|
|
|
q = f'''SELECT |
|
qd.*, pc.* |
|
FROM |
|
path_cache pc |
|
JOIN query_details qd ON qd.detail_id = pc.detail_id |
|
WHERE |
|
pc.path = ? |
|
LIMIT 1''' |
|
|
|
cur.execute(q, (path, )) |
|
row = cur.fetchone() |
|
|
|
if errors: |
|
for e in errors: |
|
current_app.logger.error(e) |
|
|
|
res = Response(json.dumps(buildViewResult(row), cls=result_encoder)) |
|
res.headers.set('Content-Type', 'application/json') |
|
return res |
|
|
|
|
|
@bp.route('/create_path_mapping', methods=['POST']) |
|
def create_path_mapping(): |
|
message = '' |
|
errors = [] |
|
|
|
if not ('rel_path' in request.form |
|
and 'detail_id' in request.form |
|
): |
|
errors.append(f'required parameters: rel_path, id') |
|
elif request.form['rel_path'] is '': |
|
errors.append('rel_path is empty') |
|
elif request.form['detail_id'] is '0': |
|
errors.append('detail_id is unset') |
|
else: |
|
detail_id = request.form['detail_id'] |
|
rel_path = request.form['rel_path'] |
|
conn = db.get_db() |
|
cur = conn.cursor() |
|
current_app.logger.debug(f'mapping path "{rel_path}" for detail_id: {detail_id}') |
|
try: |
|
cur.execute(f'INSERT INTO path_cache(detail_id, path, add_date) ' |
|
f'VALUES(?,?, datetime())', |
|
(detail_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}) |
|
|
|
|
|
@bp.route('/images/<string:image_name>') |
|
def cache_image(image_name:str = ''): |
|
current_app.logger.debug(f'image_name: {image_name}') |
|
|
|
if image_name != '': |
|
s = re.sub(r'/', '', image_name) |
|
ext = re.compile(r'.[a-zA-Z]{3}$').search(image_name).group(0) |
|
print(f'ext: "{ext}"') |
|
|
|
if not ext in VALID_IMAGE_EXTENSIONS: |
|
print('oops') |
|
else: |
|
print(f'seems okay: {ext}') |
|
|
|
url = f'{TMDB_POSTER_PREFIX}/{s}' |
|
|
|
#res = Response(json.dumps(results, cls=result_encoder)) |
|
#res.headers.set('Content-Type', 'application/json') |
|
#return str(f'image_name: {image_name}, url: {url}') |
|
return redirect(url_for('static', filename=DEFAULT_POSTER_IMAGE)) |
|
|
|
|
|
################### |
|
# helpers |
|
|
|
def doQueryAlgorithm(query:query_data) -> dict: |
|
errors, view_results = [], [] |
|
response = response_data() |
|
|
|
# TODO: if query_obj.media_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, query.movie_year): |
|
view_results = makeLocalQuery(query) |
|
current_app.logger.debug(f'view_results length: {len(view_results)}') |
|
elif makeTMDBQuery(query, response): |
|
try: |
|
response.parsed_response = json.loads(response.response_body) |
|
except json.JSONDecodeError as e: |
|
errors.append(str(e)) |
|
|
|
try: |
|
response.result_count = response.parsed_response['total_results'] |
|
except TypeError as e: |
|
errors.append(str(e)) |
|
|
|
# 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.media_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.media_type = media_type.MOVIE |
|
return True |
|
|
|
# NOTE: if not a movie, check for a string that looks like a tv show |
|
if query_obj.media_type == media_type.UNKNOWN: |
|
m = re.compile(r's[0-9]{1,2}|season ?[0-9]{1,2}', re.IGNORECASE).search(s) |
|
|
|
if m: |
|
query_obj.parsed_query_str = s[0:m.start()].strip() |
|
query_obj.media_type = media_type.TV_SHOW |
|
return True |
|
|
|
current_app.logger.warning(f'unable to guess media type for "{query_obj.raw_query_str}"') |
|
return False |
|
|
|
|
|
def isQueryCached(query_str:str, year) -> bool: |
|
conn = db.get_db() |
|
cur = conn.cursor() |
|
cur.execute('SELECT COUNT(*) FROM query_cache WHERE query=? AND year=?', (query_str, year)) |
|
count = cur.fetchone()[0] |
|
current_app.logger.debug(f'query_cache count: {count}, query: "{query_str}", year: {year}') |
|
|
|
return (count > 0) |
|
|
|
|
|
def getIDColumn(mediaType:int) -> str: |
|
if int(mediaType) == media_type.MOVIE: |
|
return 'movie_id' |
|
elif int(mediaType) == media_type.TV_SHOW: |
|
return 'tv_id' |
|
|
|
return 'unknown' |
|
|
|
|
|
# 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 = [] |
|
id_column = getIDColumn(query_obj.media_type) |
|
|
|
query = f'''SELECT |
|
qd.* |
|
FROM |
|
query_cache qc |
|
JOIN query_mapping map ON qc.query_id = map.query_id |
|
JOIN query_details qd ON qd.detail_id = map.detail_id |
|
WHERE |
|
qc.query = ?''' |
|
|
|
for row in cur.execute(query, (query_obj.parsed_query_str, )): |
|
results.append(buildViewResult(row)) |
|
|
|
return results |
|
|
|
|
|
def buildViewResult(db_row): |
|
if not db_row: |
|
current_app.logger.error('empty db_row parameter') |
|
return None |
|
|
|
vr = view_result() |
|
vr.detail_id = db_row['detail_id'] |
|
vr.media_type = db_row['media_type'] |
|
vr.title = db_row['title'] |
|
vr.overview = db_row['overview'] |
|
vr.release_date = db_row['release_date'] |
|
vr.original_language = db_row['original_language'] |
|
|
|
if vr.media_type == media_type.MOVIE: |
|
vr.movie_id = db_row['movie_id'] |
|
vr.tmdb_link = f'{TMDB_MOVIE_LINK_URL}{vr.movie_id}' |
|
elif vr.media_type == media_type.TV_SHOW: |
|
vr.tv_id = db_row['tv_id'] |
|
vr.tmdb_link = f'{TMDB_TV_LINK_URL}{vr.tv_id}' |
|
|
|
if db_row['poster_path'] != None: |
|
vr.poster_path = db_row['poster_path'] |
|
|
|
return vr |
|
|
|
|
|
# TODO: need to check response headers for rate limiting requests |
|
def makeTMDBQuery(query_obj, response_obj): |
|
if query_obj.media_type == media_type.MOVIE: |
|
search_part = TMDB_SEARCH_MOVIE |
|
elif query_obj.media_type == media_type.TV_SHOW: |
|
search_part = TMDB_SEARCH_TV |
|
else: |
|
query_obj.errors.append('query type is not set') |
|
return False |
|
|
|
if query_obj.parsed_query_str == '': |
|
query_obj.errors.append(f'{util.getCurrentFileName()}:{util.getLineNum()},' |
|
f'{util.getFunctionName()}, empty query string') |
|
return False |
|
|
|
current_app.logger.info( |
|
f'making remote query to TMDB API: ' |
|
f'{query_obj.media_type.name}, "{query_obj.parsed_query_str}"' |
|
) |
|
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(str(e)) |
|
return False |
|
except URLError as e: |
|
query_obj.errors.append(str(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('no charset key in ct_header') |
|
response_obj.response_body = response.read() |
|
return False |
|
|
|
return True |
|
|
|
|
|
# NOTE: @param result is parsed JSON object from tmdb API |
|
def parseTMDBResult(result:object, media_type:int) -> view_result: |
|
parsed = view_result() |
|
|
|
try: |
|
parsed.id = result['id'] |
|
parsed.overview = result['overview'] |
|
parsed.genre_ids = result['genre_ids'] |
|
parsed.original_language = result['original_language'] |
|
parsed.poster_path = result['poster_path'] |
|
parsed.media_type = media_type |
|
|
|
if media_type == media_type.MOVIE: |
|
parsed.tmdb_link = TMDB_MOVIE_LINK_URL + str(result['id']) |
|
parsed.title = result['title'] |
|
parsed.release_date = result['release_date'] |
|
elif media_type == media_type.TV_SHOW: |
|
parsed.tmdb_link = TMDB_TV_LINK_URL + str(result['id']) |
|
parsed.title = result['name'] |
|
parsed.release_date = result['first_air_date'] |
|
except KeyError as e: |
|
current_app.logger.warning(f'missing key in JSON for "{parsed.title}": {e}') |
|
|
|
return parsed |
|
|
|
|
|
def storeTMDBQueryResults(query_obj, result_list): |
|
conn = db.get_db() |
|
cur = conn.cursor() |
|
query_id = 0 |
|
|
|
current_app.logger.debug(f'adding query string ' |
|
f'"{query_obj.parsed_query_str}" into local cache') |
|
|
|
try: |
|
cur.execute('BEGIN TRANSACTION') |
|
cur.execute('INSERT INTO query_cache(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 r in result_list: |
|
id_column = getIDColumn(r.media_type) |
|
|
|
# NOTE: check stored details before trying to add to local db to avoid unique key |
|
# constraint error |
|
cur.execute(f'SELECT COUNT(*) FROM query_details WHERE {id_column}=?', (r.id, )) |
|
count = cur.fetchone()[0] |
|
if count > 0: |
|
current_app.logger.debug( |
|
f'skipping result that is already saved. id: {r.id}, title: {r.title}' |
|
) |
|
break |
|
|
|
try: |
|
current_app.logger.debug(f'adding query details for "{r.title}"') |
|
cur.execute(f'INSERT INTO query_details({id_column}, ' |
|
+ 'media_type, title, overview, release_date, original_language, poster_path)' |
|
+ ' VALUES (?,?,?,?,?,?,?)', |
|
(r.id, r.media_type, r.title, r.overview, r.release_date, r.original_language, r.poster_path) |
|
) |
|
|
|
# NOTE: need to set detail_id here to avoid a second query in create_path_mapping |
|
r.detail_id = cur.lastrowid |
|
|
|
cur.execute(f'INSERT INTO query_mapping(query_id, detail_id)' |
|
+ ' VALUES (?,?)', (query_id, cur.lastrowid) |
|
) |
|
except DatabaseError as e: |
|
query_obj.errors.append(f'database error \'{e}\'') |
|
|
|
if len(query_obj.errors) == 0: |
|
conn.commit() |
|
else: |
|
conn.rollback()
|
|
|