[15] add f1_season attribute to metadata

This commit is contained in:
meisnate12 2022-03-21 18:18:51 -04:00
parent 1679fabbdf
commit c06fff3f3a
6 changed files with 181 additions and 4 deletions

View file

@ -1 +1 @@
1.16.2-develop14
1.16.2-develop15

View file

@ -214,6 +214,15 @@ class Cache:
frightening TEXT,
expiration_date TEXT)"""
)
cursor.execute(
"""CREATE TABLE IF NOT EXISTS ergast_race (
key INTEGER PRIMARY KEY,
season INTEGER,
round INTEGER,
name TEXT,
date TEXT,
expiration_date TEXT)"""
)
cursor.execute("SELECT count(name) FROM sqlite_master WHERE type='table' AND name='image_map'")
if cursor.fetchone()[0] > 0:
cursor.execute(f"SELECT DISTINCT library FROM image_map")
@ -736,4 +745,36 @@ class Cache:
update_sql = "UPDATE imdb_parental SET nudity = ?, violence = ?, profanity = ?, alcohol = ?, " \
"frightening = ?, expiration_date = ? WHERE imdb_id = ?"
cursor.execute(update_sql, (parental["nudity"], parental["violence"], parental["profanity"], parental["alcohol"],
parental["frightening"], expiration_date.strftime("%Y-%m-%d"), imdb_id))
parental["frightening"], expiration_date.strftime("%Y-%m-%d"), imdb_id))
def query_ergast(self, year, expiration):
ergast_list = []
expired = None
with sqlite3.connect(self.cache_path) as connection:
connection.row_factory = sqlite3.Row
with closing(connection.cursor()) as cursor:
cursor.execute("SELECT * FROM ergast_race WHERE season = ?", (year,))
for row in cursor.fetchall():
if row:
ergast_list.append({
"season": row["season"] if row["season"] else None,
"round": row["round"] if row["round"] else None,
"raceName": row["name"] if row["name"] else None,
"date": row["date"] if row["date"] else None
})
if not expired:
datetime_object = datetime.strptime(row["expiration_date"], "%Y-%m-%d")
time_between_insertion = datetime.now() - datetime_object
expired = time_between_insertion.days > expiration
return ergast_list, expired
def update_ergast(self, expired, season, races, expiration):
expiration_date = datetime.now() if expired is True else (datetime.now() - timedelta(days=random.randint(1, expiration)))
with sqlite3.connect(self.cache_path) as connection:
connection.row_factory = sqlite3.Row
with closing(connection.cursor()) as cursor:
cursor.excute("DELETE FROM ergast_race WHERE season = ?", (season,))
cursor.executemany("INSERT OR IGNORE INTO ergast_race(season, round) VALUES(?, ?)", [(r.season, r.round) for r in races])
cursor.executemany("UPDATE ergast_race SET name = ?, date = ?, expiration_date = ? WHERE season = ? AND round = ?",
[(r.name, r.date.strftime("%Y-%m-%d") if r.date else None,
expiration_date.strftime("%Y-%m-%d"), r.season, r.round) for r in races])

View file

@ -6,6 +6,7 @@ from modules.anidb import AniDB
from modules.anilist import AniList
from modules.cache import Cache
from modules.convert import Convert
from modules.ergast import Ergast
from modules.flixpatrol import FlixPatrol
from modules.icheckmovies import ICheckMovies
from modules.imdb import IMDb
@ -542,6 +543,7 @@ class ConfigFile:
self.ICheckMovies = ICheckMovies(self)
self.Letterboxd = Letterboxd(self)
self.Reciperr = Reciperr(self)
self.Ergast = Ergast(self)
logger.separator()

77
modules/ergast.py Normal file
View file

@ -0,0 +1,77 @@
from datetime import datetime, timedelta
from modules import util
from modules.util import Failed
logger = util.logger
base_url = "http://ergast.com/api/f1/"
class Race:
def __init__(self, data):
self._data = data
self.season = util.check_num(self._data["season"], is_int=True)
self.round = util.check_num(self._data["round"], is_int=True)
self.name = self._data["raceName"]
try:
self.date = datetime.strptime(self._data["date"], "%Y-%m-%d")
except (ValueError, TypeError):
self.date = None
def format_name(self, round_prefix, shorten_gp):
output = f"{self.round:02} - {self.name}" if round_prefix else self.name
return output.replace("Grand Prix", "GP") if shorten_gp else output
def session_info(self, title, sprint_weekend):
title = title.lower()
if "fp1" in title or "free practice 1" in title:
output = "Free Practice 1"
elif "fp2" in title or "free practice 2" in title:
output = "Free Practice 2"
elif "fp3" in title or "free practice 3" in title:
output = "Free Practice 3"
elif "sprint" in title and "pre" in title:
output = "Pre-Sprint Build-up"
elif "sprint" in title and "post" in title:
output = "Post-Sprint Analysis"
elif "sprint" in title:
output = "Sprint Qualifying"
elif "quali" in title and "pre" in title:
output = "Pre-Qualifying Build-up"
elif "quali" in title and "post" in title:
output = "Post-Qualifying Analysis"
elif "quali" in title:
output = "Qualifying Session"
elif "summary" in title or "highlight" in title:
output = "Highlights"
else:
output = "Race Session"
if "2160" in title or "4K" in title:
output = f"{output} (4K)"
if (sprint_weekend and ("Sprint" in output or "Free Practice 2" in output)) or \
(not sprint_weekend and ("Qualifying" in output or "Free Practice 3" in output)):
return output, self.date - timedelta(days=1)
elif (sprint_weekend and ("Qualifying" in output or "Free Practice 1" in output)) or \
(not sprint_weekend and ("Free Practice 1" in output or "Free Practice 2" in output)):
return output, self.date - timedelta(days=2)
else:
return output, self.date
class Ergast:
def __init__(self, config):
self.config = config
def get_races(self, year, ignore_cache=False):
expired = None
if self.config.Cache and not ignore_cache:
race_list, expired = self.config.Cache.query_ergast(year, self.config.Cache.expiration)
if race_list and expired is False:
return [Race(r) for r in race_list]
response = self.config.get(f"{base_url}{year}.json")
if response.status_code < 400:
races = [Race(r) for r in response.json()["MRData"]["RaceTable"]["Races"]]
if self.config.Cache and not ignore_cache:
self.config.Cache.update_ergast(expired, year, races, self.config.Cache.expiration)
return races
else:
raise Failed(f"Ergast Error: F1 Season: {year} Not found")

View file

@ -571,7 +571,7 @@ class MetadataFile(DataFile):
updated = False
edits = {}
def add_edit(name, current_item, group, alias, key=None, value=None, var_type="str"):
def add_edit(name, current_item, group=None, alias=None, key=None, value=None, var_type="str"):
if value or name in alias:
if value or group[alias[name]]:
if key is None: key = name
@ -918,6 +918,63 @@ class MetadataFile(DataFile):
updated = True
logger.info(f"Track: {track_num} on Album: {title} of {mapping_name} Details Update {'Complete' if updated else 'Not Needed'}")
if "f1_season" in methods and self.library.is_show:
f1_season = None
current_year = datetime.now().year
if meta[methods["f1_season"]] is None:
raise Failed("Metadata Error: f1_season attribute is blank")
try:
year_value = int(str(meta[methods["f1_season"]]))
if 1950 <= year_value <= current_year:
f1_season = year_value
except ValueError:
pass
if f1_season is None:
raise Failed(f"Metadata Error: f1_season attribute must be an integer between 1950 and {current_year}")
round_prefix = False
if "round_prefix" in methods:
if meta[methods["round_prefix"]] is True:
round_prefix = True
else:
logger.error("Metadata Error: round_prefix must be true to do anything")
shorten_gp = False
if "shorten_gp" in methods:
if meta[methods["shorten_gp"]] is True:
shorten_gp = True
else:
logger.error("Metadata Error: shorten_gp must be true to do anything")
logger.info(f"Setting Metadata of {item.title} to F1 Season {f1_season}")
races = self.config.Ergast.get_races(f1_season)
race_lookup = {r.round: r for r in races}
for season in item.seasons():
if season.seasonNumber is 0:
continue
sprint_weekend = False
for episode in season.episodes():
if "sprint" in episode.locations[0].lower():
sprint_weekend = True
break
if season.seasonNumber in race_lookup:
race = race_lookup[season.seasonNumber]
title = race.format_name(round_prefix, shorten_gp)
updated = False
edits = {}
add_edit("title", season, value=title)
if self.library.edit_item(season, title, "Season", edits):
updated = True
logger.info(f"Race {season.seasonNumber} of F1 Season {f1_season}: Details Update {'Complete' if updated else 'Not Needed'}")
for episode in season.episodes():
if len(episode.locations) > 0:
ep_title, session_date = race.session_info(episode.locations[0], sprint_weekend)
add_edit("title", episode, value=ep_title)
add_edit("originally_available", episode, key="originallyAvailableAt", var_type="date", value=session_date)
if self.library.edit_item(episode, f"{season.seasonNumber} Episode: {episode.episodeNumber}", "Season", edits):
updated = True
logger.info(f"Session {episode.title}: Details Update {'Complete' if updated else 'Not Needed'}")
else:
logger.warning(f"Ergast Error: No Round: {season.seasonNumber} for Season {f1_season}")
class PlaylistFile(DataFile):
def __init__(self, config, file_type, path):

View file

@ -487,7 +487,7 @@ def library_operations(config, library):
try:
parental_guide = config.IMDb.parental_guide(imdb_id)
labels = [f"{k.capitalize()}:{v}" for k, v in parental_guide.items() if library.mass_imdb_parental_labels == "with_none" or v != "None"]
library.edit_tags("label", item, append_tags=labels)
library.edit_tags("label", item, add_tags=labels)
except Failed:
pass