Quico/mvw-dl
1
0
Fork 0
Code Issues 1 Pull Requests Projects Releases Wiki Activity

Use Pydantic JSON model files

Browse Source
This commit is contained in:
hygienic-books 2022-03-15 04:27:28 +01:00
parent 1110f3b024
commit 65fe4b707c
1 changed files with 19 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
main.py → mvw-dl.py
View File

@ -10,6 +10,9 @@ from rich.traceback import install
from rich import print from rich import print
import typing as t import typing as t
from rich.console import Console from rich.console import Console
from type_def.mvw_json_request import MVWJSONRequest
from type_def.mvw_json_response import MVWJSONResponse
# Without width # Without width
console = Console(width=180) console = Console(width=180)
p = inflect.engine() p = inflect.engine()
@ -158,16 +161,22 @@ def query_string_from_file(filename: str) -> str:
return query_string return query_string
def get_query_payload(section_name: str, config_obj: configparser.ConfigParser()) -> JSONType: def get_query_payload(
section_name: str,
config_obj: configparser.ConfigParser()) -> MVWJSONRequest:
log.debug(f"Generating HTTP POST JSON payload ...") log.debug(f"Generating HTTP POST JSON payload ...")
query = config_obj.get(section_name, "query") query = config_obj.get(section_name, "query")
if query[0] == "@": if query[0] == "@":
query = query.split("@", 1)[1] query = query.split("@", 1)[1]
query = query_string_from_file(query) query = query_string_from_file(query)
return json.loads(query) got_query_payload = MVWJSONRequest(**json.loads(query))
return got_query_payload
def get_json_response(section_name: str, config_obj: configparser.ConfigParser(), payload: JSONType) -> JSONType: def get_json_response(
section_name: str,
config_obj: configparser.ConfigParser(),
payload: MVWJSONRequest) -> MVWJSONResponse:
log.debug(f"Downloading JSON list of Mediathek files that match search criteria") log.debug(f"Downloading JSON list of Mediathek files that match search criteria")
serialized_payload = json.dumps(payload) serialized_payload = json.dumps(payload)
url = config_obj.get(section_name, "mvw_endpoint") url = config_obj.get(section_name, "mvw_endpoint")
@ -181,10 +190,15 @@ def get_json_response(section_name: str, config_obj: configparser.ConfigParser()
f"""{newline.join(f"Header '{header}': '{value}'" for header, value in list(req.headers.items()))}\n""" f"""{newline.join(f"Header '{header}': '{value}'" for header, value in list(req.headers.items()))}\n"""
f"Payload: {payload}") f"Payload: {payload}")
with s.send(prepped) as s: with s.send(prepped) as s:
return json.loads(s.content) # return json.loads(s.content)
got_json_response = MVWJSONResponse(**json.loads(s.content))
return got_json_response
def filter_json_by_duration(section_name: str, config_obj: configparser.ConfigParser(), json_obj: JSONType) -> JSONType: def filter_json_by_duration(
section_name: str,
config_obj: configparser.ConfigParser(),
json_obj: MVWJSONResponse) -> MVWJSONResponse:
min_length = config_obj.getint(section_name, "min_duration") min_length = config_obj.getint(section_name, "min_duration")
if min_length >= 0: if min_length >= 0:
log.debug(f"""Filtering JSON for minimum length of {min_length} {p.plural("second", min_length)}""") log.debug(f"""Filtering JSON for minimum length of {min_length} {p.plural("second", min_length)}""")