pvdb/video_metadata/views/tmdb.py

import cgi
from dataclasses import dataclass, field
from datetime import date
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, 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')


####################
# 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 = ''


####################
# 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
                ORDER BY
                    movie_details.title'''

    # 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():
    errors = []
    query = query_data()
    response = response_data()
    view_results = []

    # 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:
        view_results.clear()
        flash(errors)
        for e in errors:
            current_app.logger.error(e)

    return render_template(
            'tmdb/query.html',
            view_data = view_results,
            rel_path = query.rel_path,
            debug_str = '{} \n {} \n {} \n'.format(
                    util.dumpObject(query),
                    util.dumpObject(response),
                    util.dumpObject(view_results)
                )
        )

# 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',
            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 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('movie_queries count: "{}, query: {}"'.format(count, 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

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('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()