From ee60bfd43a76c2afa9b1dce56dedfbcceeadc2ed Mon Sep 17 00:00:00 2001 From: Sthornberry9 <46094434+Sthornberry9@users.noreply.github.com> Date: Tue, 11 Aug 2026 05:26:44 -0400 Subject: [PATCH] Move library and tool implementations into domains --- MODULARIZATION_PLAN.md | 11 +- external_tools.py | 251 ++---------------- external_tools_gui.py | 6 +- library_service.py | 329 +----------------------- main.py | 8 +- modern_gui.py | 2 +- tests.py | 26 +- tool_catalog.py | 322 ++--------------------- unityscraper/domains/library/models.py | 20 +- unityscraper/domains/library/service.py | 324 ++++++++++++++++++++++- unityscraper/domains/tools/catalog.py | 290 ++++++++++++++++++++- unityscraper/domains/tools/models.py | 58 ++++- unityscraper/domains/tools/runner.py | 226 +++++++++++++++- 13 files changed, 970 insertions(+), 903 deletions(-) diff --git a/MODULARIZATION_PLAN.md b/MODULARIZATION_PLAN.md index 6de4c85..4cc564d 100644 --- a/MODULARIZATION_PLAN.md +++ b/MODULARIZATION_PLAN.md @@ -97,9 +97,14 @@ domain/ - `unityscraper.domains.backups.migrations` owns the backup schema function; `backup_service.ensure_backup_schema` remains import-compatible. - `unityscraper.domains.profiles` exposes profile/save models and helpers. -- `unityscraper.domains.knowledge`, `unityscraper.domains.library`, and - `unityscraper.domains.tools` expose first-class model, repository, catalog, - and runner modules around their existing implementations. +- `unityscraper.domains.knowledge` exposes first-class model and repository + adapters around its existing implementations. +- `unityscraper.domains.library` owns the read-only library model and query + service. Its XboxUnity title catalog remains behind a package adapter until + the catalog migration and knowledge dependencies move into package modules. +- `unityscraper.domains.tools` owns its data models, catalog, executable + discovery, command construction, and process runner. `tool_catalog.py` and + `external_tools.py` remain compatibility wrappers. ## Feature Ownership diff --git a/external_tools.py b/external_tools.py index fdbd59b..17a182d 100644 --- a/external_tools.py +++ b/external_tools.py @@ -1,237 +1,20 @@ -"""Safe process runner for user-supplied Xbox command-line tools.""" +"""Compatibility wrapper for package-owned external tool execution.""" from __future__ import annotations -import os -import shlex -import subprocess -import threading -import time -from dataclasses import dataclass -from pathlib import Path -from typing import Iterable - - -class ExternalToolError(RuntimeError): - """Raised when an external tool cannot be configured or launched.""" - - -@dataclass(frozen=True) -class ToolResult: - """Captured result from one external tool invocation.""" - - command: tuple[str, ...] - returncode: int - stdout: str - stderr: str - duration_seconds: float - cancelled: bool - - -@dataclass(frozen=True) -class ToolLaunch: - """Details for a detached graphical tool launch.""" - - command: tuple[str, ...] - pid: int - - -def split_arguments(value: str, *, windows: bool | None = None) -> list[str]: - """Split an editable argument template without passing it through a shell.""" - use_windows_rules = os.name == "nt" if windows is None else windows - arguments = shlex.split(value, posix=not use_windows_rules) - if use_windows_rules: - return [ - argument[1:-1] - if len(argument) >= 2 and argument[0] == argument[-1] == '"' - else argument - for argument in arguments - ] - return arguments - - -def format_command(command: Iterable[str], *, windows: bool | None = None) -> str: - """Format an argument vector for display only.""" - values = list(command) - use_windows_rules = os.name == "nt" if windows is None else windows - return subprocess.list2cmdline(values) if use_windows_rules else shlex.join(values) - - -class ExternalToolRunner: - """Run one selected executable at a time without shell interpretation.""" - - def __init__(self) -> None: - self._lock = threading.Lock() - self._process: subprocess.Popen[str] | None = None - self._cancel_requested = False - - def build_command( - self, - executable: str | Path, - argument_template: Iterable[str], - *, - input_path: str | Path | None = None, - output_path: str | Path | None = None, - input_kind: str = "file", - output_kind: str = "optional", - ) -> tuple[str, ...]: - tool = Path(executable).expanduser().resolve() - if not tool.is_file(): - raise ExternalToolError(f"Tool executable was not found: {tool}") - - source = self._resolve_input(input_path, input_kind) - output = self._resolve_output(output_path, output_kind) - arguments: list[str] = [] - for value in argument_template: - if "{input}" in value and source is None: - raise ExternalToolError("This command requires an input file") - if "{output}" in value and output is None: - raise ExternalToolError("This command requires an output path") - arguments.append( - value.replace("{input}", str(source) if source else "") - .replace("{output}", str(output) if output else "") - ) - return (str(tool), *arguments) - - def launch_detached( - self, - executable: str | Path, - argument_template: Iterable[str] = (), - *, - input_path: str | Path | None = None, - output_path: str | Path | None = None, - input_kind: str = "none", - output_kind: str = "none", - ) -> ToolLaunch: - """Launch a GUI utility without waiting for it to exit.""" - command = self.build_command( - executable, - argument_template, - input_path=input_path, - output_path=output_path, - input_kind=input_kind, - output_kind=output_kind, - ) - creation_flags = 0 - if os.name == "nt": - creation_flags = subprocess.CREATE_NEW_PROCESS_GROUP - try: - process = subprocess.Popen( - command, - cwd=Path(command[0]).parent, - shell=False, - creationflags=creation_flags, - close_fds=os.name != "nt", - ) - except OSError as exc: - raise ExternalToolError(f"Could not start external tool: {exc}") from exc - return ToolLaunch(command, process.pid) - - def run( - self, - executable: str | Path, - argument_template: Iterable[str], - *, - input_path: str | Path | None = None, - output_path: str | Path | None = None, - timeout: float = 300, - input_kind: str = "file", - output_kind: str = "optional", - ) -> ToolResult: - command = self.build_command( - executable, - argument_template, - input_path=input_path, - output_path=output_path, - input_kind=input_kind, - output_kind=output_kind, - ) - source = self._resolve_input(input_path, input_kind) - working_directory = source.parent if source else Path(command[0]).parent - creation_flags = subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0 - started = time.monotonic() - - with self._lock: - if self._process is not None: - raise ExternalToolError("Another external tool is already running") - self._cancel_requested = False - try: - self._process = subprocess.Popen( - command, - cwd=working_directory, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - shell=False, - creationflags=creation_flags, - ) - except OSError as exc: - raise ExternalToolError(f"Could not start external tool: {exc}") from exc - process = self._process - - try: - stdout, stderr = process.communicate(timeout=max(1, timeout)) - except subprocess.TimeoutExpired as exc: - process.kill() - stdout, stderr = process.communicate() - raise ExternalToolError( - f"External tool exceeded the {timeout:g}-second timeout" - ) from exc - finally: - with self._lock: - cancelled = self._cancel_requested - self._process = None - - return ToolResult( - command, - process.returncode, - stdout, - stderr, - time.monotonic() - started, - cancelled, - ) - - def cancel(self) -> bool: - """Terminate the active process, returning whether one was running.""" - with self._lock: - if self._process is None: - return False - self._cancel_requested = True - self._process.terminate() - return True - - @staticmethod - def _resolve_input(value: str | Path | None, kind: str = "file") -> Path | None: - if kind not in {"file", "directory", "any", "optional", "none"}: - raise ExternalToolError(f"Unsupported input path kind: {kind}") - if kind == "none": - return None - if value is None or not str(value).strip(): - if kind not in {"none", "optional"}: - raise ExternalToolError("This command requires an input path") - return None - path = Path(value).expanduser().resolve() - if kind == "file" and not path.is_file(): - raise ExternalToolError(f"Input file was not found: {path}") - if kind == "directory" and not path.is_dir(): - raise ExternalToolError(f"Input folder was not found: {path}") - if kind in {"any", "optional"} and not path.exists(): - raise ExternalToolError(f"Input path was not found: {path}") - return path - - @staticmethod - def _resolve_output(value: str | Path | None, kind: str = "file") -> Path | None: - if kind not in {"file", "directory", "optional", "none"}: - raise ExternalToolError(f"Unsupported output path kind: {kind}") - if kind == "none": - return None - if value is None or not str(value).strip(): - if kind not in {"none", "optional"}: - raise ExternalToolError("This command requires an output path") - return None - path = Path(value).expanduser().resolve() - if kind == "directory" and not path.is_dir(): - raise ExternalToolError(f"Output folder was not found: {path}") - if kind != "directory" and not path.parent.is_dir(): - raise ExternalToolError(f"Output folder was not found: {path.parent}") - return path +from unityscraper.domains.tools.models import ToolLaunch, ToolResult +from unityscraper.domains.tools.runner import ( + ExternalToolError, + ExternalToolRunner, + format_command, + split_arguments, +) + +__all__ = [ + "ExternalToolError", + "ExternalToolRunner", + "ToolLaunch", + "ToolResult", + "format_command", + "split_arguments", +] diff --git a/external_tools_gui.py b/external_tools_gui.py index 05c4caa..095a86a 100644 --- a/external_tools_gui.py +++ b/external_tools_gui.py @@ -12,15 +12,17 @@ from typing import Any, Callable from app_paths import resource_path -from external_tools import ( +from unityscraper.domains.tools import ( ExternalToolError, ExternalToolRunner, + ToolCatalog, + ToolDefinition, ToolLaunch, + ToolOperation, ToolResult, format_command, split_arguments, ) -from tool_catalog import ToolCatalog, ToolDefinition, ToolOperation from ui_theme import PALETTE diff --git a/library_service.py b/library_service.py index 0a178c3..5836d2a 100644 --- a/library_service.py +++ b/library_service.py @@ -1,329 +1,8 @@ -""" -Read-only library queries and archive health checks for the desktop interface. - -This module intentionally uses the existing UnityScraper SQLite schema. It does -not perform network requests and can safely be used while browsing the library. -""" +"""Compatibility wrapper for the package-owned library service.""" from __future__ import annotations -import hashlib -import json -import sqlite3 -from contextlib import closing -from dataclasses import dataclass -from pathlib import Path -from typing import Any, Iterable, Optional - -from app_paths import DATABASE_PATH - - -@dataclass(frozen=True) -class GameSummary: - """Compact game information displayed in the library list.""" - - titleid: str - name: str - publisher: str - last_scraped: str - covers_total: int - covers_downloaded: int - updates_total: int - updates_downloaded: int - updates_failed: int - - -class LibraryService: - """Provide UI-friendly queries over the existing database.""" - - def __init__(self, database_path: Path | str = DATABASE_PATH) -> None: - self.database_path = Path(database_path) - - def _connect(self) -> sqlite3.Connection: - self.database_path.parent.mkdir(parents=True, exist_ok=True) - connection = sqlite3.connect(self.database_path) - connection.row_factory = sqlite3.Row - return connection - - def list_games(self, search: str = "") -> list[GameSummary]: - """Return all games, optionally filtered by title, publisher, or TitleID.""" - if not self.database_path.exists(): - return [] - - query = """ - SELECT - t.titleid, - CASE - WHEN t.name IS NULL OR TRIM(t.name) = '' - OR UPPER(TRIM(t.name)) = UPPER(t.titleid) - OR LOWER(TRIM(t.name)) IN ( - 'unknown', 'unknown game', 'unknown title', - 'n/a', 'none', 'null' - ) - THEN COALESCE(NULLIF(TRIM(xc.name), ''), 'Unknown game') - ELSE t.name - END AS name, - COALESCE(t.publisher, '') AS publisher, - COALESCE(t.last_scraped, '') AS last_scraped, - COUNT(DISTINCT cv.id) AS covers_total, - COUNT(DISTINCT CASE WHEN cv.status = 'downloaded' THEN cv.id END) - AS covers_downloaded, - COUNT(DISTINCT u.id) AS updates_total, - COUNT(DISTINCT CASE WHEN u.status = 'downloaded' THEN u.id END) - AS updates_downloaded, - COUNT(DISTINCT CASE WHEN u.status = 'failed' THEN u.id END) - AS updates_failed - FROM titleids AS t - LEFT JOIN xboxunity_title_catalog AS xc ON xc.titleid = t.titleid - LEFT JOIN covers AS cv ON cv.titleid = t.titleid - LEFT JOIN title_updates AS u ON u.titleid = t.titleid - """ - parameters: list[Any] = [] - - if search.strip(): - query += """ - WHERE LOWER(t.titleid) LIKE ? - OR LOWER(COALESCE(t.name, '')) LIKE ? - OR LOWER(COALESCE(xc.name, '')) LIKE ? - OR LOWER(COALESCE(t.publisher, '')) LIKE ? - """ - value = f"%{search.strip().lower()}%" - parameters.extend([value, value, value, value]) - - query += """ - GROUP BY t.titleid, t.name, t.publisher, t.last_scraped, xc.name - ORDER BY name COLLATE NOCASE, t.titleid - """ - - with closing(self._connect()) as connection: - rows = connection.execute(query, parameters).fetchall() - - return [ - GameSummary( - titleid=row["titleid"], - name=row["name"], - publisher=row["publisher"], - last_scraped=row["last_scraped"], - covers_total=row["covers_total"], - covers_downloaded=row["covers_downloaded"], - updates_total=row["updates_total"], - updates_downloaded=row["updates_downloaded"], - updates_failed=row["updates_failed"], - ) - for row in rows - ] - - def get_game_details(self, titleid: str) -> dict[str, Any]: - """Return one title and all of its cover/update records.""" - if not self.database_path.exists(): - return {} - - with closing(self._connect()) as connection: - title = connection.execute( - """ - SELECT t.*, xc.name AS catalog_name - FROM titleids AS t - LEFT JOIN xboxunity_title_catalog AS xc ON xc.titleid = t.titleid - WHERE t.titleid = ? - """, - (titleid,), - ).fetchone() - - if title is None: - return {} - - covers = connection.execute( - """ - SELECT * - FROM covers - WHERE titleid = ? - ORDER BY - CASE status - WHEN 'downloaded' THEN 0 - WHEN 'pending' THEN 1 - ELSE 2 - END, - download_date DESC - """, - (titleid,), - ).fetchall() - - updates = connection.execute( - """ - SELECT * - FROM title_updates - WHERE titleid = ? - ORDER BY media_id, CAST(version AS INTEGER) DESC, version DESC - """, - (titleid,), - ).fetchall() - - title_record = dict(title) - current_name = title_record.get("name") - unknown_names = { - "", - "unknown", - "unknown game", - "unknown title", - "n/a", - "none", - "null", - } - if ( - current_name is None - or str(current_name).strip().casefold() in unknown_names - or str(current_name).strip().upper() == titleid.upper() - ): - title_record["name"] = title_record.get("catalog_name") - title_record.pop("catalog_name", None) - - return { - "title": title_record, - "covers": [dict(row) for row in covers], - "updates": [dict(row) for row in updates], - } - - def get_dashboard_counts(self) -> dict[str, int]: - """Return summary counts used by the dashboard header.""" - if not self.database_path.exists(): - return { - "games": 0, - "updates_available": 0, - "updates_downloaded": 0, - "failed": 0, - "covers_downloaded": 0, - } - - sql = { - "games": "SELECT COUNT(*) FROM titleids", - "updates_available": "SELECT COUNT(*) FROM title_updates", - "updates_downloaded": ( - "SELECT COUNT(*) FROM title_updates WHERE status = 'downloaded'" - ), - "failed": """ - SELECT - (SELECT COUNT(*) FROM title_updates WHERE status = 'failed') + - (SELECT COUNT(*) FROM covers WHERE status = 'failed') - """, - "covers_downloaded": ( - "SELECT COUNT(*) FROM covers WHERE status = 'downloaded'" - ), - } - - with closing(self._connect()) as connection: - return { - name: int(connection.execute(statement).fetchone()[0] or 0) - for name, statement in sql.items() - } - - def find_database_duplicates(self) -> list[dict[str, Any]]: - """ - Find duplicate logical records. - - Title updates are grouped by TitleID, MediaID, and version. Covers are - grouped by TitleID and URL because the existing schema does not enforce - a unique cover constraint. - """ - if not self.database_path.exists(): - return [] - - with closing(self._connect()) as connection: - update_rows = connection.execute( - """ - SELECT - 'update' AS item_type, - titleid, - COALESCE(media_id, '') AS identity_a, - COALESCE(version, '') AS identity_b, - COUNT(*) AS duplicate_count - FROM title_updates - GROUP BY titleid, media_id, version - HAVING COUNT(*) > 1 - """ - ).fetchall() - - cover_rows = connection.execute( - """ - SELECT - 'cover' AS item_type, - titleid, - COALESCE(cover_url, '') AS identity_a, - '' AS identity_b, - COUNT(*) AS duplicate_count - FROM covers - GROUP BY titleid, cover_url - HAVING COUNT(*) > 1 - """ - ).fetchall() - - return [dict(row) for row in (*update_rows, *cover_rows)] - - def scan_archive_health(self) -> dict[str, Any]: - """Check downloaded database records for missing, empty, and duplicate files.""" - report: dict[str, Any] = { - "checked": 0, - "healthy": [], - "missing": [], - "empty": [], - "duplicate_files": [], - "database_duplicates": self.find_database_duplicates(), - } - - if not self.database_path.exists(): - return report - - with closing(self._connect()) as connection: - rows = connection.execute( - """ - SELECT 'cover' AS item_type, id, titleid, file_path, file_size - FROM covers - WHERE status = 'downloaded' - UNION ALL - SELECT 'update' AS item_type, id, titleid, file_path, file_size - FROM title_updates - WHERE status = 'downloaded' - """ - ).fetchall() - - hashes: dict[str, list[dict[str, Any]]] = {} - - for row in rows: - item = dict(row) - report["checked"] += 1 - raw_path = item.get("file_path") - - if not raw_path: - report["missing"].append({**item, "reason": "No file path stored"}) - continue - - path = Path(raw_path) - if not path.exists(): - report["missing"].append({**item, "reason": "File does not exist"}) - continue - - size = path.stat().st_size - if size == 0: - report["empty"].append({**item, "actual_size": 0}) - continue - - digest = self._sha256(path) - hashes.setdefault(digest, []).append( - {**item, "actual_size": size, "sha256": digest} - ) - report["healthy"].append({**item, "actual_size": size, "sha256": digest}) - - report["duplicate_files"] = [ - {"sha256": digest, "items": items} - for digest, items in hashes.items() - if len(items) > 1 - ] - return report +from unityscraper.domains.library.models import GameSummary +from unityscraper.domains.library.service import LibraryService - @staticmethod - def _sha256(path: Path) -> str: - """Calculate SHA-256 without loading the complete file into memory.""" - hasher = hashlib.sha256() - with path.open("rb") as stream: - for chunk in iter(lambda: stream.read(1024 * 1024), b""): - hasher.update(chunk) - return hasher.hexdigest() +__all__ = ["GameSummary", "LibraryService"] diff --git a/main.py b/main.py index ec3c474..c23ebd8 100644 --- a/main.py +++ b/main.py @@ -1141,8 +1141,12 @@ def main(): # Initialize scraper if args.list_tools or args.tool_id: try: - from external_tools import ExternalToolRunner, format_command - from tool_catalog import ToolCatalog, operation_for + from unityscraper.domains.tools import ( + ExternalToolRunner, + ToolCatalog, + format_command, + operation_for, + ) catalog = ToolCatalog(CONFIG_PATH) if args.list_tools: diff --git a/modern_gui.py b/modern_gui.py index ac988da..03438f2 100644 --- a/modern_gui.py +++ b/modern_gui.py @@ -48,7 +48,7 @@ from knowledge_scheduler import KnowledgeScheduler from knowledge_gui import KnowledgePage from i18n import SUPPORTED_LANGUAGES, init_translator, t -from library_service import GameSummary, LibraryService +from unityscraper.domains.library import GameSummary, LibraryService from platform_support import open_path from profile_gui import ProfileSavePage from profile_manager import ProfileSaveManager diff --git a/tests.py b/tests.py index 032adb7..7322aa1 100644 --- a/tests.py +++ b/tests.py @@ -85,7 +85,9 @@ from unityscraper.domains.library.models import GameSummary as ModularGameSummary from unityscraper.domains.library.service import LibraryService as ModularLibraryService from unityscraper.domains.packages.commands import InspectStfsPackage, InventoryStfsFileTable +from unityscraper.domains.tools.catalog import ToolCatalog as ModularToolCatalog from unityscraper.domains.tools.models import ToolDefinition as ModularToolDefinition +from unityscraper.domains.tools.runner import ExternalToolRunner as ModularToolRunner class TestPlatformSupport(unittest.TestCase): @@ -211,7 +213,7 @@ def test_pyinstaller_collects_package_modules(self): class TestModularFoundation(unittest.TestCase): - """Test package-level adapters that support the modular architecture.""" + """Test package ownership and legacy compatibility boundaries.""" def test_domain_service_exports_preserve_existing_implementations(self): self.assertIs(ModularBackupService, BackupService) @@ -219,6 +221,26 @@ def test_domain_service_exports_preserve_existing_implementations(self): self.assertIs(ModularGameSummary, GameSummary) self.assertIs(ModularEntityRecord, EntityRecord) self.assertIs(ModularToolDefinition, ToolDefinition) + self.assertIs(ModularToolCatalog, ToolCatalog) + self.assertIs(ModularToolRunner, ExternalToolRunner) + + def test_migrated_implementations_are_domain_owned(self): + self.assertEqual( + ModularLibraryService.__module__, + "unityscraper.domains.library.service", + ) + self.assertEqual( + ModularGameSummary.__module__, + "unityscraper.domains.library.models", + ) + self.assertEqual( + ModularToolCatalog.__module__, + "unityscraper.domains.tools.catalog", + ) + self.assertEqual( + ModularToolRunner.__module__, + "unityscraper.domains.tools.runner", + ) def test_backup_schema_is_domain_owned_with_legacy_compatibility(self): from backup_service import ensure_backup_schema as LegacyBackupSchema @@ -1000,7 +1022,7 @@ def test_catalog_persists_and_hashes_user_selected_executable(self): hashlib.sha256(b"test executable").hexdigest().upper(), ) - @patch("external_tools.subprocess.Popen") + @patch("unityscraper.domains.tools.runner.subprocess.Popen") def test_detached_launch_uses_argument_vector(self, popen): popen.return_value.pid = 360 runner = ExternalToolRunner() diff --git a/tool_catalog.py b/tool_catalog.py index 4ea2fde..a661523 100644 --- a/tool_catalog.py +++ b/tool_catalog.py @@ -1,311 +1,21 @@ -"""Declarative catalog and conservative discovery for community Xbox tools.""" +"""Compatibility wrapper for the package-owned external tool catalog.""" from __future__ import annotations -import hashlib -import json -import os -import shutil -import sys -from dataclasses import dataclass -from pathlib import Path -from typing import Iterable - -from app_paths import CONFIG_PATH, executable_root, resource_path - - -@dataclass(frozen=True) -class ToolOperation: - """One supported, reviewable action exposed by a tool.""" - - id: str - label: str - arguments: tuple[str, ...] = () - input_kind: str = "none" - output_kind: str = "none" - detached: bool = False - destructive: bool = False - - -@dataclass(frozen=True) -class ToolDefinition: - """Metadata and discovery hints for a community utility.""" - - id: str - name: str - author: str - homepage: str - platforms: tuple[str, ...] - executable_names: tuple[str, ...] - operations: tuple[ToolOperation, ...] - bundled_path: tuple[str, ...] = () - - def supports_current_platform(self) -> bool: - return platform_key() in self.platforms or "all" in self.platforms - - -TOOLS: tuple[ToolDefinition, ...] = ( - ToolDefinition( - "xextool", - "XeXTool", - "xorloser", - "https://github.com/XboxChef/XexToolGUI", - ("windows",), - ("xextool.exe",), - ( - ToolOperation( - "extended-info", "Extended information", ("-l", "{input}"), "file" - ), - ToolOperation("basic-info", "Basic information", ("{input}",), "file"), - ToolOperation( - "custom", "Custom arguments", (), "optional", destructive=True - ), - ), - ("assets", "tools", "xextool", "xextool.exe"), - ), - ToolDefinition( - "extract-xiso", - "extract-xiso", - "XboxDev", - "https://github.com/XboxDev/extract-xiso", - ("windows", "linux", "macos"), - ("extract-xiso.exe", "extract-xiso"), - ( - ToolOperation("list", "List image contents", ("-l", "{input}"), "file"), - ToolOperation( - "extract", - "Extract image", - ("-x", "{input}", "-d", "{output}"), - "file", - "directory", - ), - ToolOperation( - "create", - "Create image", - ("-c", "{input}", "{output}"), - "directory", - "file", - ), - ToolOperation( - "rewrite", - "Rewrite image", - ("-r", "{input}"), - "file", - destructive=True, - ), - ), - ), - ToolDefinition( - "xenia", - "Xenia", - "Xenia Project", - "https://github.com/xenia-project/xenia", - ("windows",), - ("xenia.exe",), - ( - ToolOperation( - "launch-game", "Launch game", ("{input}",), "any", detached=True - ), - ToolOperation("open", "Open emulator", detached=True), - ), - ), - ToolDefinition( - "xenia-canary", - "Xenia Canary", - "Xenia Canary Project", - "https://github.com/xenia-canary/xenia-canary", - ("windows",), - ("xenia_canary.exe", "xenia-canary.exe"), - ( - ToolOperation( - "launch-game", "Launch game", ("{input}",), "any", detached=True - ), - ToolOperation("open", "Open emulator", detached=True), - ), - ), - ToolDefinition( - "velocity", - "Velocity", - "Velocity contributors", - "https://github.com/Gualdimar/Velocity", - ("windows",), - ("Velocity.exe",), - (ToolOperation("open", "Open Velocity", detached=True),), - ), - ToolDefinition( - "iso2god", - "Iso2God", - "Iso2God contributors", - "https://github.com/r4dius/Iso2God", - ("windows",), - ("Iso2God.exe",), - (ToolOperation("open", "Open Iso2God", detached=True),), - ), - ToolDefinition( - "god2iso", - "God2ISO", - "Community utility", - "", - ("windows",), - ("God2Iso.exe", "God2ISO.exe"), - (ToolOperation("open", "Open God2ISO", detached=True),), - ), - ToolDefinition( - "xbox-image-browser", - "Xbox Image Browser", - "Community utility", - "", - ("windows",), - ("Xbox Image Browser.exe", "XboxImageBrowser.exe"), - (ToolOperation("open", "Open Xbox Image Browser", detached=True),), - ), - ToolDefinition( - "le-fluffie", - "Le Fluffie", - "Dalavin (DJ SkunkieButt)", - "", - ("windows",), - ("Le Fluffie.exe", "LeFluffie.exe"), - (ToolOperation("open", "Open Le Fluffie", detached=True),), - ), - ToolDefinition( - "custom", - "Custom CLI tool", - "User supplied", - "", - ("all",), - (), - ( - ToolOperation( - "custom", - "Custom arguments", - (), - "optional", - "optional", - destructive=True, - ), - ), - ), +from unityscraper.domains.tools.catalog import ( + TOOLS, + ToolCatalog, + ToolDefinition, + ToolOperation, + operation_for, + platform_key, ) - -def platform_key() -> str: - if os.name == "nt": - return "windows" - if sys.platform == "darwin": - return "macos" - return "linux" - - -class ToolCatalog: - """Resolve built-in definitions and user-approved executable paths.""" - - def __init__(self, config_path: Path | str = CONFIG_PATH) -> None: - self.config_path = Path(config_path) - self._definitions = {tool.id: tool for tool in TOOLS} - - def definitions(self, *, supported_only: bool = False) -> tuple[ToolDefinition, ...]: - values = TOOLS - if supported_only: - values = tuple(tool for tool in values if tool.supports_current_platform()) - return values - - def get(self, tool_id: str) -> ToolDefinition: - try: - return self._definitions[tool_id] - except KeyError as exc: - raise KeyError(f"Unknown external tool: {tool_id}") from exc - - def configured_path(self, tool_id: str) -> Path | None: - config = self._read_config() - tools = config.get("external_tools", {}) - if not isinstance(tools, dict): - return None - paths = tools.get("paths", {}) - value = paths.get(tool_id, "") if isinstance(paths, dict) else "" - if not value and tool_id == "xextool": - value = tools.get("xextool_path", "") - if not value and tool_id == "custom": - value = tools.get("custom_tool_path", "") - path = Path(str(value)).expanduser() - return path.resolve() if path.is_file() else None - - def discover(self, tool_id: str) -> Path | None: - tool = self.get(tool_id) - if not tool.supports_current_platform(): - return None - configured = self.configured_path(tool_id) - if configured: - return configured - if tool.bundled_path: - bundled = resource_path(*tool.bundled_path) - if bundled.is_file(): - return bundled.resolve() - for name in tool.executable_names: - located = shutil.which(name) - if located and Path(located).is_file(): - return Path(located).resolve() - for candidate in self._conventional_candidates(tool): - if candidate.is_file(): - return candidate.resolve() - return None - - def save_path(self, tool_id: str, path: Path | str) -> Path: - executable = Path(path).expanduser().resolve() - if not executable.is_file(): - raise FileNotFoundError(executable) - config = self._read_config() - tools = config.setdefault("external_tools", {}) - if not isinstance(tools, dict): - tools = {} - config["external_tools"] = tools - paths = tools.setdefault("paths", {}) - if not isinstance(paths, dict): - paths = {} - tools["paths"] = paths - paths[tool_id] = str(executable) - self.config_path.parent.mkdir(parents=True, exist_ok=True) - self.config_path.write_text(json.dumps(config, indent=2), encoding="utf-8") - return executable - - def checksum(self, path: Path | str) -> str: - digest = hashlib.sha256() - with Path(path).open("rb") as handle: - for chunk in iter(lambda: handle.read(1024 * 1024), b""): - digest.update(chunk) - return digest.hexdigest().upper() - - def _read_config(self) -> dict[str, object]: - if not self.config_path.exists(): - return {} - try: - value = json.loads(self.config_path.read_text(encoding="utf-8")) - return value if isinstance(value, dict) else {} - except (OSError, ValueError): - return {} - - @staticmethod - def _conventional_candidates(tool: ToolDefinition) -> Iterable[Path]: - root = executable_root() - home = Path.home() - locations = [ - root, - root / "tools" / tool.id, - home / "Documents" / tool.name, - home / "Downloads" / tool.name, - ] - if os.name == "nt": - for variable in ("ProgramFiles", "ProgramFiles(x86)"): - value = os.environ.get(variable) - if value: - locations.append(Path(value) / tool.name) - for location in locations: - for name in tool.executable_names: - yield location / name - - -def operation_for(tool: ToolDefinition, operation_id: str) -> ToolOperation: - try: - return next(item for item in tool.operations if item.id == operation_id) - except StopIteration as exc: - raise KeyError(f"{tool.name} does not support operation {operation_id}") from exc +__all__ = [ + "TOOLS", + "ToolCatalog", + "ToolDefinition", + "ToolOperation", + "operation_for", + "platform_key", +] diff --git a/unityscraper/domains/library/models.py b/unityscraper/domains/library/models.py index bffc112..fd14a2f 100644 --- a/unityscraper/domains/library/models.py +++ b/unityscraper/domains/library/models.py @@ -2,8 +2,24 @@ from __future__ import annotations -from library_service import GameSummary +from dataclasses import dataclass + from title_catalog import CatalogSyncResult, TitleSuggestion -__all__ = ["CatalogSyncResult", "GameSummary", "TitleSuggestion"] +@dataclass(frozen=True) +class GameSummary: + """Compact game information displayed in the library list.""" + + titleid: str + name: str + publisher: str + last_scraped: str + covers_total: int + covers_downloaded: int + updates_total: int + updates_downloaded: int + updates_failed: int + + +__all__ = ["CatalogSyncResult", "GameSummary", "TitleSuggestion"] diff --git a/unityscraper/domains/library/service.py b/unityscraper/domains/library/service.py index 0b37edf..21c2743 100644 --- a/unityscraper/domains/library/service.py +++ b/unityscraper/domains/library/service.py @@ -1,16 +1,318 @@ -"""Package-facing library service exports.""" +""" +Read-only library queries and archive health checks for the desktop interface. + +This module intentionally uses the existing UnityScraper SQLite schema. It does +not perform network requests and can safely be used while browsing the library. +""" from __future__ import annotations -from library_service import LibraryService +import hashlib +import json +import sqlite3 +from contextlib import closing +from pathlib import Path +from typing import Any, Iterable, Optional + +from unityscraper.core.paths import DATABASE_PATH + +from .models import GameSummary + + +class LibraryService: + """Provide UI-friendly queries over the existing database.""" + + def __init__(self, database_path: Path | str = DATABASE_PATH) -> None: + self.database_path = Path(database_path) + + def _connect(self) -> sqlite3.Connection: + self.database_path.parent.mkdir(parents=True, exist_ok=True) + connection = sqlite3.connect(self.database_path) + connection.row_factory = sqlite3.Row + return connection + + def list_games(self, search: str = "") -> list[GameSummary]: + """Return all games, optionally filtered by title, publisher, or TitleID.""" + if not self.database_path.exists(): + return [] + + query = """ + SELECT + t.titleid, + CASE + WHEN t.name IS NULL OR TRIM(t.name) = '' + OR UPPER(TRIM(t.name)) = UPPER(t.titleid) + OR LOWER(TRIM(t.name)) IN ( + 'unknown', 'unknown game', 'unknown title', + 'n/a', 'none', 'null' + ) + THEN COALESCE(NULLIF(TRIM(xc.name), ''), 'Unknown game') + ELSE t.name + END AS name, + COALESCE(t.publisher, '') AS publisher, + COALESCE(t.last_scraped, '') AS last_scraped, + COUNT(DISTINCT cv.id) AS covers_total, + COUNT(DISTINCT CASE WHEN cv.status = 'downloaded' THEN cv.id END) + AS covers_downloaded, + COUNT(DISTINCT u.id) AS updates_total, + COUNT(DISTINCT CASE WHEN u.status = 'downloaded' THEN u.id END) + AS updates_downloaded, + COUNT(DISTINCT CASE WHEN u.status = 'failed' THEN u.id END) + AS updates_failed + FROM titleids AS t + LEFT JOIN xboxunity_title_catalog AS xc ON xc.titleid = t.titleid + LEFT JOIN covers AS cv ON cv.titleid = t.titleid + LEFT JOIN title_updates AS u ON u.titleid = t.titleid + """ + parameters: list[Any] = [] + + if search.strip(): + query += """ + WHERE LOWER(t.titleid) LIKE ? + OR LOWER(COALESCE(t.name, '')) LIKE ? + OR LOWER(COALESCE(xc.name, '')) LIKE ? + OR LOWER(COALESCE(t.publisher, '')) LIKE ? + """ + value = f"%{search.strip().lower()}%" + parameters.extend([value, value, value, value]) + + query += """ + GROUP BY t.titleid, t.name, t.publisher, t.last_scraped, xc.name + ORDER BY name COLLATE NOCASE, t.titleid + """ + + with closing(self._connect()) as connection: + rows = connection.execute(query, parameters).fetchall() + + return [ + GameSummary( + titleid=row["titleid"], + name=row["name"], + publisher=row["publisher"], + last_scraped=row["last_scraped"], + covers_total=row["covers_total"], + covers_downloaded=row["covers_downloaded"], + updates_total=row["updates_total"], + updates_downloaded=row["updates_downloaded"], + updates_failed=row["updates_failed"], + ) + for row in rows + ] + + def get_game_details(self, titleid: str) -> dict[str, Any]: + """Return one title and all of its cover/update records.""" + if not self.database_path.exists(): + return {} + + with closing(self._connect()) as connection: + title = connection.execute( + """ + SELECT t.*, xc.name AS catalog_name + FROM titleids AS t + LEFT JOIN xboxunity_title_catalog AS xc ON xc.titleid = t.titleid + WHERE t.titleid = ? + """, + (titleid,), + ).fetchone() + + if title is None: + return {} + + covers = connection.execute( + """ + SELECT * + FROM covers + WHERE titleid = ? + ORDER BY + CASE status + WHEN 'downloaded' THEN 0 + WHEN 'pending' THEN 1 + ELSE 2 + END, + download_date DESC + """, + (titleid,), + ).fetchall() + + updates = connection.execute( + """ + SELECT * + FROM title_updates + WHERE titleid = ? + ORDER BY media_id, CAST(version AS INTEGER) DESC, version DESC + """, + (titleid,), + ).fetchall() + + title_record = dict(title) + current_name = title_record.get("name") + unknown_names = { + "", + "unknown", + "unknown game", + "unknown title", + "n/a", + "none", + "null", + } + if ( + current_name is None + or str(current_name).strip().casefold() in unknown_names + or str(current_name).strip().upper() == titleid.upper() + ): + title_record["name"] = title_record.get("catalog_name") + title_record.pop("catalog_name", None) + + return { + "title": title_record, + "covers": [dict(row) for row in covers], + "updates": [dict(row) for row in updates], + } + + def get_dashboard_counts(self) -> dict[str, int]: + """Return summary counts used by the dashboard header.""" + if not self.database_path.exists(): + return { + "games": 0, + "updates_available": 0, + "updates_downloaded": 0, + "failed": 0, + "covers_downloaded": 0, + } + + sql = { + "games": "SELECT COUNT(*) FROM titleids", + "updates_available": "SELECT COUNT(*) FROM title_updates", + "updates_downloaded": ( + "SELECT COUNT(*) FROM title_updates WHERE status = 'downloaded'" + ), + "failed": """ + SELECT + (SELECT COUNT(*) FROM title_updates WHERE status = 'failed') + + (SELECT COUNT(*) FROM covers WHERE status = 'failed') + """, + "covers_downloaded": ( + "SELECT COUNT(*) FROM covers WHERE status = 'downloaded'" + ), + } + + with closing(self._connect()) as connection: + return { + name: int(connection.execute(statement).fetchone()[0] or 0) + for name, statement in sql.items() + } + + def find_database_duplicates(self) -> list[dict[str, Any]]: + """ + Find duplicate logical records. + + Title updates are grouped by TitleID, MediaID, and version. Covers are + grouped by TitleID and URL because the existing schema does not enforce + a unique cover constraint. + """ + if not self.database_path.exists(): + return [] + + with closing(self._connect()) as connection: + update_rows = connection.execute( + """ + SELECT + 'update' AS item_type, + titleid, + COALESCE(media_id, '') AS identity_a, + COALESCE(version, '') AS identity_b, + COUNT(*) AS duplicate_count + FROM title_updates + GROUP BY titleid, media_id, version + HAVING COUNT(*) > 1 + """ + ).fetchall() + + cover_rows = connection.execute( + """ + SELECT + 'cover' AS item_type, + titleid, + COALESCE(cover_url, '') AS identity_a, + '' AS identity_b, + COUNT(*) AS duplicate_count + FROM covers + GROUP BY titleid, cover_url + HAVING COUNT(*) > 1 + """ + ).fetchall() + + return [dict(row) for row in (*update_rows, *cover_rows)] + + def scan_archive_health(self) -> dict[str, Any]: + """Check downloaded database records for missing, empty, and duplicate files.""" + report: dict[str, Any] = { + "checked": 0, + "healthy": [], + "missing": [], + "empty": [], + "duplicate_files": [], + "database_duplicates": self.find_database_duplicates(), + } + + if not self.database_path.exists(): + return report + + with closing(self._connect()) as connection: + rows = connection.execute( + """ + SELECT 'cover' AS item_type, id, titleid, file_path, file_size + FROM covers + WHERE status = 'downloaded' + UNION ALL + SELECT 'update' AS item_type, id, titleid, file_path, file_size + FROM title_updates + WHERE status = 'downloaded' + """ + ).fetchall() + + hashes: dict[str, list[dict[str, Any]]] = {} + + for row in rows: + item = dict(row) + report["checked"] += 1 + raw_path = item.get("file_path") + + if not raw_path: + report["missing"].append({**item, "reason": "No file path stored"}) + continue + + path = Path(raw_path) + if not path.exists(): + report["missing"].append({**item, "reason": "File does not exist"}) + continue + + size = path.stat().st_size + if size == 0: + report["empty"].append({**item, "actual_size": 0}) + continue + + digest = self._sha256(path) + hashes.setdefault(digest, []).append( + {**item, "actual_size": size, "sha256": digest} + ) + report["healthy"].append({**item, "actual_size": size, "sha256": digest}) + + report["duplicate_files"] = [ + {"sha256": digest, "items": items} + for digest, items in hashes.items() + if len(items) > 1 + ] + return report + + @staticmethod + def _sha256(path: Path) -> str: + """Calculate SHA-256 without loading the complete file into memory.""" + hasher = hashlib.sha256() + with path.open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + hasher.update(chunk) + return hasher.hexdigest() -from .catalog import XboxUnityTitleCatalog -from .models import CatalogSyncResult, GameSummary, TitleSuggestion -__all__ = [ - "CatalogSyncResult", - "GameSummary", - "LibraryService", - "TitleSuggestion", - "XboxUnityTitleCatalog", -] +__all__ = ["GameSummary", "LibraryService"] diff --git a/unityscraper/domains/tools/catalog.py b/unityscraper/domains/tools/catalog.py index e405a30..20abb56 100644 --- a/unityscraper/domains/tools/catalog.py +++ b/unityscraper/domains/tools/catalog.py @@ -1,8 +1,292 @@ -"""External tool catalog exports.""" +"""Declarative catalog and conservative discovery for community Xbox tools.""" from __future__ import annotations -from tool_catalog import ToolCatalog, operation_for, platform_key +import hashlib +import json +import os +import shutil +import sys +from pathlib import Path +from typing import Iterable -__all__ = ["ToolCatalog", "operation_for", "platform_key"] +from unityscraper.core.paths import CONFIG_PATH, executable_root, resource_path +from .models import ToolDefinition, ToolOperation + + +TOOLS: tuple[ToolDefinition, ...] = ( + ToolDefinition( + "xextool", + "XeXTool", + "xorloser", + "https://github.com/XboxChef/XexToolGUI", + ("windows",), + ("xextool.exe",), + ( + ToolOperation( + "extended-info", "Extended information", ("-l", "{input}"), "file" + ), + ToolOperation("basic-info", "Basic information", ("{input}",), "file"), + ToolOperation( + "custom", "Custom arguments", (), "optional", destructive=True + ), + ), + ("assets", "tools", "xextool", "xextool.exe"), + ), + ToolDefinition( + "extract-xiso", + "extract-xiso", + "XboxDev", + "https://github.com/XboxDev/extract-xiso", + ("windows", "linux", "macos"), + ("extract-xiso.exe", "extract-xiso"), + ( + ToolOperation("list", "List image contents", ("-l", "{input}"), "file"), + ToolOperation( + "extract", + "Extract image", + ("-x", "{input}", "-d", "{output}"), + "file", + "directory", + ), + ToolOperation( + "create", + "Create image", + ("-c", "{input}", "{output}"), + "directory", + "file", + ), + ToolOperation( + "rewrite", + "Rewrite image", + ("-r", "{input}"), + "file", + destructive=True, + ), + ), + ), + ToolDefinition( + "xenia", + "Xenia", + "Xenia Project", + "https://github.com/xenia-project/xenia", + ("windows",), + ("xenia.exe",), + ( + ToolOperation( + "launch-game", "Launch game", ("{input}",), "any", detached=True + ), + ToolOperation("open", "Open emulator", detached=True), + ), + ), + ToolDefinition( + "xenia-canary", + "Xenia Canary", + "Xenia Canary Project", + "https://github.com/xenia-canary/xenia-canary", + ("windows",), + ("xenia_canary.exe", "xenia-canary.exe"), + ( + ToolOperation( + "launch-game", "Launch game", ("{input}",), "any", detached=True + ), + ToolOperation("open", "Open emulator", detached=True), + ), + ), + ToolDefinition( + "velocity", + "Velocity", + "Velocity contributors", + "https://github.com/Gualdimar/Velocity", + ("windows",), + ("Velocity.exe",), + (ToolOperation("open", "Open Velocity", detached=True),), + ), + ToolDefinition( + "iso2god", + "Iso2God", + "Iso2God contributors", + "https://github.com/r4dius/Iso2God", + ("windows",), + ("Iso2God.exe",), + (ToolOperation("open", "Open Iso2God", detached=True),), + ), + ToolDefinition( + "god2iso", + "God2ISO", + "Community utility", + "", + ("windows",), + ("God2Iso.exe", "God2ISO.exe"), + (ToolOperation("open", "Open God2ISO", detached=True),), + ), + ToolDefinition( + "xbox-image-browser", + "Xbox Image Browser", + "Community utility", + "", + ("windows",), + ("Xbox Image Browser.exe", "XboxImageBrowser.exe"), + (ToolOperation("open", "Open Xbox Image Browser", detached=True),), + ), + ToolDefinition( + "le-fluffie", + "Le Fluffie", + "Dalavin (DJ SkunkieButt)", + "", + ("windows",), + ("Le Fluffie.exe", "LeFluffie.exe"), + (ToolOperation("open", "Open Le Fluffie", detached=True),), + ), + ToolDefinition( + "custom", + "Custom CLI tool", + "User supplied", + "", + ("all",), + (), + ( + ToolOperation( + "custom", + "Custom arguments", + (), + "optional", + "optional", + destructive=True, + ), + ), + ), +) + + +def platform_key() -> str: + if os.name == "nt": + return "windows" + if sys.platform == "darwin": + return "macos" + return "linux" + + +class ToolCatalog: + """Resolve built-in definitions and user-approved executable paths.""" + + def __init__(self, config_path: Path | str = CONFIG_PATH) -> None: + self.config_path = Path(config_path) + self._definitions = {tool.id: tool for tool in TOOLS} + + def definitions(self, *, supported_only: bool = False) -> tuple[ToolDefinition, ...]: + values = TOOLS + if supported_only: + values = tuple(tool for tool in values if tool.supports_current_platform()) + return values + + def get(self, tool_id: str) -> ToolDefinition: + try: + return self._definitions[tool_id] + except KeyError as exc: + raise KeyError(f"Unknown external tool: {tool_id}") from exc + + def configured_path(self, tool_id: str) -> Path | None: + config = self._read_config() + tools = config.get("external_tools", {}) + if not isinstance(tools, dict): + return None + paths = tools.get("paths", {}) + value = paths.get(tool_id, "") if isinstance(paths, dict) else "" + if not value and tool_id == "xextool": + value = tools.get("xextool_path", "") + if not value and tool_id == "custom": + value = tools.get("custom_tool_path", "") + path = Path(str(value)).expanduser() + return path.resolve() if path.is_file() else None + + def discover(self, tool_id: str) -> Path | None: + tool = self.get(tool_id) + if not tool.supports_current_platform(): + return None + configured = self.configured_path(tool_id) + if configured: + return configured + if tool.bundled_path: + bundled = resource_path(*tool.bundled_path) + if bundled.is_file(): + return bundled.resolve() + for name in tool.executable_names: + located = shutil.which(name) + if located and Path(located).is_file(): + return Path(located).resolve() + for candidate in self._conventional_candidates(tool): + if candidate.is_file(): + return candidate.resolve() + return None + + def save_path(self, tool_id: str, path: Path | str) -> Path: + executable = Path(path).expanduser().resolve() + if not executable.is_file(): + raise FileNotFoundError(executable) + config = self._read_config() + tools = config.setdefault("external_tools", {}) + if not isinstance(tools, dict): + tools = {} + config["external_tools"] = tools + paths = tools.setdefault("paths", {}) + if not isinstance(paths, dict): + paths = {} + tools["paths"] = paths + paths[tool_id] = str(executable) + self.config_path.parent.mkdir(parents=True, exist_ok=True) + self.config_path.write_text(json.dumps(config, indent=2), encoding="utf-8") + return executable + + def checksum(self, path: Path | str) -> str: + digest = hashlib.sha256() + with Path(path).open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest().upper() + + def _read_config(self) -> dict[str, object]: + if not self.config_path.exists(): + return {} + try: + value = json.loads(self.config_path.read_text(encoding="utf-8")) + return value if isinstance(value, dict) else {} + except (OSError, ValueError): + return {} + + @staticmethod + def _conventional_candidates(tool: ToolDefinition) -> Iterable[Path]: + root = executable_root() + home = Path.home() + locations = [ + root, + root / "tools" / tool.id, + home / "Documents" / tool.name, + home / "Downloads" / tool.name, + ] + if os.name == "nt": + for variable in ("ProgramFiles", "ProgramFiles(x86)"): + value = os.environ.get(variable) + if value: + locations.append(Path(value) / tool.name) + for location in locations: + for name in tool.executable_names: + yield location / name + + +def operation_for(tool: ToolDefinition, operation_id: str) -> ToolOperation: + try: + return next(item for item in tool.operations if item.id == operation_id) + except StopIteration as exc: + raise KeyError(f"{tool.name} does not support operation {operation_id}") from exc + + +__all__ = [ + "TOOLS", + "ToolCatalog", + "ToolDefinition", + "ToolOperation", + "operation_for", + "platform_key", +] diff --git a/unityscraper/domains/tools/models.py b/unityscraper/domains/tools/models.py index 4bee02f..932693e 100644 --- a/unityscraper/domains/tools/models.py +++ b/unityscraper/domains/tools/models.py @@ -1,9 +1,59 @@ -"""External tool domain data models.""" +"""Data contracts for external Xbox utility discovery and execution.""" from __future__ import annotations -from external_tools import ToolLaunch, ToolResult -from tool_catalog import ToolDefinition, ToolOperation +from dataclasses import dataclass -__all__ = ["ToolDefinition", "ToolLaunch", "ToolOperation", "ToolResult"] +@dataclass(frozen=True) +class ToolOperation: + """One supported, reviewable action exposed by a tool.""" + + id: str + label: str + arguments: tuple[str, ...] = () + input_kind: str = "none" + output_kind: str = "none" + detached: bool = False + destructive: bool = False + + +@dataclass(frozen=True) +class ToolDefinition: + """Metadata and discovery hints for a community utility.""" + + id: str + name: str + author: str + homepage: str + platforms: tuple[str, ...] + executable_names: tuple[str, ...] + operations: tuple[ToolOperation, ...] + bundled_path: tuple[str, ...] = () + + def supports_current_platform(self) -> bool: + from .catalog import platform_key + + return platform_key() in self.platforms or "all" in self.platforms + + +@dataclass(frozen=True) +class ToolResult: + """Captured result from one external tool invocation.""" + + command: tuple[str, ...] + returncode: int + stdout: str + stderr: str + duration_seconds: float + cancelled: bool + + +@dataclass(frozen=True) +class ToolLaunch: + """Details for a detached graphical tool launch.""" + + command: tuple[str, ...] + pid: int + +__all__ = ["ToolDefinition", "ToolLaunch", "ToolOperation", "ToolResult"] diff --git a/unityscraper/domains/tools/runner.py b/unityscraper/domains/tools/runner.py index 8c0af0c..6311da7 100644 --- a/unityscraper/domains/tools/runner.py +++ b/unityscraper/domains/tools/runner.py @@ -1,18 +1,228 @@ -"""External tool execution exports.""" +"""Safe process runner for user-supplied Xbox command-line tools.""" from __future__ import annotations -from external_tools import ( - ExternalToolError, - ExternalToolRunner, - format_command, - split_arguments, -) +import os +import shlex +import subprocess +import threading +import time +from pathlib import Path +from typing import Iterable + +from .models import ToolLaunch, ToolResult + + +class ExternalToolError(RuntimeError): + """Raised when an external tool cannot be configured or launched.""" + + +def split_arguments(value: str, *, windows: bool | None = None) -> list[str]: + """Split an editable argument template without passing it through a shell.""" + use_windows_rules = os.name == "nt" if windows is None else windows + arguments = shlex.split(value, posix=not use_windows_rules) + if use_windows_rules: + return [ + argument[1:-1] + if len(argument) >= 2 and argument[0] == argument[-1] == '"' + else argument + for argument in arguments + ] + return arguments + + +def format_command(command: Iterable[str], *, windows: bool | None = None) -> str: + """Format an argument vector for display only.""" + values = list(command) + use_windows_rules = os.name == "nt" if windows is None else windows + return subprocess.list2cmdline(values) if use_windows_rules else shlex.join(values) + + +class ExternalToolRunner: + """Run one selected executable at a time without shell interpretation.""" + + def __init__(self) -> None: + self._lock = threading.Lock() + self._process: subprocess.Popen[str] | None = None + self._cancel_requested = False + + def build_command( + self, + executable: str | Path, + argument_template: Iterable[str], + *, + input_path: str | Path | None = None, + output_path: str | Path | None = None, + input_kind: str = "file", + output_kind: str = "optional", + ) -> tuple[str, ...]: + tool = Path(executable).expanduser().resolve() + if not tool.is_file(): + raise ExternalToolError(f"Tool executable was not found: {tool}") + + source = self._resolve_input(input_path, input_kind) + output = self._resolve_output(output_path, output_kind) + arguments: list[str] = [] + for value in argument_template: + if "{input}" in value and source is None: + raise ExternalToolError("This command requires an input file") + if "{output}" in value and output is None: + raise ExternalToolError("This command requires an output path") + arguments.append( + value.replace("{input}", str(source) if source else "") + .replace("{output}", str(output) if output else "") + ) + return (str(tool), *arguments) + + def launch_detached( + self, + executable: str | Path, + argument_template: Iterable[str] = (), + *, + input_path: str | Path | None = None, + output_path: str | Path | None = None, + input_kind: str = "none", + output_kind: str = "none", + ) -> ToolLaunch: + """Launch a GUI utility without waiting for it to exit.""" + command = self.build_command( + executable, + argument_template, + input_path=input_path, + output_path=output_path, + input_kind=input_kind, + output_kind=output_kind, + ) + creation_flags = 0 + if os.name == "nt": + creation_flags = subprocess.CREATE_NEW_PROCESS_GROUP + try: + process = subprocess.Popen( + command, + cwd=Path(command[0]).parent, + shell=False, + creationflags=creation_flags, + close_fds=os.name != "nt", + ) + except OSError as exc: + raise ExternalToolError(f"Could not start external tool: {exc}") from exc + return ToolLaunch(command, process.pid) + + def run( + self, + executable: str | Path, + argument_template: Iterable[str], + *, + input_path: str | Path | None = None, + output_path: str | Path | None = None, + timeout: float = 300, + input_kind: str = "file", + output_kind: str = "optional", + ) -> ToolResult: + command = self.build_command( + executable, + argument_template, + input_path=input_path, + output_path=output_path, + input_kind=input_kind, + output_kind=output_kind, + ) + source = self._resolve_input(input_path, input_kind) + working_directory = source.parent if source else Path(command[0]).parent + creation_flags = subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0 + started = time.monotonic() + + with self._lock: + if self._process is not None: + raise ExternalToolError("Another external tool is already running") + self._cancel_requested = False + try: + self._process = subprocess.Popen( + command, + cwd=working_directory, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + shell=False, + creationflags=creation_flags, + ) + except OSError as exc: + raise ExternalToolError(f"Could not start external tool: {exc}") from exc + process = self._process + + try: + stdout, stderr = process.communicate(timeout=max(1, timeout)) + except subprocess.TimeoutExpired as exc: + process.kill() + stdout, stderr = process.communicate() + raise ExternalToolError( + f"External tool exceeded the {timeout:g}-second timeout" + ) from exc + finally: + with self._lock: + cancelled = self._cancel_requested + self._process = None + + return ToolResult( + command, + process.returncode, + stdout, + stderr, + time.monotonic() - started, + cancelled, + ) + + def cancel(self) -> bool: + """Terminate the active process, returning whether one was running.""" + with self._lock: + if self._process is None: + return False + self._cancel_requested = True + self._process.terminate() + return True + + @staticmethod + def _resolve_input(value: str | Path | None, kind: str = "file") -> Path | None: + if kind not in {"file", "directory", "any", "optional", "none"}: + raise ExternalToolError(f"Unsupported input path kind: {kind}") + if kind == "none": + return None + if value is None or not str(value).strip(): + if kind not in {"none", "optional"}: + raise ExternalToolError("This command requires an input path") + return None + path = Path(value).expanduser().resolve() + if kind == "file" and not path.is_file(): + raise ExternalToolError(f"Input file was not found: {path}") + if kind == "directory" and not path.is_dir(): + raise ExternalToolError(f"Input folder was not found: {path}") + if kind in {"any", "optional"} and not path.exists(): + raise ExternalToolError(f"Input path was not found: {path}") + return path + + @staticmethod + def _resolve_output(value: str | Path | None, kind: str = "file") -> Path | None: + if kind not in {"file", "directory", "optional", "none"}: + raise ExternalToolError(f"Unsupported output path kind: {kind}") + if kind == "none": + return None + if value is None or not str(value).strip(): + if kind not in {"none", "optional"}: + raise ExternalToolError("This command requires an output path") + return None + path = Path(value).expanduser().resolve() + if kind == "directory" and not path.is_dir(): + raise ExternalToolError(f"Output folder was not found: {path}") + if kind != "directory" and not path.parent.is_dir(): + raise ExternalToolError(f"Output folder was not found: {path.parent}") + return path + __all__ = [ "ExternalToolError", "ExternalToolRunner", + "ToolLaunch", + "ToolResult", "format_command", "split_arguments", ] -