cinnaboot
/
pvdb
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

use tmdbv3api https://github.com/AnthonyBloomer/tmdbv3api

master
cinnaboot 6 years ago
parent
2e3ecd90cc
commit
97f625a7d6
1 changed files with 50 additions and 117 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 161
      video_metadata/views/tmdb.py

161
video_metadata/views/tmdb.py
Unescape View File

@ -10,7 +10,9 @@ from pathlib import Path
import re
from sqlite3 import DatabaseError
import ssl
import tmdbv3api
from typing import Any, Dict, List
# TODO: might as well replace urllib with requests library since it's a dep of tmdbv3api anyway
from urllib.error import URLError, HTTPError
import urllib.parse
import urllib.request
@ -80,15 +82,15 @@ class response_data:
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 = ''
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/
@ -177,6 +179,8 @@ def manual_query():
view_results = []
errors = []
flash('boo')
if 'query' in request.form:
query.raw_query_str = request.form['query']
q_res = doQueryAlgorithm(query)
@ -344,34 +348,50 @@ def cache_image(image_name:str = ''):
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))
else:
tmdb = tmdbv3api.TMDb()
tmdb.api_key = current_app.config['TMDB_API_KEY']
res = []
try:
response.result_count = response.parsed_response['total_results']
except TypeError as e:
errors.append(str(e))
if query.media_type == media_type.MOVIE:
movie = tmdbv3api.Movie()
res = movie.search(query.parsed_query_str)
elif query.media_type == media_type.TV_SHOW:
tv = tmdbv3api.TV()
res = tv.search(query.parsed_query_str)
# 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))
for r in res:
current_app.logger.debug(util.dumpObject(r))
vr = view_result()
vr.detail_id = r.id
vr.media_type = media_type.MOVIE
vr.overview = r.overview
vr.genre_ids = r.genre_ids
vr.original_language = r.original_language
vr.poster_path = r.poster_path
if query.media_type == media_type.MOVIE:
vr.title = r.title
vr.tmdb_link = TMDB_MOVIE_LINK_URL + str(r.id)
vr.release_date = r.release_date
elif query.media_type == media_type.TV_SHOW:
vr.title = r.name
vr.tmdb_link = TMDB_TV_LINK_URL + str(r.id)
vr.release_date = r.first_air_date
view_results.append(vr)
storeTMDBQueryResults(query, view_results)
errors.extend(query.errors)
errors.extend(response.errors)
return {'errors': errors, 'results': view_results}
@ -468,94 +488,6 @@ def buildViewResult(db_row):
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()
@ -578,11 +510,11 @@ def storeTMDBQueryResults(query_obj, result_list):
# 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, ))
cur.execute(f'SELECT COUNT(*) FROM query_details WHERE {id_column}=?', (r.detail_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}'
f'skipping result that is already saved. id: {r.detail_id}, title: {r.title}'
)
break
@ -591,7 +523,8 @@ def storeTMDBQueryResults(query_obj, result_list):
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)
(r.detail_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

Write Preview
Loading…
Cancel
Save