diff --git a/CHANGELOG.md b/CHANGELOG.md index 8f004e7..736be6f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ Notable changes to UnityScraper are documented here. The project follows ### Added +- Offline XboxUnity title catalog with background refresh, sync history, + TitleID/name autocomplete, and a manual/CLI refresh path. +- Additive schema migration 5 for cached XboxUnity titles and catalog sync + runs. - Versioned additive migrations for collection snapshots, preservation matches, repair plans, console inventories, resumable jobs, overrides, and recovery state. @@ -35,6 +39,11 @@ Notable changes to UnityScraper are documented here. The project follows ### Changed +- Library rows now show `Unknown game` instead of duplicating the TitleID when + no real game name is known. +- Cached XboxUnity names enrich only blank, unknown, or TitleID-shaped values + and never replace an existing preferred title. +- Library queries now close SQLite handles immediately after use. - Version advanced to `1.0.0-beta.1`. - Download queues now use atomic writes and recover interrupted items. - Update checks select a platform artifact and require its SHA-256 sidecar diff --git a/README.md b/README.md index 1965e6c..62bca1c 100644 --- a/README.md +++ b/README.md @@ -21,6 +21,7 @@ archives before running large imports or transfers. ### Library and Downloads - Collects XboxUnity cover and Title Update metadata. +- Caches the XboxUnity title catalog for offline name and TitleID autocomplete. - Reviews results before selectively downloading files. - Tracks pending, downloaded, failed, and verified content. - Supports retries, rate limiting, bandwidth limits, and resumable downloads. @@ -128,7 +129,7 @@ Linux source setup: | Workspace | Purpose | | --- | --- | | Library | Browse games, covers, MediaIDs, and available updates | -| Add Games | Import or enter TitleIDs | +| Add Games | Search cached game names, select TitleIDs, or import lists | | Downloads | Review and manage download activity | | Backup Manager | Scan, install, verify, export, convert, and transfer owned content | | Collections | Identify storage, compare Title Updates, verify preservation data, and preview repairs | @@ -173,6 +174,9 @@ python main.py --help ### XboxUnity Metadata ```powershell +# Refresh every XboxUnity title name for local autocomplete +python main.py --sync-title-catalog + # Collect metadata for one or more TitleIDs python main.py 4D5307E6 --metadata-only @@ -186,6 +190,15 @@ python main.py --verify-integrity Providing TitleIDs without `--metadata-only` starts the download workflow. Review the destination and settings before doing this. +The desktop application refreshes the XboxUnity title catalog in the +background when the local copy is missing or more than seven days old. The +**Add Games** search box always queries SQLite, so suggestions remain fast and +available offline. Suggestions include the game name, TitleID, and content +type. Use **Refresh Catalog** on that page to request an immediate update. + +Catalog names fill only blank, unknown, or accidentally TitleID-shaped game +names. Existing user names and better source-attributed metadata are preserved. + ### Knowledge Sources ```powershell diff --git a/database.py b/database.py index 482cc7b..96f8f60 100644 --- a/database.py +++ b/database.py @@ -170,6 +170,7 @@ def add_titleid(self, titleid: str, name: Optional[str] = None, # Update search index self._update_search_index(conn, titleid, name, publisher, metadata) self._enrich_unknown_titleid_metadata(conn, titleid) + self._enrich_unknown_titleid_from_catalog(conn, titleid) logger.info(f"Added/updated TitleID: {titleid}") return True @@ -243,6 +244,43 @@ def _enrich_unknown_titleid_metadata(self, conn, titleid: str) -> int: ) self._update_search_index(conn, titleid, new_name, new_publisher, metadata) return 1 + + def _enrich_unknown_titleid_from_catalog(self, conn, titleid: str) -> int: + """Use the cached XboxUnity title only when the library name is unknown.""" + row = conn.execute( + """ + SELECT t.name, t.publisher, t.metadata, c.name AS catalog_name + FROM titleids AS t + LEFT JOIN xboxunity_title_catalog AS c ON c.titleid = t.titleid + WHERE t.titleid = ? + """, + (titleid,), + ).fetchone() + if not row or not row["catalog_name"]: + return 0 + current_name = row["name"] + if not (is_unknown(current_name) or str(current_name).upper() == titleid.upper()): + return 0 + + metadata = {} + if row["metadata"]: + try: + metadata = json.loads(row["metadata"]) + except json.JSONDecodeError: + metadata = {} + metadata["title_source"] = "XboxUnity title catalog" + conn.execute( + "UPDATE titleids SET name = ?, metadata = ? WHERE titleid = ?", + (row["catalog_name"], json.dumps(metadata, sort_keys=True), titleid), + ) + self._update_search_index( + conn, + titleid, + row["catalog_name"], + row["publisher"], + metadata, + ) + return 1 def _update_search_index(self, conn, titleid: str, name: Optional[str] = None, publisher: Optional[str] = None, metadata: Optional[Dict] = None): diff --git a/database_migrations.py b/database_migrations.py index cbe6b62..c140d25 100644 --- a/database_migrations.py +++ b/database_migrations.py @@ -8,7 +8,7 @@ from pathlib import Path -SCHEMA_VERSION = 4 +SCHEMA_VERSION = 5 def _now() -> str: @@ -65,6 +65,7 @@ def ensure_application_schema(connection: sqlite3.Connection) -> int: (2, "preservation records", _migration_preservation), (3, "console synchronization", _migration_console_sync), (4, "user overrides and recovery", _migration_reliability), + (5, "XboxUnity title catalog", _migration_xboxunity_catalog), ) for version, name, migration in migrations: if version in applied: @@ -240,3 +241,40 @@ def _migration_reliability(connection: sqlite3.Connection) -> None: ); """ ) + + +def _migration_xboxunity_catalog(connection: sqlite3.Connection) -> None: + connection.executescript( + """ + CREATE TABLE IF NOT EXISTS xboxunity_title_catalog ( + titleid TEXT PRIMARY KEY, + name TEXT NOT NULL, + hb_titleid TEXT, + title_type TEXT, + link_enabled INTEGER NOT NULL DEFAULT 0, + covers_count INTEGER NOT NULL DEFAULT 0, + updates_count INTEGER NOT NULL DEFAULT 0, + media_id_count INTEGER NOT NULL DEFAULT 0, + user_count INTEGER NOT NULL DEFAULT 0, + newest_content TEXT, + source_url TEXT NOT NULL, + raw_json TEXT NOT NULL, + fetched_at TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_xboxunity_catalog_name + ON xboxunity_title_catalog(name COLLATE NOCASE); + CREATE INDEX IF NOT EXISTS idx_xboxunity_catalog_type + ON xboxunity_title_catalog(title_type); + + CREATE TABLE IF NOT EXISTS xboxunity_catalog_sync_runs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + started_at TEXT NOT NULL, + completed_at TEXT, + status TEXT NOT NULL, + pages_expected INTEGER NOT NULL DEFAULT 0, + pages_fetched INTEGER NOT NULL DEFAULT 0, + items_upserted INTEGER NOT NULL DEFAULT 0, + error_message TEXT + ); + """ + ) diff --git a/library_service.py b/library_service.py index 89f93ba..5fbbf67 100644 --- a/library_service.py +++ b/library_service.py @@ -10,6 +10,7 @@ import hashlib import json import sqlite3 +from contextlib import closing from dataclasses import dataclass from pathlib import Path from typing import Any, Iterable, Optional @@ -52,7 +53,12 @@ def list_games(self, search: str = "") -> list[GameSummary]: query = """ SELECT t.titleid, - COALESCE(NULLIF(t.name, ''), t.titleid) AS name, + CASE + WHEN t.name IS NULL OR TRIM(t.name) = '' + OR UPPER(TRIM(t.name)) = UPPER(t.titleid) + THEN 'Unknown game' + ELSE t.name + END AS name, COALESCE(t.publisher, '') AS publisher, COALESCE(t.last_scraped, '') AS last_scraped, COUNT(DISTINCT c.id) AS covers_total, @@ -83,7 +89,7 @@ def list_games(self, search: str = "") -> list[GameSummary]: ORDER BY name COLLATE NOCASE, t.titleid """ - with self._connect() as connection: + with closing(self._connect()) as connection: rows = connection.execute(query, parameters).fetchall() return [ @@ -106,7 +112,7 @@ def get_game_details(self, titleid: str) -> dict[str, Any]: if not self.database_path.exists(): return {} - with self._connect() as connection: + with closing(self._connect()) as connection: title = connection.execute( "SELECT * FROM titleids WHERE titleid = ?", (titleid,), @@ -174,7 +180,7 @@ def get_dashboard_counts(self) -> dict[str, int]: ), } - with self._connect() as connection: + with closing(self._connect()) as connection: return { name: int(connection.execute(statement).fetchone()[0] or 0) for name, statement in sql.items() @@ -191,7 +197,7 @@ def find_database_duplicates(self) -> list[dict[str, Any]]: if not self.database_path.exists(): return [] - with self._connect() as connection: + with closing(self._connect()) as connection: update_rows = connection.execute( """ SELECT @@ -236,7 +242,7 @@ def scan_archive_health(self) -> dict[str, Any]: if not self.database_path.exists(): return report - with self._connect() as connection: + with closing(self._connect()) as connection: rows = connection.execute( """ SELECT 'cover' AS item_type, id, titleid, file_path, file_size diff --git a/main.py b/main.py index 9994f1d..ea17c61 100644 --- a/main.py +++ b/main.py @@ -791,6 +791,11 @@ def main(): 'environment variable' ) ) + parser.add_argument( + '--sync-title-catalog', + action='store_true', + help='Refresh the local XboxUnity title-name catalog for offline autocomplete' + ) parser.add_argument( '--sync-knowledge', action='store_true', @@ -948,6 +953,28 @@ def main(): logger.info("Configuration saved") # Initialize scraper + if args.sync_title_catalog: + try: + from title_catalog import XboxUnityTitleCatalog + + DatabaseManager() + summary = XboxUnityTitleCatalog( + request_interval=config.rate_limit, + timeout=config.timeout, + ).sync( + progress=lambda page, pages, items: logger.info( + "XboxUnity catalog page %s/%s (%s titles)", + page, + pages, + items, + ) + ) + logger.info("XboxUnity title catalog sync completed: %s", summary) + sys.exit(0) + except Exception as e: + logger.error("XboxUnity title catalog sync failed: %s", e) + sys.exit(1) + if args.sync_knowledge: try: from knowledge_sync import sync_consolemods_knowledge diff --git a/modern_gui.py b/modern_gui.py index 772185c..834be92 100644 --- a/modern_gui.py +++ b/modern_gui.py @@ -8,14 +8,16 @@ from __future__ import annotations import json +import queue import sqlite3 +import threading import tkinter as tk import webbrowser from pathlib import Path from tkinter import filedialog, messagebox, ttk +from typing import Any, Callable from PIL import Image, ImageOps, ImageTk -from typing import Any from app_paths import ( BASE_DIR, @@ -43,6 +45,7 @@ from library_service import GameSummary, LibraryService from platform_support import desktop_font_family, open_path from setup_wizard import run_first_run_wizard +from title_catalog import TitleSuggestion, XboxUnityTitleCatalog from updater import VersionChecker @@ -175,6 +178,124 @@ def _cover_resize(image: Image.Image, width: int, height: int) -> Image.Image: return background +class TitleAutocomplete(ttk.Frame): + """Entry with a non-blocking local title suggestion popup.""" + + def __init__( + self, + parent: tk.Misc, + search: Callable[[str], list[TitleSuggestion]], + on_select: Callable[[TitleSuggestion], None], + ) -> None: + super().__init__(parent) + self.search = search + self.on_select = on_select + self.variable = tk.StringVar() + self.suggestions: list[TitleSuggestion] = [] + self._search_job: str | None = None + + self.entry = ttk.Entry(self, textvariable=self.variable) + self.entry.pack(fill=tk.X, expand=True) + self.entry.bind("", self._key_released) + self.entry.bind("", self._focus_suggestions) + self.entry.bind("", self._accept_first) + self.entry.bind("", lambda _event: self.hide()) + self.entry.bind("", self._queue_hide) + + self.popup = tk.Toplevel(self) + self.popup.withdraw() + self.popup.overrideredirect(True) + self.popup.configure(background=BORDER) + self.popup.transient(self.winfo_toplevel()) + self.listbox = tk.Listbox( + self.popup, + height=8, + background="#070b08", + foreground=TEXT, + selectbackground="#315f12", + selectforeground=TEXT, + highlightthickness=1, + highlightbackground=BORDER, + relief=tk.FLAT, + font=(UI_FONT, 10), + activestyle="none", + ) + self.listbox.pack(fill=tk.BOTH, expand=True) + self.listbox.bind("", self._accept) + self.listbox.bind("", self._accept) + self.listbox.bind("", self._accept) + self.listbox.bind("", lambda _event: self.hide()) + self.listbox.bind("", self._queue_hide) + + def _key_released(self, event: tk.Event[Any]) -> None: + if event.keysym in {"Up", "Down", "Return", "Escape"}: + return + if self._search_job is not None: + self.after_cancel(self._search_job) + self._search_job = self.after(120, self._refresh) + + def _refresh(self) -> None: + self._search_job = None + self.suggestions = self.search(self.variable.get()) + self.listbox.delete(0, tk.END) + for suggestion in self.suggestions: + self.listbox.insert(tk.END, suggestion.label) + if self.suggestions: + self._show_popup() + else: + self.hide() + + def _show_popup(self) -> None: + self.update_idletasks() + width = max(self.entry.winfo_width(), 460) + x = self.entry.winfo_rootx() + y = self.entry.winfo_rooty() + self.entry.winfo_height() + height = min(8, len(self.suggestions)) * 26 + 2 + self.popup.geometry(f"{width}x{height}+{x}+{y}") + self.popup.deiconify() + self.popup.lift() + + def _focus_suggestions(self, _event: tk.Event[Any]) -> str: + if not self.suggestions: + self._refresh() + if self.suggestions: + self.listbox.selection_clear(0, tk.END) + self.listbox.selection_set(0) + self.listbox.activate(0) + self.listbox.focus_set() + return "break" + + def _accept(self, _event: tk.Event[Any] | None = None) -> str: + selection = self.listbox.curselection() + if selection: + suggestion = self.suggestions[int(selection[0])] + self.variable.set(suggestion.label) + self.on_select(suggestion) + self.hide() + self.entry.focus_set() + return "break" + + def _accept_first(self, _event: tk.Event[Any]) -> str: + if not self.suggestions: + self._refresh() + if self.suggestions: + self.listbox.selection_clear(0, tk.END) + self.listbox.selection_set(0) + return self._accept() + return "break" + + def _queue_hide(self, _event: tk.Event[Any]) -> None: + self.after(100, self._hide_unfocused) + + def _hide_unfocused(self) -> None: + focused = self.focus_get() + if focused not in {self.entry, self.listbox}: + self.hide() + + def hide(self) -> None: + self.popup.withdraw() + + class UnityScraperDesktop: """Main multi-page application window.""" @@ -185,7 +306,10 @@ def __init__(self, root: tk.Tk) -> None: self.backups = BackupService() self.collections = CollectionIntelligenceService() self.database = DatabaseManager() + self.title_catalog = XboxUnityTitleCatalog() self.current_game: str | None = None + self._catalog_syncing = False + self._catalog_events: queue.Queue[tuple[str, Any]] = queue.Queue() ensure_app_dirs() ensure_user_titleids_file() @@ -202,6 +326,7 @@ def __init__(self, root: tk.Tk) -> None: if run_first_run_wizard(self.root): self.refresh_library() + self.root.after(750, self._start_catalog_sync_if_stale) else: self.root.after(0, self.root.destroy) @@ -585,7 +710,9 @@ def _load_game(self, titleid: str) -> None: return title = details["title"] - name = title.get("name") or titleid + name = title.get("name") + if not name or str(name).strip().upper() == titleid.upper(): + name = "Unknown game" publisher = title.get("publisher") or "Unknown publisher" self.detail_title.configure(text=f"{name} • {titleid} • {publisher}") @@ -639,14 +766,43 @@ def show_add_games(self) -> None: ttk.Label( panel, - text="Enter one or more 8-character hexadecimal TitleIDs:", + text="Search the cached XboxUnity title catalog:", + ).grid(row=0, column=0, sticky=tk.W) + + self.title_autocomplete = TitleAutocomplete( + panel, + self.title_catalog.search, + self._accept_title_suggestion, + ) + self.title_autocomplete.grid(row=1, column=0, sticky="ew", pady=(6, 12)) + + catalog_row = ttk.Frame(panel) + catalog_row.grid(row=2, column=0, sticky="ew", pady=(0, 14)) + catalog_row.columnconfigure(0, weight=1) + self.catalog_status_var = tk.StringVar( + value=f"{self.title_catalog.count():,} XboxUnity titles cached locally" + ) + ttk.Label( + catalog_row, + textvariable=self.catalog_status_var, + foreground=MUTED, ).grid(row=0, column=0, sticky=tk.W) + ttk.Button( + catalog_row, + text="Refresh Catalog", + command=self._start_catalog_sync, + ).grid(row=0, column=1, sticky=tk.E) + + ttk.Label( + panel, + text="Selected TitleIDs:", + ).grid(row=3, column=0, sticky=tk.W) self.add_titleids_text = tk.Text(panel, height=8, wrap=tk.WORD, background="#070b08", foreground=TEXT, insertbackground=ACCENT, selectbackground="#315f12", selectforeground=TEXT, relief=tk.FLAT, highlightthickness=1, highlightbackground=BORDER, highlightcolor=ACCENT) - self.add_titleids_text.grid(row=1, column=0, sticky="ew", pady=8) + self.add_titleids_text.grid(row=4, column=0, sticky="ew", pady=8) actions = ttk.Frame(panel) - actions.grid(row=2, column=0, sticky="ew") + actions.grid(row=5, column=0, sticky="ew") ttk.Button( actions, text="Import Text File…", command=self._import_titleid_file ).pack(side=tk.LEFT) @@ -661,7 +817,72 @@ def show_add_games(self) -> None: "to scan XboxUnity for covers and title updates." ), wraplength=760, - ).grid(row=3, column=0, sticky=tk.W, pady=(18, 0)) + ).grid(row=6, column=0, sticky=tk.W, pady=(18, 0)) + + def _accept_title_suggestion(self, suggestion: TitleSuggestion) -> None: + current = self.add_titleids_text.get("1.0", tk.END).strip() + separator = "\n" if current else "" + self.add_titleids_text.insert(tk.END, f"{separator}{suggestion.titleid}") + + def _start_catalog_sync_if_stale(self) -> None: + if self.title_catalog.is_stale(): + self._start_catalog_sync(silent=True) + + def _start_catalog_sync(self, silent: bool = False) -> None: + if self._catalog_syncing: + if not silent: + self._set_catalog_status("XboxUnity catalog refresh already running") + return + self._catalog_syncing = True + self._set_catalog_status("Starting XboxUnity catalog refresh...") + + def worker() -> None: + try: + result = self.title_catalog.sync( + progress=lambda page, pages, items: self._catalog_events.put( + ("progress", (page, pages, items)) + ) + ) + self._catalog_events.put(("completed", result)) + except Exception as exc: + self._catalog_events.put(("failed", str(exc))) + + threading.Thread( + target=worker, + name="xboxunity-catalog-sync", + daemon=True, + ).start() + self.root.after(100, self._poll_catalog_events) + + def _poll_catalog_events(self) -> None: + while True: + try: + event, value = self._catalog_events.get_nowait() + except queue.Empty: + break + if event == "progress": + page, pages, items = value + self._set_catalog_status( + f"Caching XboxUnity titles: page {page}/{pages} ({items:,} titles)" + ) + elif event == "completed": + self._catalog_syncing = False + self._set_catalog_status( + f"{value.items_upserted:,} XboxUnity titles cached locally" + ) + if hasattr(self, "game_tree") and self.game_tree.winfo_exists(): + self.refresh_library() + elif event == "failed": + self._catalog_syncing = False + self._set_catalog_status( + f"Catalog refresh failed; using local cache ({value})" + ) + if self._catalog_syncing: + self.root.after(150, self._poll_catalog_events) + + def _set_catalog_status(self, text: str) -> None: + if hasattr(self, "catalog_status_var"): + self.catalog_status_var.set(text) def _import_titleid_file(self) -> None: selected = filedialog.askopenfilename( diff --git a/tests.py b/tests.py index 35bb62f..1b81da9 100644 --- a/tests.py +++ b/tests.py @@ -26,6 +26,8 @@ from dat_adapters import parse_dat from knowledge_service import KnowledgeService from knowledge_sources import KnowledgeImportService, SourceInfo +from library_service import LibraryService +from title_catalog import XboxUnityTitleCatalog from wiki_adapters import extract_article_text, parse_sitemap from backup_manager import ( BackupItem, @@ -444,6 +446,107 @@ def test_enrich_unknown_metadata_from_knowledge(self): self.assertEqual(info["publisher"], "User Publisher") +class TestXboxUnityTitleCatalog(unittest.TestCase): + """Test persistent, HTTP-only XboxUnity title autocomplete data.""" + + def setUp(self): + self.temp_dir = tempfile.mkdtemp() + self.db_path = Path(self.temp_dir) / "catalog.db" + self.database = DatabaseManager(self.db_path) + + def tearDown(self): + shutil.rmtree(self.temp_dir) + + @staticmethod + def _response(items, *, pages=1, page=0): + response = Mock() + response.url = ( + "http://xboxunity.net/Resources/Lib/TitleList.php" + f"?category=0&count=100&page={page}" + ) + response.raise_for_status.return_value = None + response.json.return_value = { + "Items": items, + "Count": len(items), + "Pages": pages, + "Page": page, + } + return response + + def test_sync_caches_every_page_and_searches_name_or_titleid(self): + session = Mock() + session.get.side_effect = [ + self._response( + [ + { + "TitleID": "4D5307E6", + "Name": "Halo 3", + "TitleType": "360", + "Covers": "40", + "Updates": "12", + } + ], + pages=2, + page=0, + ), + self._response( + [ + { + "TitleID": "4D53085B", + "Name": "Halo: Reach", + "TitleType": "360", + "Covers": "28", + "Updates": "11", + } + ], + pages=2, + page=1, + ), + ] + catalog = XboxUnityTitleCatalog( + self.db_path, + session=session, + request_interval=0, + ) + + result = catalog.sync() + + self.assertEqual(result.pages_fetched, 2) + self.assertEqual(catalog.count(), 2) + self.assertEqual(catalog.search("reach")[0].titleid, "4D53085B") + self.assertEqual(catalog.search("4D5307")[0].name, "Halo 3") + self.assertTrue(session.get.call_args_list[0].args[0].startswith("http://")) + + def test_catalog_enrichment_never_replaces_a_known_name(self): + catalog = XboxUnityTitleCatalog(self.db_path) + catalog._store_page( + [ + {"TitleID": "4D5307E6", "Name": "Halo 3", "TitleType": "360"}, + {"TitleID": "4D53085B", "Name": "Halo: Reach", "TitleType": "360"}, + ], + "http://xboxunity.net/Resources/Lib/TitleList.php?page=0", + ) + self.database.add_titleid("4D5307E6", name="4D5307E6") + self.database.add_titleid("4D53085B", name="My Preferred Reach Name") + + self.assertEqual(self.database.get_titleid_info("4D5307E6")["name"], "Halo 3") + self.assertEqual( + self.database.get_titleid_info("4D53085B")["name"], + "My Preferred Reach Name", + ) + + def test_library_does_not_display_titleid_as_the_game_name(self): + self.database.add_titleid("555308C5") + + games = LibraryService(self.db_path).list_games() + + self.assertEqual(games[0].name, "Unknown game") + + def test_non_http_xboxunity_base_url_is_rejected(self): + with self.assertRaises(ValueError): + XboxUnityTitleCatalog(self.db_path, base_url="https://xboxunity.net") + + class TestConsoleModsAdapters(unittest.TestCase): """Test ConsoleMods parsing helpers.""" @@ -1047,11 +1150,12 @@ def test_versioned_migrations_create_all_foundation_tables(self): versions = connection.execute( "SELECT version FROM app_schema_migrations ORDER BY version" ).fetchall() - self.assertEqual([row[0] for row in versions], [1, 2, 3, 4]) + self.assertEqual([row[0] for row in versions], [1, 2, 3, 4, 5]) self.assertIn("collection_snapshots", tables) self.assertIn("preservation_matches", tables) self.assertIn("console_transfer_jobs", tables) self.assertIn("metadata_overrides", tables) + self.assertIn("xboxunity_title_catalog", tables) def test_xex_execution_info_is_parsed(self): from backup_manager import inspect_xex @@ -1155,6 +1259,7 @@ def run_tests(): suite.addTests(loader.loadTestsFromTestCase(TestRateLimiter)) suite.addTests(loader.loadTestsFromTestCase(TestUnityScraper)) suite.addTests(loader.loadTestsFromTestCase(TestDatabaseManager)) + suite.addTests(loader.loadTestsFromTestCase(TestXboxUnityTitleCatalog)) suite.addTests(loader.loadTestsFromTestCase(TestConsoleModsAdapters)) suite.addTests(loader.loadTestsFromTestCase(TestKnowledgeApplication)) suite.addTests(loader.loadTestsFromTestCase(TestDownloadProgress)) diff --git a/title_catalog.py b/title_catalog.py new file mode 100644 index 0000000..390f2d8 --- /dev/null +++ b/title_catalog.py @@ -0,0 +1,376 @@ +"""Persistent XboxUnity title catalog and offline autocomplete queries.""" + +from __future__ import annotations + +import json +import sqlite3 +import time +from contextlib import contextmanager +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any, Callable, Generator, Iterable + +import requests + +from app_paths import DATABASE_PATH +from database_migrations import ensure_application_schema +from knowledge_base import is_unknown + + +XBOXUNITY_BASE_URL = "http://xboxunity.net" +TITLE_LIST_PATH = "/Resources/Lib/TitleList.php" + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _as_int(value: Any) -> int: + try: + return int(value) + except (TypeError, ValueError): + return 0 + + +@dataclass(frozen=True) +class TitleSuggestion: + """One locally cached title shown by autocomplete.""" + + titleid: str + name: str + title_type: str + + @property + def label(self) -> str: + suffix = f" [{self.title_type}]" if self.title_type else "" + return f"{self.name} - {self.titleid}{suffix}" + + +@dataclass(frozen=True) +class CatalogSyncResult: + """Summary of a completed XboxUnity catalog refresh.""" + + pages_fetched: int + items_upserted: int + library_names_enriched: int + + +class XboxUnityTitleCatalog: + """Sync XboxUnity's paginated title list and query it without network access.""" + + def __init__( + self, + database_path: Path | str = DATABASE_PATH, + *, + session: requests.Session | None = None, + base_url: str = XBOXUNITY_BASE_URL, + request_interval: float = 0.35, + timeout: float = 30, + sleep: Callable[[float], None] = time.sleep, + ) -> None: + if not base_url.startswith("http://"): + raise ValueError("XboxUnity catalog access must remain HTTP-only") + self.database_path = Path(database_path) + self.session = session or requests.Session() + self.base_url = base_url.rstrip("/") + self.request_interval = max(0.0, request_interval) + self.timeout = timeout + self.sleep = sleep + self._ensure_schema() + + def _connect(self) -> sqlite3.Connection: + self.database_path.parent.mkdir(parents=True, exist_ok=True) + connection = sqlite3.connect(self.database_path, timeout=30) + connection.row_factory = sqlite3.Row + return connection + + @contextmanager + def _connection(self) -> Generator[sqlite3.Connection, None, None]: + connection = self._connect() + try: + yield connection + connection.commit() + except Exception: + connection.rollback() + raise + finally: + connection.close() + + def _ensure_schema(self) -> None: + with self._connection() as connection: + ensure_application_schema(connection) + + def count(self) -> int: + with self._connection() as connection: + row = connection.execute( + "SELECT COUNT(*) AS count FROM xboxunity_title_catalog" + ).fetchone() + return int(row["count"]) + + def is_stale(self, max_age_days: int = 7) -> bool: + """Return whether no successful, sufficiently recent sync exists.""" + cutoff = datetime.now(timezone.utc) - timedelta(days=max_age_days) + with self._connection() as connection: + row = connection.execute( + """ + SELECT completed_at + FROM xboxunity_catalog_sync_runs + WHERE status = 'completed' + ORDER BY id DESC + LIMIT 1 + """ + ).fetchone() + if row is None or not row["completed_at"]: + return True + try: + completed = datetime.fromisoformat(row["completed_at"]) + except ValueError: + return True + if completed.tzinfo is None: + completed = completed.replace(tzinfo=timezone.utc) + return completed < cutoff + + def search(self, query: str, limit: int = 12) -> list[TitleSuggestion]: + """Search cached names and TitleIDs, ranking prefixes ahead of substrings.""" + value = query.strip() + if not value: + return [] + lowered = value.lower() + contains = f"%{lowered}%" + prefix = f"{lowered}%" + with self._connection() as connection: + rows = connection.execute( + """ + SELECT titleid, name, COALESCE(title_type, '') AS title_type + FROM xboxunity_title_catalog + WHERE LOWER(titleid) LIKE ? OR LOWER(name) LIKE ? + ORDER BY + CASE + WHEN LOWER(titleid) = ? THEN 0 + WHEN LOWER(titleid) LIKE ? THEN 1 + WHEN LOWER(name) LIKE ? THEN 2 + ELSE 3 + END, + name COLLATE NOCASE, + titleid + LIMIT ? + """, + (contains, contains, lowered, prefix, prefix, max(1, limit)), + ).fetchall() + return [ + TitleSuggestion(row["titleid"], row["name"], row["title_type"]) + for row in rows + ] + + def sync( + self, + *, + progress: Callable[[int, int, int], None] | None = None, + page_size: int = 100, + ) -> CatalogSyncResult: + """Refresh every title page, preserving the last usable cache on failure.""" + started_at = _now() + with self._connection() as connection: + cursor = connection.execute( + """ + INSERT INTO xboxunity_catalog_sync_runs(started_at, status) + VALUES (?, 'running') + """, + (started_at,), + ) + run_id = _as_int(cursor.lastrowid) + if run_id <= 0: + raise RuntimeError("Could not create XboxUnity catalog sync run") + + pages_expected = 0 + pages_fetched = 0 + items_upserted = 0 + try: + page = 0 + while page == 0 or page < pages_expected: + payload, source_url = self._fetch_page(page, page_size) + pages_expected = max(1, _as_int(payload.get("Pages"))) + items = payload.get("Items", []) + if not isinstance(items, list): + raise ValueError("XboxUnity title list returned invalid Items data") + items_upserted += self._store_page(items, source_url) + pages_fetched += 1 + if progress: + progress(pages_fetched, pages_expected, items_upserted) + page += 1 + if page < pages_expected and self.request_interval: + self.sleep(self.request_interval) + + with self._connection() as connection: + connection.execute( + "DELETE FROM xboxunity_title_catalog WHERE fetched_at < ?", + (started_at,), + ) + enriched = self.enrich_library_names() + with self._connection() as connection: + connection.execute( + """ + UPDATE xboxunity_catalog_sync_runs + SET completed_at = ?, status = 'completed', + pages_expected = ?, pages_fetched = ?, items_upserted = ? + WHERE id = ? + """, + ( + _now(), + pages_expected, + pages_fetched, + items_upserted, + run_id, + ), + ) + return CatalogSyncResult(pages_fetched, items_upserted, enriched) + except Exception as exc: + with self._connection() as connection: + connection.execute( + """ + UPDATE xboxunity_catalog_sync_runs + SET completed_at = ?, status = 'failed', + pages_expected = ?, pages_fetched = ?, + items_upserted = ?, error_message = ? + WHERE id = ? + """, + ( + _now(), + pages_expected, + pages_fetched, + items_upserted, + str(exc)[:1000], + run_id, + ), + ) + raise + + def _fetch_page(self, page: int, page_size: int) -> tuple[dict[str, Any], str]: + url = f"{self.base_url}{TITLE_LIST_PATH}" + params: dict[str, str | int] = { + "category": 0, + "count": min(max(page_size, 10), 100), + "direction": 1, + "filter": 0, + "page": page, + "search": "", + "sort": 3, + } + response = self.session.get(url, params=params, timeout=self.timeout) + response.raise_for_status() + payload = response.json() + if not isinstance(payload, dict): + raise ValueError("XboxUnity title list returned a non-object response") + return payload, response.url + + def _store_page(self, items: Iterable[Any], source_url: str) -> int: + fetched_at = _now() + records: list[tuple[Any, ...]] = [] + for item in items: + if not isinstance(item, dict): + continue + titleid = str(item.get("TitleID", "")).strip().upper() + name = str(item.get("Name", "")).strip() + if len(titleid) != 8 or not name: + continue + records.append( + ( + titleid, + name, + str(item.get("HBTitleID", "")).strip().upper() or None, + str(item.get("TitleType", "")).strip() or None, + _as_int(item.get("LinkEnabled")), + _as_int(item.get("Covers")), + _as_int(item.get("Updates")), + _as_int(item.get("MediaIDCount")), + _as_int(item.get("UserCount")), + str(item.get("NewestContent", "")).strip() or None, + source_url, + json.dumps(item, sort_keys=True), + fetched_at, + ) + ) + if not records: + return 0 + with self._connection() as connection: + connection.executemany( + """ + INSERT INTO xboxunity_title_catalog( + titleid, name, hb_titleid, title_type, link_enabled, + covers_count, updates_count, media_id_count, user_count, + newest_content, source_url, raw_json, fetched_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(titleid) DO UPDATE SET + name = excluded.name, + hb_titleid = excluded.hb_titleid, + title_type = excluded.title_type, + link_enabled = excluded.link_enabled, + covers_count = excluded.covers_count, + updates_count = excluded.updates_count, + media_id_count = excluded.media_id_count, + user_count = excluded.user_count, + newest_content = excluded.newest_content, + source_url = excluded.source_url, + raw_json = excluded.raw_json, + fetched_at = excluded.fetched_at + """, + records, + ) + return len(records) + + def enrich_library_names(self) -> int: + """Fill only missing, unknown, or TitleID-shaped library names.""" + changed = 0 + with self._connection() as connection: + rows = connection.execute( + """ + SELECT + t.titleid, + t.name AS current_name, + t.publisher, + t.metadata, + c.name AS catalog_name + FROM titleids AS t + JOIN xboxunity_title_catalog AS c ON c.titleid = t.titleid + """ + ).fetchall() + for row in rows: + current = row["current_name"] + if not (is_unknown(current) or str(current).upper() == row["titleid"]): + continue + metadata = {} + if row["metadata"]: + try: + metadata = json.loads(row["metadata"]) + except json.JSONDecodeError: + metadata = {} + metadata["title_source"] = "XboxUnity title catalog" + connection.execute( + "UPDATE titleids SET name = ?, metadata = ? WHERE titleid = ?", + ( + row["catalog_name"], + json.dumps(metadata, sort_keys=True), + row["titleid"], + ), + ) + search_parts = [ + row["titleid"], + row["catalog_name"], + row["publisher"] or "", + *(str(value) for value in metadata.values() if value), + ] + connection.execute( + """ + INSERT INTO search_index(titleid, search_text) + VALUES (?, ?) + ON CONFLICT(titleid) DO UPDATE SET + search_text = excluded.search_text + """, + ( + row["titleid"], + " ".join(search_parts).lower(), + ), + ) + changed += 1 + return changed