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.
609 lines
19 KiB
609 lines
19 KiB
|
|
import cgi |
|
from dataclasses import dataclass, field |
|
from datetime import date, datetime |
|
from enum import IntEnum |
|
import json |
|
import logging |
|
import os |
|
from pathlib import Path |
|
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, make_response, request, \ |
|
redirect, render_template, Response, url_for |
|
from guessit import guessit |
|
|
|
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' |
|
LOCAL_POSTER_PATH = 'image_cache' |
|
|
|
|
|
#################### |
|
# 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) |
|
is_valid:bool = True |
|
raw_query_str:str = '' |
|
parsed_query_str:str = '' |
|
title:str = '' |
|
media_type:media_type = media_type.UNKNOWN |
|
rel_path:str = '' |
|
query_params:str = '' |
|
full_url:str = '' |
|
movie_year:int = 0 |
|
tv_season:str = '' |
|
tv_episode: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 = '' |
|
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() |
|
view_results = [] |
|
errors = [] |
|
|
|
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') |
|
@bp.route('/images/<string:image_name>') |
|
def cache_image(image_name:str = ''): |
|
# validate image name |
|
if (image_name == '' |
|
or re.search(r'[^ a-zA-z0-9./]', image_name) |
|
or not Path(image_name).suffix in VALID_IMAGE_EXTENSIONS): |
|
current_app.logger.error(f'invalid image name: {image_name}') |
|
return redirect(url_for('static', filename=DEFAULT_POSTER_IMAGE)) |
|
|
|
# look for image in cache folder |
|
local_dir = Path(current_app.instance_path).joinpath(LOCAL_POSTER_PATH) |
|
image = Path(image_name).name |
|
p = local_dir.joinpath(image) |
|
current_app.logger.debug(f'checking for local path: {p}') |
|
|
|
if not p.exists(): |
|
# make request for remote image |
|
response = util.URLResponse |
|
|
|
if util.getURL(f'{TMDB_POSTER_PREFIX}/{image}', response, current_app.logger): |
|
# store remote image |
|
current_app.logger.debug(f'content-type for {image}: {response.content_type}') |
|
try: |
|
fp = open(p, 'wb') |
|
fp.write(response.body) |
|
fp.close() |
|
except OSError as e: |
|
current_app.logger.error(e) |
|
return redirect(url_for('static', filename=DEFAULT_POSTER_IMAGE)) |
|
|
|
# redirect to saved image |
|
try: |
|
fp = open(p, 'rb') |
|
image_bytes = fp.read() |
|
except OSError as e: |
|
current_app.logger.error(e) |
|
return redirect(url_for('static', filename=DEFAULT_POSTER_IMAGE)) |
|
|
|
def getCType(s): |
|
if s is '.jpg': |
|
return 'image/jpeg' |
|
elif s is '.png': |
|
return 'image/png' |
|
elif s is '.svg': |
|
return 'image/svg+xml' |
|
|
|
return make_response(tuple([image_bytes, {'Content-Type': getCType(p.suffix)} ])) |
|
|
|
|
|
################### |
|
# helpers |
|
|
|
def doQueryAlgorithm(query:query_data) -> dict: |
|
errors, view_results = [], [] |
|
response = response_data() |
|
query = guessMedia(query) |
|
|
|
if query.is_valid: |
|
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} |
|
|
|
|
|
def guessMedia(query:query_data) -> query_data: |
|
# NOTE: don't log debug messages for rebulk, which are excessive |
|
logging.disable(logging.DEBUG) |
|
guess = guessit(query.raw_query_str) |
|
logging.disable(logging.NOTSET) |
|
|
|
query.title = guess['title'] |
|
query.parsed_query_str = guess['title'].lower() |
|
|
|
try: |
|
if guess['type'] == 'movie': |
|
query.media_type = media_type.MOVIE |
|
query.movie_year = guess['year'] |
|
else: # NOTE: 'type' has to be 'episode' here. guessit has no 'unknown' equivalent |
|
query.media_type = media_type.TV_SHOW |
|
query.tv_season = guess['season'] |
|
query.tv_episode = guess['episode'] |
|
except KeyError as e: |
|
query.is_valid = False |
|
current_app.logger.error(f'invalid key: {e}, for query: {query.raw_query_str}') |
|
|
|
return query |
|
|
|
|
|
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 |
|
# TODO: if query_obj.media_type is still 'UNKNOWN' from parseQueryStr(), make queries to |
|
# both tmdb endpoints (movie and tv) |
|
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()
|
|
|