Add games library with IGDB, file explorer and static files script
This commit is contained in:
1 parent
a16444451a
commit
ec57bd73b3
65 files changed
+2230
-352
No files matched your search
+14
-6
@@ -7,6 +7,7 @@ from textual.widgets import DataTable, Footer, Header, TabbedContent, TabPane
|
||||
|
||||
import portfolio
|
||||
from tui.books_pane import BooksPane
|
||||
from tui.games_pane import GamesPane
|
||||
from tui.media_pane import MediaPane
|
||||
from tui.music_pane import MusicPane
|
||||
from tui.profile_pane import ProfilePane
|
||||
@@ -20,6 +21,7 @@ TABS = [
|
||||
("shows", "Shows"),
|
||||
("movies", "Movies"),
|
||||
("books", "Books"),
|
||||
("games", "Games"),
|
||||
("profile", "Profile"),
|
||||
("experience", "Experience"),
|
||||
("leadership", "Leadership"),
|
||||
@@ -28,6 +30,10 @@ TABS = [
|
||||
]
|
||||
|
||||
|
||||
def tab_key(number: int) -> str:
|
||||
return str(number % 10)
|
||||
|
||||
|
||||
class ContentApp(App):
|
||||
TITLE = "sirblob@engineering"
|
||||
SUB_TITLE = "~/content"
|
||||
@@ -35,7 +41,7 @@ class ContentApp(App):
|
||||
BINDINGS = [
|
||||
Binding("q", "quit", "Quit"),
|
||||
Binding("t", "next_theme", "Theme"),
|
||||
*[Binding(str(number), f"show_tab('{tab_id}')", label, show=False) for number, (tab_id, label) in enumerate(TABS, start=1)],
|
||||
*[Binding(tab_key(number), f"show_tab('{tab_id}')", label, show=False) for number, (tab_id, label) in enumerate(TABS, start=1)],
|
||||
]
|
||||
|
||||
def __init__(self):
|
||||
@@ -74,15 +80,17 @@ class ContentApp(App):
|
||||
yield MediaPane("movie")
|
||||
with TabPane("4 Books", id="books"):
|
||||
yield BooksPane()
|
||||
with TabPane("5 Profile", id="profile"):
|
||||
with TabPane("5 Games", id="games"):
|
||||
yield GamesPane()
|
||||
with TabPane("6 Profile", id="profile"):
|
||||
yield ProfilePane()
|
||||
with TabPane("6 Experience", id="experience"):
|
||||
with TabPane("7 Experience", id="experience"):
|
||||
yield RecordListPane(portfolio.EXPERIENCE)
|
||||
with TabPane("7 Leadership", id="leadership"):
|
||||
with TabPane("8 Leadership", id="leadership"):
|
||||
yield RecordListPane(portfolio.LEADERSHIP)
|
||||
with TabPane("8 Projects", id="projects"):
|
||||
with TabPane("9 Projects", id="projects"):
|
||||
yield ProjectsPane()
|
||||
with TabPane("9 Themes", id="themes"):
|
||||
with TabPane("0 Themes", id="themes"):
|
||||
yield ThemesPane()
|
||||
yield Footer()
|
||||
|
||||
|
||||
@@ -0,0 +1,105 @@
|
||||
import igdb
|
||||
import library_api
|
||||
|
||||
STATUSES = ["played", "playing", "planning", "replayable"]
|
||||
STATUS_ORDER = {"playing": 0, "replayable": 1, "played": 2, "planning": 3}
|
||||
PROFILE_FLAGS = {"favorite": "favorite", "liked": "liked"}
|
||||
SUGGESTED_TAGS = ["Casual", "Competitive", "Love it", "Open to play", "Looking to discuss"]
|
||||
|
||||
|
||||
def check_rating(rating: int | None) -> int | None:
|
||||
if rating is not None and not 0 <= rating <= 10:
|
||||
raise ValueError("Rating must be between 0 and 10")
|
||||
return rating
|
||||
|
||||
|
||||
def check_status(status: str) -> str:
|
||||
if status not in STATUSES:
|
||||
raise ValueError(f"Status must be one of {', '.join(STATUSES)}")
|
||||
return status
|
||||
|
||||
|
||||
def list_games() -> list[dict]:
|
||||
games = library_api.fetch_library()["games"]
|
||||
rating = lambda game: game["rating"] if game.get("rating") is not None else -1
|
||||
return sorted(games, key=lambda game: (STATUS_ORDER[game["status"]], -rating(game), game["name"].lower()))
|
||||
|
||||
|
||||
def lookup(link: str | None, name: str | None) -> dict:
|
||||
return igdb.details(igdb.find_game(link or None, name or None))
|
||||
|
||||
|
||||
def add_game(link: str | None, name: str | None, status: str = "played", rating: int | None = None, image: str | None = None) -> dict:
|
||||
if not link and not name:
|
||||
raise ValueError("Enter a store link, an IGDB link or a name")
|
||||
check_status(status)
|
||||
check_rating(rating)
|
||||
found = lookup(link, name)
|
||||
identifier = found.pop("igdbId")
|
||||
if any(game["igdbId"] == identifier for game in list_games()):
|
||||
raise ValueError(f"{found['name']} is already listed")
|
||||
is_store_link = bool(link) and not igdb.IGDB_SLUG_PATTERN.search(link)
|
||||
fields = {
|
||||
**found,
|
||||
"name": name or found["name"],
|
||||
"url": link if is_store_link else None,
|
||||
"status": status,
|
||||
"rating": rating,
|
||||
}
|
||||
if image:
|
||||
fields["imageSource"] = image
|
||||
return library_api.save_game(identifier, fields)
|
||||
|
||||
|
||||
def update_game(identifier: str, fields: dict) -> dict:
|
||||
return library_api.save_game(identifier, fields)
|
||||
|
||||
|
||||
def rate_game(identifier: str, rating: int | None) -> dict:
|
||||
return update_game(identifier, {"rating": check_rating(rating)})
|
||||
|
||||
|
||||
def set_status(identifier: str, status: str) -> dict:
|
||||
return update_game(identifier, {"status": check_status(status)})
|
||||
|
||||
|
||||
def set_dates(identifier: str, started: str | None, finished: str | None) -> dict:
|
||||
return update_game(identifier, {"started": started, "finished": finished})
|
||||
|
||||
|
||||
def set_links(identifier: str, url: str | None, image: str | None) -> dict:
|
||||
fields = {"url": url}
|
||||
if image:
|
||||
fields["imageSource"] = image
|
||||
return update_game(identifier, fields)
|
||||
|
||||
|
||||
def toggle_flag(game: dict, flag: str) -> dict:
|
||||
if flag not in PROFILE_FLAGS:
|
||||
raise ValueError(f"{flag} is not one of {', '.join(PROFILE_FLAGS)}")
|
||||
return update_game(game["igdbId"], {flag: not game.get(flag, False)})
|
||||
|
||||
|
||||
def set_tags(identifier: str, tags_text: str | None, note: str | None) -> dict:
|
||||
tags = [tag.strip() for tag in (tags_text or "").split(",") if tag.strip()]
|
||||
return update_game(identifier, {"tags": tags, "note": note or None})
|
||||
|
||||
|
||||
def profile_text(game: dict) -> str:
|
||||
return ", ".join(label for flag, label in PROFILE_FLAGS.items() if game.get(flag)) or "-"
|
||||
|
||||
|
||||
def remove_game(identifier: str) -> None:
|
||||
library_api.delete_game(identifier)
|
||||
|
||||
|
||||
def refresh_game(game: dict) -> dict:
|
||||
result = igdb.game_by_id(game["igdbId"])
|
||||
if not result:
|
||||
raise ValueError(f"{game['name']} is no longer on IGDB")
|
||||
found = igdb.details(result)
|
||||
found.pop("igdbId", None)
|
||||
found.pop("name", None)
|
||||
if game.get("image"):
|
||||
found.pop("imageSource", None)
|
||||
return update_game(game["igdbId"], found)
|
||||
@@ -0,0 +1,47 @@
|
||||
import sys
|
||||
import time
|
||||
|
||||
import games
|
||||
import jsonfile
|
||||
|
||||
ABOUT_FILE = "about.json"
|
||||
REQUEST_DELAY_SECONDS = 0.3
|
||||
|
||||
|
||||
def main() -> int:
|
||||
about = jsonfile.load(ABOUT_FILE)
|
||||
listed_games = about.get("games", [])
|
||||
if not listed_games:
|
||||
print("about.json has no games to import")
|
||||
return 0
|
||||
|
||||
try:
|
||||
already_listed = {game["name"].lower() for game in games.list_games()}
|
||||
except (OSError, ValueError) as error:
|
||||
print(f"Import stopped: {error}", file=sys.stderr)
|
||||
return 1
|
||||
failed = []
|
||||
for game in listed_games:
|
||||
if game["name"].lower() in already_listed:
|
||||
print(f"Already listed {game['name']}")
|
||||
continue
|
||||
try:
|
||||
saved = games.add_game(game.get("url"), game["name"], "played", None, game.get("image"))
|
||||
print(f"Imported {saved['name']} (IGDB {saved['igdbId']})")
|
||||
except (OSError, ValueError) as error:
|
||||
print(f"Skipped {game['name']}: {error}", file=sys.stderr)
|
||||
failed.append(game)
|
||||
time.sleep(REQUEST_DELAY_SECONDS)
|
||||
|
||||
if failed:
|
||||
print(f"{len(failed)} games were not imported, so about.json was left as it is. Fix them and run this again.", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
del about["games"]
|
||||
jsonfile.save(ABOUT_FILE, about)
|
||||
print(f"Imported {len(listed_games)} games and removed them from about.json")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,215 @@
|
||||
import argparse
|
||||
import datetime
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
|
||||
from envfile import env_value
|
||||
from jsonfile import PROJECT_ROOT
|
||||
|
||||
TOKEN_URL = "https://id.twitch.tv/oauth2/token"
|
||||
GAMES_URL = "https://api.igdb.com/v4/games"
|
||||
EXTERNAL_GAMES_URL = "https://api.igdb.com/v4/external_games"
|
||||
IMAGE_URL = "https://images.igdb.com/igdb/image/upload/{size}/{image_id}.jpg"
|
||||
TOKEN_CACHE_PATH = PROJECT_ROOT / "tools" / "content" / ".cache" / "igdb_token.json"
|
||||
TOKEN_EXPIRY_MARGIN_SECONDS = 300
|
||||
SUMMARY_MAX_LENGTH = 600
|
||||
GAME_FIELDS = ", ".join([
|
||||
"name",
|
||||
"slug",
|
||||
"url",
|
||||
"summary",
|
||||
"first_release_date",
|
||||
"total_rating",
|
||||
"genres.name",
|
||||
"platforms.name",
|
||||
"involved_companies.developer",
|
||||
"involved_companies.publisher",
|
||||
"involved_companies.company.name",
|
||||
"cover.image_id",
|
||||
"artworks.image_id",
|
||||
"screenshots.image_id",
|
||||
])
|
||||
IGDB_SLUG_PATTERN = re.compile(r"igdb\.com/games/([\w-]+)")
|
||||
STORE_ID_PATTERNS = {
|
||||
"store.steampowered.com": re.compile(r"store\.steampowered\.com/app/(\d+)"),
|
||||
"xbox.com": re.compile(r"xbox\.com/.*?/([0-9A-Za-z]{12})(?:[/?#]|$)"),
|
||||
}
|
||||
LINK_NOISE_SEGMENTS = {"store", "products", "product", "games", "game", "app", "p", "us", "en-us"}
|
||||
SEGMENT_ID_PATTERN = re.compile(r"\d+|[0-9A-Z]{12}")
|
||||
EPIC_SUFFIX_PATTERN = re.compile(r"-(?=[0-9a-f]*\d)[0-9a-f]{6}$")
|
||||
NAME_NOISE_PATTERN = re.compile(r"\b(switch|ps4|ps5|pc|xbox)\b", re.I)
|
||||
|
||||
|
||||
class IgdbError(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
def credentials() -> tuple[str, str]:
|
||||
client_id = env_value("IGDB_CLIENT_ID")
|
||||
client_secret = env_value("IGDB_CLIENT_SECRET")
|
||||
if not client_id or not client_secret:
|
||||
raise IgdbError("Set IGDB_CLIENT_ID and IGDB_CLIENT_SECRET in .env to look up games")
|
||||
return client_id, client_secret
|
||||
|
||||
|
||||
def cached_token() -> str | None:
|
||||
try:
|
||||
cached = json.loads(TOKEN_CACHE_PATH.read_text(encoding="utf-8"))
|
||||
except (OSError, ValueError):
|
||||
return None
|
||||
return cached["token"] if cached.get("expiresAt", 0) > time.time() else None
|
||||
|
||||
|
||||
def fetch_token(client_id: str, client_secret: str) -> str:
|
||||
parameters = urllib.parse.urlencode({"client_id": client_id, "client_secret": client_secret, "grant_type": "client_credentials"})
|
||||
request = urllib.request.Request(f"{TOKEN_URL}?{parameters}", method="POST")
|
||||
try:
|
||||
with urllib.request.urlopen(request, timeout=15) as response:
|
||||
result = json.load(response)
|
||||
except urllib.error.HTTPError as error:
|
||||
raise IgdbError(f"Twitch refused the IGDB keys ({error.code}). Check IGDB_CLIENT_ID and IGDB_CLIENT_SECRET") from error
|
||||
TOKEN_CACHE_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||||
expires_at = time.time() + result["expires_in"] - TOKEN_EXPIRY_MARGIN_SECONDS
|
||||
TOKEN_CACHE_PATH.write_text(json.dumps({"token": result["access_token"], "expiresAt": expires_at}), encoding="utf-8")
|
||||
return result["access_token"]
|
||||
|
||||
|
||||
def query(url: str, body: str) -> list[dict]:
|
||||
client_id, client_secret = credentials()
|
||||
token = cached_token() or fetch_token(client_id, client_secret)
|
||||
headers = {"Client-ID": client_id, "Authorization": f"Bearer {token}", "Accept": "application/json"}
|
||||
request = urllib.request.Request(url, data=body.encode("utf-8"), headers=headers, method="POST")
|
||||
try:
|
||||
with urllib.request.urlopen(request, timeout=15) as response:
|
||||
return json.load(response)
|
||||
except urllib.error.HTTPError as error:
|
||||
if error.code == 401:
|
||||
TOKEN_CACHE_PATH.unlink(missing_ok=True)
|
||||
raise IgdbError(f"IGDB returned {error.code}: {error.read().decode('utf-8', errors='replace')[:200]}") from error
|
||||
|
||||
|
||||
def quoted(text: str) -> str:
|
||||
return '"' + text.replace("\\", "\\\\").replace('"', '\\"') + '"'
|
||||
|
||||
|
||||
def game_by_id(identifier: str) -> dict | None:
|
||||
results = query(GAMES_URL, f"fields {GAME_FIELDS}; where id = {int(identifier)};")
|
||||
return results[0] if results else None
|
||||
|
||||
|
||||
def game_by_slug(slug: str) -> dict | None:
|
||||
results = query(GAMES_URL, f"fields {GAME_FIELDS}; where slug = {quoted(slug)};")
|
||||
return results[0] if results else None
|
||||
|
||||
|
||||
def game_by_store_link(link: str) -> dict | None:
|
||||
for host, pattern in STORE_ID_PATTERNS.items():
|
||||
match = pattern.search(link)
|
||||
if not match:
|
||||
continue
|
||||
uid = match.group(1).upper() if host == "xbox.com" else match.group(1)
|
||||
matches = query(EXTERNAL_GAMES_URL, f"fields game, url; where uid = {quoted(uid)}; limit 50;")
|
||||
same_store = [item for item in matches if host in (item.get("url") or "")]
|
||||
chosen = same_store or (matches if len(matches) == 1 else [])
|
||||
if chosen and chosen[0].get("game"):
|
||||
return game_by_id(str(chosen[0]["game"]))
|
||||
return None
|
||||
|
||||
|
||||
def search_games(name: str) -> list[dict]:
|
||||
return query(GAMES_URL, f"search {quoted(name)}; fields {GAME_FIELDS}; limit 10;")
|
||||
|
||||
|
||||
def search_game(name: str) -> dict | None:
|
||||
results = search_games(name)
|
||||
exact = [game for game in results if game.get("name", "").lower() == name.lower()]
|
||||
return (exact or results or [None])[0]
|
||||
|
||||
|
||||
def name_from_link(link: str) -> str:
|
||||
segments = [segment for segment in urllib.parse.urlparse(link).path.split("/") if segment]
|
||||
words = [segment for segment in segments if segment.lower() not in LINK_NOISE_SEGMENTS and not SEGMENT_ID_PATTERN.fullmatch(segment)]
|
||||
guess = EPIC_SUFFIX_PATTERN.sub("", words[-1]) if words else ""
|
||||
guess = NAME_NOISE_PATTERN.sub("", re.sub(r"[-_]+", " ", guess))
|
||||
return " ".join(guess.split())
|
||||
|
||||
|
||||
def find_game(link: str | None, name: str | None) -> dict:
|
||||
game = None
|
||||
if link and IGDB_SLUG_PATTERN.search(link):
|
||||
game = game_by_slug(IGDB_SLUG_PATTERN.search(link).group(1))
|
||||
if not game and link:
|
||||
game = game_by_store_link(link)
|
||||
search_name = name or (name_from_link(link) if link else "")
|
||||
if not game and search_name:
|
||||
game = search_game(search_name)
|
||||
if not game:
|
||||
raise IgdbError(f"Could not find {name or link} on IGDB. Try an IGDB link like https://www.igdb.com/games/hades-ii")
|
||||
return game
|
||||
|
||||
|
||||
def image_url(image: dict | None, size: str) -> str | None:
|
||||
return IMAGE_URL.format(size=size, image_id=image["image_id"]) if image and image.get("image_id") else None
|
||||
|
||||
|
||||
def first_image(game: dict, key: str) -> dict | None:
|
||||
images = game.get(key) or []
|
||||
return images[0] if images else None
|
||||
|
||||
|
||||
def company_names(game: dict, role: str) -> list[str]:
|
||||
return [item["company"]["name"] for item in game.get("involved_companies") or [] if item.get(role) and item.get("company")]
|
||||
|
||||
|
||||
def short_summary(text: str | None) -> str | None:
|
||||
if not text or len(text) <= SUMMARY_MAX_LENGTH:
|
||||
return text
|
||||
return text[:SUMMARY_MAX_LENGTH].rsplit(" ", 1)[0] + "..."
|
||||
|
||||
|
||||
def release_date(game: dict) -> str | None:
|
||||
seconds = game.get("first_release_date")
|
||||
if not isinstance(seconds, (int, float)):
|
||||
return None
|
||||
return datetime.datetime.fromtimestamp(seconds, tz=datetime.timezone.utc).date().isoformat()
|
||||
|
||||
|
||||
def details(game: dict) -> dict:
|
||||
wide_image = first_image(game, "artworks") or first_image(game, "screenshots")
|
||||
return {
|
||||
"igdbId": str(game["id"]),
|
||||
"name": game.get("name"),
|
||||
"igdbUrl": game.get("url"),
|
||||
"released": release_date(game),
|
||||
"platforms": [platform["name"] for platform in game.get("platforms") or [] if platform.get("name")],
|
||||
"developers": company_names(game, "developer"),
|
||||
"publishers": company_names(game, "publisher"),
|
||||
"genres": [genre["name"] for genre in game.get("genres") or [] if genre.get("name")],
|
||||
"summary": short_summary(game.get("summary")),
|
||||
"score": round(game["total_rating"]) if game.get("total_rating") is not None else None,
|
||||
"imageSource": image_url(wide_image, "t_1080p"),
|
||||
"coverSource": image_url(game.get("cover"), "t_cover_big"),
|
||||
}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description="Look up a video game on IGDB")
|
||||
parser.add_argument("text", help="a store link, an IGDB link or a game name")
|
||||
arguments = parser.parse_args()
|
||||
is_link = arguments.text.startswith("http")
|
||||
try:
|
||||
result = details(find_game(arguments.text if is_link else None, None if is_link else arguments.text))
|
||||
except (OSError, ValueError) as error:
|
||||
print(f"Error: {error}", file=sys.stderr)
|
||||
return 1
|
||||
print(json.dumps(result, indent="\t", ensure_ascii=False))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -75,5 +75,14 @@ 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)
|
||||
|
||||
@@ -1,209 +0,0 @@
|
||||
import argparse
|
||||
import html
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import jsonfile
|
||||
from html_page import download_page, meta_tags, page_title, plain_text
|
||||
|
||||
GAME_URL_PATTERN = re.compile(r"mobygames\.com/game/(\d+)")
|
||||
HEADING_PATTERN = re.compile(r"<h1[^>]*>(.*?)</h1>", re.S | re.I)
|
||||
METADATA_PATTERN = re.compile(r"<dt[^>]*>(.*?)</dt>\s*<dd[^>]*>(.*?)</dd>", re.S | re.I)
|
||||
LINK_PATTERN = re.compile(r"<a\s[^>]*href=[\"']([^\"']*)[\"'][^>]*>(.*?)</a>", re.S | re.I)
|
||||
DESCRIPTION_PATTERN = re.compile(r"<div[^>]*id=[\"']description-text[\"'][^>]*>(.*?)</div>", re.S | re.I)
|
||||
COVER_PATTERN = re.compile(r"https://cdn\.mobygames\.com/covers/[^\"'\s)]+")
|
||||
SCREENSHOT_PATTERN = re.compile(r"https://cdn\.mobygames\.com/screenshots/[^\"'\s)]+")
|
||||
NUMBER_PATTERN = re.compile(r"\d+(?:\.\d+)?")
|
||||
PERCENT_PATTERN = re.compile(r"(\d+)\s*%")
|
||||
BLOCKED_PAGE_HINTS = ["just a moment", "cf-chl", "captcha", "access denied", "are you a human"]
|
||||
DETAILS_FILE = "gameDetails.json"
|
||||
DESCRIPTION_MAX_LENGTH = 600
|
||||
|
||||
|
||||
class BlockedError(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
def game_id(url: str) -> str | None:
|
||||
match = GAME_URL_PATTERN.search(url)
|
||||
return match.group(1) if match else None
|
||||
|
||||
|
||||
def blocked_reason(page: str) -> str | None:
|
||||
lowered = page.lower()
|
||||
hint = next((hint for hint in BLOCKED_PAGE_HINTS if hint in lowered), None)
|
||||
if hint and not METADATA_PATTERN.search(page):
|
||||
return f"the page looks like a bot check (found '{hint}')"
|
||||
return None
|
||||
|
||||
|
||||
def metadata(page: str) -> dict[str, str]:
|
||||
fields = {}
|
||||
for label, value in METADATA_PATTERN.findall(page):
|
||||
name = (plain_text(label) or "").strip().lower()
|
||||
if name and name not in fields:
|
||||
fields[name] = value
|
||||
return fields
|
||||
|
||||
|
||||
def field_starting_with(fields: dict[str, str], *prefixes: str) -> str:
|
||||
return next((value for name, value in fields.items() if name.startswith(prefixes)), "")
|
||||
|
||||
|
||||
def link_texts(fragment: str, path_part: str | None = None) -> list[str]:
|
||||
names = []
|
||||
for href, text in LINK_PATTERN.findall(fragment):
|
||||
name = plain_text(text)
|
||||
if name and (path_part is None or path_part in href) and name not in names:
|
||||
names.append(name)
|
||||
return names
|
||||
|
||||
|
||||
def first_number(fragment: str) -> float | None:
|
||||
match = NUMBER_PATTERN.search(plain_text(fragment) or "")
|
||||
return float(match.group(0)) if match else None
|
||||
|
||||
|
||||
def first_percent(fragment: str) -> int | None:
|
||||
match = PERCENT_PATTERN.search(plain_text(fragment) or "")
|
||||
return int(match.group(1)) if match else None
|
||||
|
||||
|
||||
def short_description(text: str | None) -> str | None:
|
||||
if not text or len(text) <= DESCRIPTION_MAX_LENGTH:
|
||||
return text
|
||||
return text[:DESCRIPTION_MAX_LENGTH].rsplit(" ", 1)[0] + "..."
|
||||
|
||||
|
||||
def description_of(page: str, meta: dict) -> str | None:
|
||||
match = DESCRIPTION_PATTERN.search(page)
|
||||
text = plain_text(match.group(1)) if match else None
|
||||
return short_description(text or meta.get("og:description") or meta.get("description"))
|
||||
|
||||
|
||||
def title_of(page: str, meta: dict) -> str:
|
||||
heading = HEADING_PATTERN.search(page)
|
||||
title = plain_text(heading.group(1)) if heading else None
|
||||
return title or meta.get("og:title") or page_title(page).split(" - ")[0]
|
||||
|
||||
|
||||
def first_match(pattern: re.Pattern, page: str) -> str | None:
|
||||
match = pattern.search(page)
|
||||
return html.unescape(match.group(0)) if match else None
|
||||
|
||||
|
||||
def parse_game(page: str, identifier: str, url: str) -> dict:
|
||||
meta = meta_tags(page)
|
||||
fields = metadata(page)
|
||||
released = field_starting_with(fields, "released")
|
||||
release_fields = "".join(value for name, value in fields.items() if name.startswith(("released", "releases", "platform")))
|
||||
genre_fields = "".join(fields.get(name, "") for name in ["genre", "perspective", "gameplay", "setting"])
|
||||
|
||||
return {
|
||||
"mobygamesId": identifier,
|
||||
"url": url,
|
||||
"title": title_of(page, meta),
|
||||
"released": (link_texts(released) or [plain_text(released)])[0] if released else None,
|
||||
"platforms": link_texts(release_fields, "/platform/"),
|
||||
"developers": link_texts(field_starting_with(fields, "developer")),
|
||||
"publishers": link_texts(field_starting_with(fields, "publisher")),
|
||||
"genres": link_texts(genre_fields),
|
||||
"mobyScore": first_number(field_starting_with(fields, "moby score")),
|
||||
"criticScore": first_percent(field_starting_with(fields, "critic")),
|
||||
"playerScore": first_number(field_starting_with(fields, "player")),
|
||||
"description": description_of(page, meta),
|
||||
"cover": first_match(COVER_PATTERN, page) or meta.get("og:image"),
|
||||
"screenshot": first_match(SCREENSHOT_PATTERN, page),
|
||||
}
|
||||
|
||||
|
||||
def scrape_game(url: str) -> dict:
|
||||
identifier = game_id(url)
|
||||
if not identifier:
|
||||
raise ValueError(f"not a MobyGames game url: {url}")
|
||||
page = download_page(url)
|
||||
reason = blocked_reason(page.text)
|
||||
if reason:
|
||||
raise BlockedError(f"MobyGames blocked the request ({reason}), wait a few minutes and try again")
|
||||
return parse_game(page.text, identifier, url)
|
||||
|
||||
|
||||
def save_game(game: dict) -> None:
|
||||
details = jsonfile.load_data(DETAILS_FILE, {})
|
||||
details[game["mobygamesId"]] = {key: value for key, value in game.items() if key != "mobygamesId"}
|
||||
jsonfile.save_data(DETAILS_FILE, dict(sorted(details.items())))
|
||||
|
||||
|
||||
def print_diagnostics(url: str, save_path: Path | None):
|
||||
page = download_page(url)
|
||||
print(f"Status {page.status}")
|
||||
print(f"Final url {page.final_url}")
|
||||
print(f"Size {len(page.text):,} characters")
|
||||
print(f"Title {page_title(page.text) or '-'}")
|
||||
print(f"Fields {', '.join(metadata(page.text)) or '-'}")
|
||||
print(f"Meta tags {len(meta_tags(page.text))}")
|
||||
print(f"Blocked {blocked_reason(page.text) or 'no sign of a bot check'}")
|
||||
if save_path:
|
||||
save_path.write_text(page.text, encoding="utf-8")
|
||||
print(f"Saved html to {save_path}")
|
||||
|
||||
|
||||
def print_summary(game: dict):
|
||||
print(game["title"] or "unknown title")
|
||||
print("-" * 60)
|
||||
for label, key in [
|
||||
("Released", "released"),
|
||||
("Platforms", "platforms"),
|
||||
("Developers", "developers"),
|
||||
("Publishers", "publishers"),
|
||||
("Genres", "genres"),
|
||||
("Moby score", "mobyScore"),
|
||||
("Critics", "criticScore"),
|
||||
("Players", "playerScore"),
|
||||
("Cover", "cover"),
|
||||
("Screenshot", "screenshot"),
|
||||
]:
|
||||
value = game[key]
|
||||
text = ", ".join(value) if isinstance(value, list) else value
|
||||
print(f"{label:<11} {text if text not in (None, '') else '-'}")
|
||||
if game["description"]:
|
||||
print(f"\n{game['description']}")
|
||||
missing = [key for key, value in game.items() if value in (None, "", [])]
|
||||
if missing:
|
||||
print(f"\nNot found on the page: {', '.join(missing)}")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description="Scrape video game info from MobyGames")
|
||||
parser.add_argument("urls", nargs="+", help="MobyGames game urls, for example https://www.mobygames.com/game/223472/hades-ii/")
|
||||
parser.add_argument("--json", action="store_true", help="print the raw result as JSON")
|
||||
parser.add_argument("--save", action="store_true", help=f"save the result into data/{DETAILS_FILE} for the site")
|
||||
parser.add_argument("--debug", action="store_true", help="show what the downloaded page contains instead of parsing it")
|
||||
parser.add_argument("--save-html", type=Path, help="with --debug, save the downloaded page to this file")
|
||||
arguments = parser.parse_args()
|
||||
|
||||
exit_code = 0
|
||||
for url in arguments.urls:
|
||||
try:
|
||||
if arguments.debug:
|
||||
print_diagnostics(url, arguments.save_html)
|
||||
continue
|
||||
game = scrape_game(url)
|
||||
except (OSError, ValueError) as error:
|
||||
print(f"Error for {url}: {error}", file=sys.stderr)
|
||||
exit_code = 1
|
||||
continue
|
||||
if arguments.save:
|
||||
save_game(game)
|
||||
if arguments.json:
|
||||
print(json.dumps(game, indent="\t", ensure_ascii=False))
|
||||
else:
|
||||
print_summary(game)
|
||||
print()
|
||||
return exit_code
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -1,2 +1,3 @@
|
||||
textual>=8.2,<9
|
||||
pillow>=11
|
||||
pillow-heif>=0.18
|
||||
@@ -0,0 +1,106 @@
|
||||
import argparse
|
||||
import shutil
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
from PIL import Image, ImageOps
|
||||
from pillow_heif import register_heif_opener
|
||||
|
||||
from jsonfile import PROJECT_ROOT
|
||||
|
||||
FILES_DIR = PROJECT_ROOT / "static" / "files"
|
||||
ORIGINALS_DIR = PROJECT_ROOT / "tools" / "content" / ".cache" / "heic-originals"
|
||||
HEIC_EXTENSIONS = {".heic", ".heif"}
|
||||
PHOTO_EXTENSIONS = {".jpg", ".jpeg", ".png", ".webp"}
|
||||
JPEG_QUALITY = 90
|
||||
GPS_TAG = 0x8825
|
||||
|
||||
register_heif_opener()
|
||||
|
||||
|
||||
def visible_files(folder: Path, extensions: set[str]) -> list[Path]:
|
||||
return sorted(
|
||||
path
|
||||
for path in folder.rglob("*")
|
||||
if path.is_file() and path.suffix.lower() in extensions and not any(part.startswith(".") for part in path.relative_to(folder).parts)
|
||||
)
|
||||
|
||||
|
||||
def save_without_metadata(image: Image.Image, target: Path) -> None:
|
||||
upright = ImageOps.exif_transpose(image)
|
||||
if target.suffix.lower() in {".jpg", ".jpeg"}:
|
||||
upright.convert("RGB").save(target, "JPEG", quality=JPEG_QUALITY, optimize=True)
|
||||
else:
|
||||
upright.save(target)
|
||||
|
||||
|
||||
def keep_original(source: Path) -> Path:
|
||||
backup = ORIGINALS_DIR / source.relative_to(FILES_DIR)
|
||||
backup.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.move(source, backup)
|
||||
return backup
|
||||
|
||||
|
||||
def convert_heic(source: Path, delete_originals: bool) -> str:
|
||||
target = source.with_suffix(".jpg")
|
||||
if target.exists():
|
||||
return f"skipped {source.relative_to(FILES_DIR)}, {target.name} already exists"
|
||||
with Image.open(source) as image:
|
||||
save_without_metadata(image, target)
|
||||
if delete_originals:
|
||||
source.unlink()
|
||||
return f"converted {target.relative_to(FILES_DIR)}"
|
||||
return f"converted {target.relative_to(FILES_DIR)}, original kept in {keep_original(source).relative_to(PROJECT_ROOT)}"
|
||||
|
||||
|
||||
def has_location(path: Path) -> bool:
|
||||
try:
|
||||
with Image.open(path) as image:
|
||||
return bool(image.getexif().get_ifd(GPS_TAG))
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
|
||||
def strip_location(path: Path) -> str:
|
||||
with Image.open(path) as image:
|
||||
image.load()
|
||||
save_without_metadata(image, path)
|
||||
return f"removed location from {path.relative_to(FILES_DIR)}"
|
||||
|
||||
|
||||
def report(action) -> int:
|
||||
try:
|
||||
print(action())
|
||||
return 0
|
||||
except (OSError, ValueError) as error:
|
||||
print(f"Error: {error}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description="Convert HEIC photos in static/files to JPEG and remove location data from photos")
|
||||
parser.add_argument("--delete-originals", action="store_true", help="delete converted HEIC files instead of moving them to tools/content/.cache")
|
||||
parser.add_argument("--keep-location", action="store_true", help="leave GPS data in existing photos")
|
||||
arguments = parser.parse_args()
|
||||
|
||||
if not FILES_DIR.exists():
|
||||
print(f"{FILES_DIR.relative_to(PROJECT_ROOT)} does not exist", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
heic_files = visible_files(FILES_DIR, HEIC_EXTENSIONS)
|
||||
located_photos = [] if arguments.keep_location else [photo for photo in visible_files(FILES_DIR, PHOTO_EXTENSIONS) if has_location(photo)]
|
||||
if not heic_files and not located_photos:
|
||||
print("Nothing to do: no HEIC files and no photos with location data")
|
||||
return 0
|
||||
|
||||
failed = 0
|
||||
for source in heic_files:
|
||||
failed += report(lambda: convert_heic(source, arguments.delete_originals))
|
||||
for photo in located_photos:
|
||||
failed += report(lambda: strip_location(photo))
|
||||
print(f"Done: {len(heic_files) + len(located_photos) - failed} changed, {failed} failed")
|
||||
return 1 if failed else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,195 @@
|
||||
from rich.text import Text
|
||||
from textual.app import ComposeResult
|
||||
from textual.binding import Binding
|
||||
from textual.widgets import DataTable
|
||||
|
||||
import games
|
||||
from tui.cache_indicator import cache_state, cache_text
|
||||
from tui.dialogs import ConfirmDialog, FormDialog, FormField
|
||||
from tui.library_pane import LibraryPane
|
||||
|
||||
STATUS_STYLES = {"playing": "blue", "played": "green", "planning": "magenta", "replayable": "cyan"}
|
||||
|
||||
|
||||
def rating_text(game: dict) -> str:
|
||||
return f"{game['rating']}/10" if game.get("rating") is not None else "-"
|
||||
|
||||
|
||||
def image_state(game: dict) -> str:
|
||||
has_details = bool(game.get("released") or game.get("genres"))
|
||||
return cache_state(has_details, bool(game.get("image")))
|
||||
|
||||
|
||||
class GamesPane(LibraryPane):
|
||||
BINDINGS = [
|
||||
Binding("a", "add", "Add"),
|
||||
Binding("e", "rate", "Rate"),
|
||||
Binding("s", "status", "Status"),
|
||||
Binding("l", "links", "Links"),
|
||||
Binding("f", "toggle('favorite')", "Favorite"),
|
||||
Binding("k", "toggle('liked')", "Like"),
|
||||
Binding("g", "tags", "Tags"),
|
||||
Binding("h", "dates", "Dates"),
|
||||
Binding("d", "delete", "Delete"),
|
||||
Binding("u", "refresh", "Refresh"),
|
||||
Binding("r", "reload", "Reload"),
|
||||
]
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
super().__init__(**kwargs)
|
||||
self.border_subtitle = "games"
|
||||
self.games: dict[str, dict] = {}
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
yield DataTable(cursor_type="row", zebra_stripes=True)
|
||||
|
||||
def on_mount(self):
|
||||
table = self.table()
|
||||
table.add_column("Status", key="status")
|
||||
table.add_column("Rating", key="rating")
|
||||
table.add_column("Name", key="name")
|
||||
table.add_column("Released", key="released")
|
||||
table.add_column("Developer", key="developer")
|
||||
table.add_column("Profile", key="profile")
|
||||
table.add_column("Details", key="details")
|
||||
table.add_column("IGDB", key="igdb")
|
||||
self.reload()
|
||||
|
||||
def reload(self, focus_key: str | None = None):
|
||||
table = self.table()
|
||||
table.clear()
|
||||
loaded = self.load_entries(games.list_games)
|
||||
self.games = {game["igdbId"]: game for game in loaded}
|
||||
for game in loaded:
|
||||
table.add_row(
|
||||
Text(game["status"], style=STATUS_STYLES[game["status"]]),
|
||||
rating_text(game),
|
||||
game["name"],
|
||||
game.get("released") or "-",
|
||||
(game.get("developers") or ["-"])[0],
|
||||
games.profile_text(game),
|
||||
cache_text(image_state(game)),
|
||||
game["igdbId"],
|
||||
key=game["igdbId"],
|
||||
)
|
||||
self.restore_cursor(focus_key)
|
||||
|
||||
def selected_game(self) -> dict | None:
|
||||
return self.games.get(self.selected_key())
|
||||
|
||||
def action_add(self):
|
||||
fields = [
|
||||
FormField("link", "Store or IGDB link", placeholder="https://store.steampowered.com/app/2868840/Slay_the_Spire_2/"),
|
||||
FormField("name", "Name (searched on IGDB when the link has no match)"),
|
||||
FormField("status", "Status", kind="select", options=games.STATUSES, value="played"),
|
||||
FormField("rating", "Your rating (0 to 10, optional)", kind="rating"),
|
||||
FormField("image", "Image (optional site path or URL, IGDB artwork when empty)", placeholder="/img/minecraft.png"),
|
||||
]
|
||||
self.app.push_screen(FormDialog("Add a game", fields, "Add"), self.add_game)
|
||||
|
||||
def add_game(self, values: dict | None):
|
||||
if not values:
|
||||
return
|
||||
|
||||
def added(game: dict):
|
||||
self.reload(game["igdbId"])
|
||||
self.notify(f"Added {game['name']} ({game['status']})")
|
||||
|
||||
self.run_in_background(
|
||||
"Looking up the game on IGDB...",
|
||||
lambda: games.add_game(values["link"], values["name"], values["status"], values["rating"], values["image"]),
|
||||
added,
|
||||
)
|
||||
|
||||
def action_rate(self):
|
||||
game = self.selected_game()
|
||||
if not game:
|
||||
return
|
||||
current = "" if game.get("rating") is None else str(game["rating"])
|
||||
fields = [FormField("rating", f"Rating for {game['name']} (empty to clear)", kind="rating", value=current)]
|
||||
|
||||
def rated(values: dict | None):
|
||||
if values is not None:
|
||||
self.run_now(lambda: games.rate_game(game["igdbId"], values["rating"]), f"Updated {game['name']}", game["igdbId"])
|
||||
|
||||
self.app.push_screen(FormDialog("Change rating", fields), rated)
|
||||
|
||||
def action_status(self):
|
||||
game = self.selected_game()
|
||||
if not game:
|
||||
return
|
||||
fields = [FormField("status", f"Status for {game['name']}", kind="select", options=games.STATUSES, value=game["status"])]
|
||||
|
||||
def changed(values: dict | None):
|
||||
if values:
|
||||
status = values["status"]
|
||||
self.run_now(lambda: games.set_status(game["igdbId"], status), f"{game['name']} is now {status}", game["igdbId"])
|
||||
|
||||
self.app.push_screen(FormDialog("Change status", fields, "Save"), changed)
|
||||
|
||||
def action_links(self):
|
||||
game = self.selected_game()
|
||||
if not game:
|
||||
return
|
||||
fields = [
|
||||
FormField("url", "Store link (empty to remove)", value=game.get("url") or ""),
|
||||
FormField("image", "New image (site path or URL, empty keeps the current one)", placeholder="/img/minecraft.png"),
|
||||
]
|
||||
|
||||
def saved(values: dict | None):
|
||||
if values is not None:
|
||||
self.run_now(lambda: games.set_links(game["igdbId"], values["url"], values["image"]), f"Saved links for {game['name']}", game["igdbId"])
|
||||
|
||||
self.app.push_screen(FormDialog("Store link and image", fields), saved)
|
||||
|
||||
def action_toggle(self, flag: str):
|
||||
game = self.selected_game()
|
||||
if not game:
|
||||
return
|
||||
label = games.PROFILE_FLAGS[flag]
|
||||
message = f"{game['name']} is no longer {label}" if game.get(flag) else f"{game['name']} is now {label}"
|
||||
self.run_now(lambda: games.toggle_flag(game, flag), message, game["igdbId"])
|
||||
|
||||
def action_tags(self):
|
||||
game = self.selected_game()
|
||||
if not game:
|
||||
return
|
||||
fields = [
|
||||
FormField("tags", "Tags, separated by commas", value=", ".join(game.get("tags") or []), placeholder=", ".join(games.SUGGESTED_TAGS)),
|
||||
FormField("note", "Why you like it (shown with your favorite)", value=game.get("note") or ""),
|
||||
]
|
||||
|
||||
def saved(values: dict | None):
|
||||
if values is not None:
|
||||
self.run_now(lambda: games.set_tags(game["igdbId"], values["tags"], values["note"]), f"Saved tags for {game['name']}", game["igdbId"])
|
||||
|
||||
self.app.push_screen(FormDialog("Tags and note", fields), saved)
|
||||
|
||||
def action_dates(self):
|
||||
game = self.selected_game()
|
||||
if not game:
|
||||
return
|
||||
identifier = game["igdbId"]
|
||||
self.edit_dates(game["name"], game, lambda started, finished: games.set_dates(identifier, started, finished), identifier)
|
||||
|
||||
def action_delete(self):
|
||||
game = self.selected_game()
|
||||
if not game:
|
||||
return
|
||||
|
||||
def confirmed(should_delete: bool | None):
|
||||
if should_delete:
|
||||
self.run_now(lambda: games.remove_game(game["igdbId"]), f"Removed {game['name']}")
|
||||
|
||||
self.app.push_screen(ConfirmDialog(f"Remove {game['name']}?"), confirmed)
|
||||
|
||||
def action_refresh(self):
|
||||
game = self.selected_game()
|
||||
if not game:
|
||||
return
|
||||
|
||||
def refreshed(_saved: dict):
|
||||
self.reload(game["igdbId"])
|
||||
self.notify(f"Refreshed {game['name']} from IGDB")
|
||||
|
||||
self.run_in_background(f"Refreshing {game['name']}...", lambda: games.refresh_game(game), refreshed)
|
||||
Reference in new issue
Block a user