110 lines
3.3 KiB
Python
110 lines
3.3 KiB
Python
import json
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
from envfile import env_value
|
|
|
|
DEFAULT_API_URL = "http://localhost:5000"
|
|
IMAGE_PATH_PREFIX = "/api/library/images/"
|
|
REQUEST_TIMEOUT_SECONDS = 60
|
|
USER_AGENT = "sirblob-content-tool/1.0"
|
|
|
|
|
|
class LibraryError(ValueError):
|
|
pass
|
|
|
|
|
|
class Site:
|
|
def __init__(self, url: str, key: str | None):
|
|
self.url = url.rstrip("/")
|
|
self.key = key
|
|
|
|
def require_key(self) -> str:
|
|
if not self.key:
|
|
raise LibraryError(f"Set an API key in .env to change the library at {self.url}")
|
|
return self.key
|
|
|
|
|
|
def api_url() -> str:
|
|
return (env_value("LIBRARY_API_URL") or DEFAULT_API_URL).rstrip("/")
|
|
|
|
|
|
def local_site() -> Site:
|
|
return Site(api_url(), env_value("LIBRARY_API_KEY"))
|
|
|
|
|
|
def push_site() -> Site:
|
|
url = env_value("LIBRARY_PUSH_URL")
|
|
if not url:
|
|
raise LibraryError("Set LIBRARY_PUSH_URL in .env to the website to push to, for example https://dev.sirblob.co")
|
|
return Site(url, env_value("LIBRARY_PUSH_KEY") or env_value("LIBRARY_API_KEY"))
|
|
|
|
|
|
def error_message(error: urllib.error.HTTPError) -> str:
|
|
try:
|
|
return json.loads(error.read().decode("utf-8")).get("message") or error.reason
|
|
except ValueError:
|
|
return str(error.reason)
|
|
|
|
|
|
def send(method: str, path: str, body: dict | None = None, signed: bool = True, site: Site | None = None):
|
|
site = site or local_site()
|
|
headers = {"Accept": "application/json", "User-Agent": USER_AGENT}
|
|
data = None
|
|
if signed:
|
|
headers["Authorization"] = f"Bearer {site.require_key()}"
|
|
if body is not None:
|
|
headers["Content-Type"] = "application/json"
|
|
data = json.dumps(body).encode("utf-8")
|
|
request = urllib.request.Request(f"{site.url}{path}", data=data, headers=headers, method=method)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=REQUEST_TIMEOUT_SECONDS) as response:
|
|
text = response.read().decode("utf-8")
|
|
except urllib.error.HTTPError as error:
|
|
raise LibraryError(f"{site.url} said {error.code}: {error_message(error)}") from error
|
|
except urllib.error.URLError as error:
|
|
raise LibraryError(f"Could not reach {site.url}: {error.reason}") from error
|
|
return json.loads(text) if text else None
|
|
|
|
|
|
def quoted(identifier: str) -> str:
|
|
return urllib.parse.quote(identifier, safe="")
|
|
|
|
|
|
def fetch_library() -> dict:
|
|
return send("GET", "/api/library", signed=False)
|
|
|
|
|
|
def fetch_export(site: Site) -> dict:
|
|
return send("GET", "/api/library/export", site=site)
|
|
|
|
|
|
def save_media(imdb_id: str, fields: dict) -> dict:
|
|
return send("PUT", f"/api/library/media/{quoted(imdb_id)}", fields)
|
|
|
|
|
|
def delete_media(imdb_id: str) -> None:
|
|
send("DELETE", f"/api/library/media/{quoted(imdb_id)}")
|
|
|
|
|
|
def save_book(goodreads_id: str, fields: dict) -> dict:
|
|
return send("PUT", f"/api/library/books/{quoted(goodreads_id)}", fields)
|
|
|
|
|
|
def delete_book(goodreads_id: str) -> None:
|
|
send("DELETE", f"/api/library/books/{quoted(goodreads_id)}")
|
|
|
|
|
|
def save_game(igdb_id: str, fields: dict) -> dict:
|
|
return send("PUT", f"/api/library/games/{quoted(igdb_id)}", fields)
|
|
|
|
|
|
def delete_game(igdb_id: str) -> None:
|
|
send("DELETE", f"/api/library/games/{quoted(igdb_id)}")
|
|
|
|
|
|
def is_stored_image(path: str | None) -> bool:
|
|
return bool(path) and path.startswith(IMAGE_PATH_PREFIX)
|
|
|