diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 622a223..3627766 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -19,7 +19,7 @@ jobs: strategy: fail-fast: false matrix: - os: [ubuntu-latest, windows-latest] + os: [ubuntu-latest, windows-latest, macos-latest] python: ["3.10", "3.12"] steps: @@ -57,6 +57,10 @@ jobs: sh -n packaging/linux/install.sh sh -n packaging/linux/uninstall.sh + - name: Validate macOS shell script + if: runner.os == 'macOS' + run: sh -n build_macos.sh + windows-build: name: Windows executable runs-on: windows-latest @@ -136,3 +140,35 @@ jobs: dist/UnityScraper-Linux-x86_64.tar.gz dist/UnityScraper-Linux-x86_64.tar.gz.sha256 if-no-files-found: error + + macos-build: + name: macOS application + runs-on: macos-14 + needs: test + + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + cache: pip + + - name: Install build dependencies + run: python -m pip install -r requirements.txt pyinstaller + + - name: Build macOS application + run: ./build_macos.sh + + - name: Validate application bundle + run: | + test -x dist/UnityScraper.app/Contents/MacOS/UnityScraper + plutil -lint dist/UnityScraper.app/Contents/Info.plist + + - uses: actions/upload-artifact@v4 + with: + name: UnityScraper-macOS-arm64 + path: | + dist/UnityScraper-macOS-arm64.tar.gz + dist/UnityScraper-macOS-arm64.tar.gz.sha256 + if-no-files-found: error diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 6b8e69a..db95f10 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -111,9 +111,30 @@ jobs: dist/UnityScraper-Linux-x86_64.tar.gz dist/UnityScraper-Linux-x86_64.tar.gz.sha256 + macos: + runs-on: macos-14 + needs: validate + + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + cache: pip + - name: Install build dependencies + run: python -m pip install -r requirements.txt pyinstaller + - name: Build macOS release + run: ./build_macos.sh + - uses: actions/upload-artifact@v4 + with: + name: release-macos + path: | + dist/UnityScraper-macOS-arm64.tar.gz + dist/UnityScraper-macOS-arm64.tar.gz.sha256 + publish: runs-on: ubuntu-latest - needs: [validate, windows, linux] + needs: [validate, windows, linux, macos] steps: - uses: actions/download-artifact@v4 diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index a388899..4bea9c5 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -22,6 +22,8 @@ for CLI and optional REST automation. SQLite is the durable local store. - `profile_gui.py` renders privacy-aware profile/save inventory and snapshots. - `collection_gui.py` renders collection analysis, matching, reports, and repair previews. +- `community_gui.py` renders unified search and the cross-domain community + workspaces. - `setup_wizard.py` handles first-run storage setup. GUI operations that can block are dispatched to background threads and return @@ -42,6 +44,12 @@ results to Tk's main loop. snapshots, and PC/console comparisons. - `database_migrations.py` applies additive schema versions and provides consistent SQLite backup/restore helpers. +- `community_services.py` coordinates guided sync plans, package workspaces, + artwork, disc and dedup audits, storage, plugins, recovery, compatibility, + and accessibility. +- `unified_search.py` ranks local results across the application domains. +- `structured_knowledge.py` extracts typed records from cached source documents + while retaining document and source relationships. ### Domain and Adapters @@ -69,6 +77,9 @@ Main schema groups: - Profiles: scan runs, profiles, saves, snapshots, snapshot files, GPD inventories, achievements, comparisons, Xenia migration runs, and auditable operations +- Community: structured records, guided sync plans, ownership previews, played + titles and images, save comparisons, artwork exports, disc and dedup audits, + storage audits, plugin state, recovery, compatibility, and accessibility Schema initialization is idempotent. New migrations should preserve existing data and be covered by tests. @@ -183,5 +194,6 @@ See [SECURITY.md](SECURITY.md) for reporting and operational guidance. `UnityScraper.spec` is the canonical cross-platform PyInstaller definition. Assets and modules loaded indirectly by the GUI are listed explicitly. GitHub -Actions validates Windows and Linux one-file builds on pull requests. Version -tags publish a Windows ZIP and Linux tarball with separate SHA-256 files. +Actions validates Windows and Linux one-file builds plus an unsigned macOS +Apple Silicon application bundle. Version tags publish platform archives with +separate SHA-256 files. diff --git a/CHANGELOG.md b/CHANGELOG.md index 4328463..6949a02 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,26 @@ Notable changes to UnityScraper are documented here. The project follows ### Added +- Community Hub with unified local search across games, source knowledge, + profiles, saves, achievements, files, and structured records. +- Additive schema migration 8 for structured knowledge, guided console plans, + ownership previews, GPD title/image records, save comparisons, artwork, + disc/dedup audits, storage and original Xbox records, plugins, recovery, + dashboard compatibility, and accessibility. +- Structured extraction of hardware, drive, dashboard, exploit, error, format, + repair, and tool records from cached source documents with provenance. +- Confirmed sync-plan queueing through the existing persistent transfer engine. +- Read-only package workspaces, profile dashboards, ownership previews, + block-level save comparison, played-title history, and validated GPD image + export. +- Artwork export presets, multi-disc audits, hash-based duplicate previews, and + recoverable duplicate quarantine/hardlink actions. +- Read-only FATX audits, original Xbox discovery, bounded plugin installation, + recovery actions, and dashboard FTP capability probes. +- High-contrast, large-text, reduced-motion, and keyboard-hint preferences. +- Unsigned macOS Apple Silicon application packaging, checksums, CI validation, + and documentation. +- Runtime repair for historically misencoded translation resources. - Read-only XDBF/GPD inspection with bounded entry parsing, achievement state, gamerscore summaries, extracted-folder discovery, and local inventory. - Profile comparison for save hashes and imported achievement state. @@ -63,6 +83,8 @@ Notable changes to UnityScraper are documented here. The project follows ### Changed +- Version advanced to `1.1.0-beta.1`. +- CI now runs the Python suite on Windows, Linux, and macOS. - Cached XboxUnity titles now resolve immediately in library lists and details, enrich matching rows page by page, and recover after interrupted refreshes. - Library rows now show `Unknown game` instead of duplicating the TitleID when @@ -70,7 +92,6 @@ Notable changes to UnityScraper are documented here. The project follows - 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 before staging it. diff --git a/COMMUNITY_HUB.md b/COMMUNITY_HUB.md new file mode 100644 index 0000000..4dac454 --- /dev/null +++ b/COMMUNITY_HUB.md @@ -0,0 +1,78 @@ +# Community Hub + +The Community Hub brings the wider Xbox and preservation workflows into one +source-attributed, offline-capable workspace. Open it from the desktop sidebar +or press `Ctrl+K` to focus unified search. + +## Twenty Integrated Capabilities + +1. Unified local search across games, identifiers, wiki knowledge, profiles, + saves, achievements, files, and structured reference records. +2. Structured extraction of motherboards, DVD drives, dashboards, exploits, + errors, formats, repairs, and tools from cached source documents. +3. Guided PC-to-console comparison using a captured, read-only console + inventory and Aurora, Freestyle Dash, XeXMenu, or stock layouts. +4. Explicit queueing of revalidated uploads into the existing persistent, + resumable transfer queue. A preview never starts a transfer by itself. +5. Profile dashboards summarizing saves, played titles, achievements, and + gamerscore from local data. +6. Read-only STFS package inspection and auditable package workspaces that keep + an untouched original and manifest. +7. Ownership-migration previews that record intended profile, console, and + device changes without changing or signing the package. +8. Block-level save comparisons with SHA-256 results and durable audit history. +9. XDBF/GPD played-title history and bounded embedded-image discovery. +10. Validated export of an embedded GPD image without modifying its source. +11. Preferred artwork selection with Aurora, Freestyle Dash, and preservation + archive export layouts plus checksum manifests. +12. Multi-disc completeness audits based on scanned disc number and count. +13. Duplicate previews using size grouping and SHA-256 verification. +14. Recoverable duplicate actions that quarantine the original and can create + a verified hardlink only after a second hash check. +15. Read-only FATX signature and mounted-storage audits. +16. Original Xbox `default.xbe` discovery alongside Xbox 360 collections. +17. Plugin discovery, checksums, permission display, enable/disable state, and + bounded ZIP installation or update with rollback. +18. Recovery scans for partial files, interrupted jobs, incomplete snapshots, + and failed operations, with conservative retry or quarantine actions. +19. Dashboard compatibility probes for login, content-root access, resume, and + advertised remote hash support on a trusted local FTP network. +20. Cross-platform accessibility and packaging: scalable text, high contrast, + reduced-motion preferences, keyboard hints, and Windows, Linux, and macOS + build paths. + +## Safety Rules + +- Sync plans are previews until the user confirms queueing. Queued jobs still + run through the normal transfer controls. +- Duplicate cleanup never deletes the only retained copy. Quarantined files + remain under `.unityscraper-dedup-quarantine` for manual recovery. +- FATX images are detected read-only. Raw-device and raw-image writes are not + implemented. +- Package workspaces and ownership changes are previews. CON/LIVE/PIRS rebuild, + rehash, signature, and ownership mutation remain disabled until independent + verification and recovery are complete. +- Traditional console FTP is unencrypted and is intended only for a trusted + local network. +- XboxUnity remains HTTP-only. This application does not invent or prefer an + HTTPS endpoint for XboxUnity. + +## Data and Provenance + +Migration 8 adds durable records for structured knowledge, sync plans, profile +previews, played titles, embedded images, save comparisons, artwork exports, +disc audits, dedup plans, storage audits, original Xbox records, plugin state, +recovery events, dashboard tests, and accessibility preferences. It is additive +and preserves existing databases. + +Imported ConsoleMods, XenonLibrary, Free60, Redump, and No-Intro information +continues to retain source, revision, citation, licensing, and conflict data. +Redump and No-Intro DAT files remain user-supplied; copyrighted game content is +never included. + +## Platform Notes + +Windows and Linux remain the primary tested release targets. macOS CI builds an +unsigned Apple Silicon application bundle and checksum. The macOS artifact is +not notarized, and the documentation does not ask users to disable Gatekeeper. +See [MACOS.md](MACOS.md) for current limitations. diff --git a/DOCS_INDEX.md b/DOCS_INDEX.md index 45cf20c..bb50305 100644 --- a/DOCS_INDEX.md +++ b/DOCS_INDEX.md @@ -20,6 +20,10 @@ - [REST API](API.md) - authentication, endpoints, configuration, and safety - [Linux Support](LINUX.md) - installation, XDG storage, desktop integration, uninstallation, and troubleshooting +- [macOS Preview](MACOS.md) - Apple Silicon bundle, source setup, checksums, + and signing limitations +- [Community Hub](COMMUNITY_HUB.md) - unified search, profile/package tools, + guided console plans, preservation, plugins, recovery, and safety boundaries - [Project Status](PROJECT_STATUS.md) - completed work, boundaries, and roadmap - [Changelog](CHANGELOG.md) - release history @@ -53,3 +57,9 @@ Linux packaging: ```bash ./build_linux.sh ``` + +macOS packaging: + +```bash +./build_macos.sh +``` diff --git a/MACOS.md b/MACOS.md new file mode 100644 index 0000000..06d38c8 --- /dev/null +++ b/MACOS.md @@ -0,0 +1,32 @@ +# macOS Support + +UnityScraper supports macOS 11 or newer from source on Intel and Apple silicon +through the same Tk desktop application and native Application Support, Caches, +and Logs paths. + +Current preview release archives are built for Apple silicon and contain +`UnityScraper.app` plus a SHA-256 checksum. They are currently unsigned and not +notarized, so macOS may require an explicit **Open** confirmation from Finder. +The project does not ask users to disable Gatekeeper. Intel users should run +from source until a universal release artifact is available. + +## Source setup + +Install Python 3.10 or newer with Tk support, then run: + +```sh +python3 -m venv .venv +.venv/bin/python -m pip install -r requirements.txt +.venv/bin/python desktop_app.py +``` + +## Build + +```sh +chmod +x build_macos.sh +./build_macos.sh +``` + +Signing and notarization can be enabled by a release maintainer when an Apple +Developer ID certificate and notarization credentials are configured as +repository secrets. Those credentials are never stored in the repository. diff --git a/PROJECT_STATUS.md b/PROJECT_STATUS.md index f545b1f..1613b67 100644 --- a/PROJECT_STATUS.md +++ b/PROJECT_STATUS.md @@ -56,6 +56,12 @@ backup-management, and source-attributed knowledge application. launch scripts, and release artifacts. - Repository contribution, security, architecture, API, and release documentation. +- Community Hub with unified search, typed knowledge extraction, guided console + upload plans, package/profile workspaces, artwork and disc management, + recoverable deduplication, storage audits, original Xbox discovery, plugin + controls, recovery actions, dashboard probes, and accessibility preferences. +- Additive schema migration 8 and local audit history for every new workspace. +- Windows, Linux, and unsigned Apple Silicon macOS CI packaging. ## Validation @@ -65,7 +71,7 @@ backup-management, and source-attributed knowledge application. end-to-end local workflow. - Network-backed source syncs remain dependent on each source's availability and access policy. Cached copies are used when available. -- Windows and Linux executable artifacts are generated by CI and releases +- Windows, Linux, and macOS artifacts are generated by CI and releases rather than committed to the source tree. ## Deliberate Boundaries @@ -86,12 +92,14 @@ backup-management, and source-attributed knowledge application. - Profile editing, achievement modification, ownership rewriting, and CON re-signing remain disabled until complete package verification and reliable cross-platform signing support are available. +- FATX and raw-device access remains read-only. Duplicate actions retain a + quarantine copy, and console plans require explicit queue confirmation. ## Future Work - Validate console resume and optional hash behavior against a broader matrix of real dashboard FTP servers. -- Expand read-only GPD coverage with dashboard title-history and safe image - previews after adding a decompression and image validation boundary. +- Add notarization and universal binaries after macOS signing infrastructure is + available. - Consider package mutation only after complete STFS extraction, rehashing, signing, verification, and automatic recovery have independent test vectors. diff --git a/README.md b/README.md index b12d39c..55d7870 100644 --- a/README.md +++ b/README.md @@ -109,6 +109,23 @@ checksums, platform notes, and safety guidance. - Exports preservation manifests, offline HTML reports, and fact provenance. - Keeps local metadata overrides separate from source-attributed knowledge. +### Community Hub + +- Searches games, knowledge, profiles, saves, achievements, files, and tools + together from the local database. +- Extracts structured hardware, dashboard, exploit, error, format, repair, and + tool records from cached source articles while retaining provenance. +- Builds console sync previews and queues confirmed uploads through the durable, + resumable transfer engine. +- Adds profile dashboards, read-only package workspaces, ownership previews, + save comparison, played-title history, and validated GPD image export. +- Manages preferred artwork, multi-disc audits, recoverable duplicate cleanup, + read-only FATX inspection, original Xbox discovery, plugins, and recovery. +- Stores high-contrast, large-text, reduced-motion, and keyboard-hint settings. + +See [COMMUNITY_HUB.md](COMMUNITY_HUB.md) for all twenty capabilities and their +safety boundaries. + ## Install ### Windows Release @@ -136,6 +153,12 @@ creates `~/.local/bin/unityscraper`. It does not require root access. See [LINUX.md](LINUX.md) for supported distributions, XDG paths, source setup, uninstallation, and troubleshooting. +### macOS Preview + +CI produces an unsigned Apple Silicon `.app` archive and SHA-256 checksum. It +is currently a preview artifact and is not notarized. See [MACOS.md](MACOS.md) +for installation, source setup, and known limitations. + ### Run From Source Requirements: @@ -180,6 +203,7 @@ Linux source setup: | External Tools | Run XeXTool and other user-supplied command-line utilities | | Collections | Identify storage, compare Title Updates, verify preservation data, and preview repairs | | Knowledge | Search sources, facts, citations, imports, and conflicts | +| Community Hub | Unified search, console plans, profiles, preservation, plugins, recovery, and compatibility | | Archive Health | Find missing or inconsistent downloaded files | | Settings | Configure storage and scraper behavior | | Help & About | Version, diagnostics, storage, and advanced tools | @@ -353,11 +377,19 @@ Build the Linux release bundle on Linux: The output is `dist/UnityScraper-Linux-.tar.gz` with a matching SHA-256 file. +Build the macOS application bundle on macOS: + +```bash +./build_macos.sh +``` + ## Documentation - [Documentation index](DOCS_INDEX.md) - [Architecture](ARCHITECTURE.md) - [Linux support](LINUX.md) +- [macOS preview](MACOS.md) +- [Community Hub](COMMUNITY_HUB.md) - [Knowledge sources and licensing](KNOWLEDGE_SOURCES.md) - [Backup manager](BACKUP_MANAGER.md) - [Collection intelligence](COLLECTION_INTELLIGENCE.md) diff --git a/SECURITY.md b/SECURITY.md index afba2ba..2d3e8a4 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -39,6 +39,15 @@ filesystem contents in a report. files and write the restored copy alongside them. - GPD parsing validates bounded entry tables and file offsets and never writes achievement, setting, sync, or image records. +- GPD image export validates the selected bounded payload and publishes through + a temporary file without changing the source GPD. +- Duplicate actions revalidate both hashes and retain the removed path in a + local quarantine; hardlinks are created only after quarantine succeeds. +- FATX image support is detection-only. Raw image and device writes are not + available. +- Package ownership changes and signed-package rebuilds remain preview-only. +- Plugin ZIPs have entry and expanded-size limits, reject traversal paths, and + are installed disabled for explicit review. - Xenia migrations require a preview and verified snapshot, publish through partial files, and never overwrite different destination data. - Optional remote hashes use read-only FTP commands and fail closed when the diff --git a/UnityScraper.spec b/UnityScraper.spec index 385c7fb..5de0ef5 100644 --- a/UnityScraper.spec +++ b/UnityScraper.spec @@ -1,9 +1,18 @@ # -*- mode: python ; coding: utf-8 -*- import sys +from pathlib import Path icon = 'assets/UnityScraper.ico' if sys.platform == 'win32' else None +tk_datas = [] +if sys.platform == 'win32': + tcl_root = Path(sys.base_prefix) / 'tcl' / 'tcl8.6' + tk_root = Path(sys.base_prefix) / 'tcl' / 'tk8.6' + if tcl_root.is_dir(): + tk_datas.append((str(tcl_root), '_tcl_data')) + if tk_root.is_dir(): + tk_datas.append((str(tk_root), '_tk_data')) a = Analysis( ['desktop_app.py'], @@ -14,13 +23,15 @@ a = Analysis( ('VERSION', '.'), ('THIRD_PARTY_NOTICES.md', '.'), ('assets', 'assets'), - ], + ] + tk_datas, hiddenimports=[ 'backup_gui', 'backup_manager', 'backup_service', 'collection_gui', 'collection_intelligence', + 'community_gui', + 'community_services', 'console_sync', 'consolemods_adapters', 'database_migrations', @@ -29,6 +40,8 @@ a = Analysis( 'knowledge_service', 'knowledge_scheduler', 'knowledge_sync', + 'structured_knowledge', + 'unified_search', 'plugins', 'gpd_parser', 'profile_gui', diff --git a/VERSION b/VERSION index 9a7c866..bc85f78 100644 --- a/VERSION +++ b/VERSION @@ -1,5 +1,5 @@ { - "version": "1.0.0b1", + "version": "1.1.0b1", "name": "Unified Xbox 360 Collection and Preservation Manager", "changes": [ "Unified Xbox 360 knowledge browser", @@ -16,7 +16,11 @@ "Redump and No-Intro local-file hash matching", "Offline HTML reports, manifests, provenance, and metadata overrides", "Versioned migrations, database backups, queue recovery, and plugin API v1", - "Platform-aware verified updates, SBOMs, and build attestations" + "Platform-aware verified updates, SBOMs, and build attestations", + "Unified Community Hub across games, knowledge, profiles, preservation, and recovery", + "Guided console sync plans and recoverable duplicate cleanup", + "Played-title history and validated embedded GPD image export", + "Windows, Linux, and unsigned Apple Silicon macOS build paths" ], "download_url": "https://github.com/TrapEmAll/UnityScraper/releases", "release_date": "2026-07-23" diff --git a/app_version.py b/app_version.py index 35497f9..43919b1 100644 --- a/app_version.py +++ b/app_version.py @@ -1,4 +1,4 @@ """Single source of truth for UnityScraper version information.""" -APP_VERSION = "1.0.0b1" -DISPLAY_VERSION = "1.0.0-beta.1" +APP_VERSION = "1.1.0b1" +DISPLAY_VERSION = "1.1.0-beta.1" diff --git a/build_macos.sh b/build_macos.sh new file mode 100755 index 0000000..44b9f44 --- /dev/null +++ b/build_macos.sh @@ -0,0 +1,26 @@ +#!/usr/bin/env sh +set -eu + +cd "$(dirname "$0")" + +PYTHON="${PYTHON:-python3}" +if [ -x ".venv/bin/python" ]; then + PYTHON=".venv/bin/python" +fi + +"$PYTHON" -m pip install -r requirements.txt pyinstaller +"$PYTHON" -m PyInstaller --clean --noconfirm UnityScraper.spec + +ARCH="$(uname -m)" +APP="dist/UnityScraper.app" +ARCHIVE="dist/UnityScraper-macOS-${ARCH}.tar.gz" + +rm -rf "$APP" +mkdir -p "$APP/Contents/MacOS" "$APP/Contents/Resources" +install -m 0755 dist/UnityScraper "$APP/Contents/MacOS/UnityScraper" +install -m 0644 packaging/macos/Info.plist "$APP/Contents/Info.plist" +install -m 0644 assets/UnityScraper.png "$APP/Contents/Resources/UnityScraper.png" +tar -C dist -czf "$ARCHIVE" UnityScraper.app +shasum -a 256 "$ARCHIVE" > "$ARCHIVE.sha256" + +printf '\nBuild complete: %s\nChecksum: %s.sha256\n' "$ARCHIVE" "$ARCHIVE" diff --git a/community_gui.py b/community_gui.py new file mode 100644 index 0000000..4412d73 --- /dev/null +++ b/community_gui.py @@ -0,0 +1,527 @@ +"""Integrated desktop workspace for the community roadmap services.""" + +from __future__ import annotations + +import json +import tkinter as tk +from pathlib import Path +from tkinter import filedialog, messagebox, ttk +from typing import Any, Callable + +from app_paths import DATABASE_PATH, DOWNLOADS_DIR, PROFILE_BACKUPS_DIR, executable_root +from backup_manager import FtpTarget +from community_services import ( + DASHBOARD_PRESETS, + AccessibilityService, + ArtworkService, + ConsolePlanService, + DashboardCompatibilityService, + PackageWorkspaceService, + PluginControlService, + PreservationPlanningService, + RecoveryService, + StorageAndXboxService, +) +from profile_intelligence import ProfileIntelligenceService +from structured_knowledge import StructuredKnowledgeService +from unified_search import UnifiedSearchService + + +class CommunityHubPage: + """One operational surface for cross-domain community workflows.""" + + def __init__(self, root: tk.Tk, parent: ttk.Frame, page_header: Callable) -> None: + self.root = root + self.parent = parent + self.search_service = UnifiedSearchService() + self.structured = StructuredKnowledgeService() + self.console_plans = ConsolePlanService() + self.packages = PackageWorkspaceService() + self.profiles = ProfileIntelligenceService() + self.artwork = ArtworkService() + self.preservation = PreservationPlanningService() + self.storage = StorageAndXboxService() + self.plugins = PluginControlService() + self.recovery = RecoveryService() + self.compatibility = DashboardCompatibilityService() + self.accessibility = AccessibilityService() + page_header("Community Hub", "Search, organize, preserve, and safely plan console changes.") + self._build() + + def _build(self) -> None: + notebook = ttk.Notebook(self.parent) + notebook.grid(row=1, column=0, sticky="nsew") + self.parent.rowconfigure(1, weight=1) + for label, builder in ( + ("Search", self._build_search), + ("Knowledge", self._build_knowledge), + ("Console Sync", self._build_sync), + ("Profiles", self._build_profiles), + ("Preservation", self._build_preservation), + ("Storage", self._build_storage), + ("Plugins", self._build_plugins), + ("Recovery", self._build_recovery), + ("Compatibility", self._build_compatibility), + ("Accessibility", self._build_accessibility), + ): + frame = ttk.Frame(notebook, padding=12) + notebook.add(frame, text=label) + builder(frame) + + def _build_search(self, frame: ttk.Frame) -> None: + frame.columnconfigure(0, weight=1) + frame.rowconfigure(2, weight=1) + self.search_var = tk.StringVar() + self.search_entry = ttk.Entry(frame, textvariable=self.search_var) + self.search_entry.grid(row=0, column=0, sticky="ew", padx=(0, 8)) + self.search_entry.bind("", lambda _event: self._run_search()) + ttk.Button(frame, text="Search", command=self._run_search).grid(row=0, column=1) + ttk.Label(frame, text="Games, knowledge, profiles, saves, achievements, files, and tools").grid( + row=1, column=0, columnspan=2, sticky="w", pady=(6, 10) + ) + self.search_tree = ttk.Treeview( + frame, columns=("category", "title", "details"), show="headings" + ) + for column, heading, width in ( + ("category", "Type", 110), ("title", "Result", 280), + ("details", "Details", 470), + ): + self.search_tree.heading(column, text=heading) + self.search_tree.column(column, width=width, stretch=column != "category") + self.search_tree.grid(row=2, column=0, columnspan=2, sticky="nsew") + + def focus_search(self) -> None: + self.search_entry.focus_set() + + def _run_search(self) -> None: + self._fill_tree( + self.search_tree, + self.search_service.search(self.search_var.get()), + lambda row: (row["category"], row["title"], row["subtitle"]), + ) + + def _build_knowledge(self, frame: ttk.Frame) -> None: + frame.columnconfigure(0, weight=1) + frame.rowconfigure(1, weight=1) + controls = ttk.Frame(frame) + controls.grid(row=0, column=0, sticky="ew", pady=(0, 10)) + ttk.Button(controls, text="Extract Cached Articles", command=self._extract_knowledge).pack(side=tk.LEFT) + self.knowledge_type = tk.StringVar(value="") + ttk.Combobox( + controls, textvariable=self.knowledge_type, state="readonly", width=20, + values=("", "motherboard", "dvd_drive", "dashboard", "exploit", "error_code", + "file_format", "repair", "tool", "reference_article"), + ).pack(side=tk.LEFT, padx=8) + ttk.Button(controls, text="Refresh", command=self._refresh_knowledge).pack(side=tk.LEFT) + self.knowledge_tree = ttk.Treeview( + frame, columns=("type", "name", "source"), show="headings" + ) + for column, heading, width in (("type", "Type", 130), ("name", "Name", 420), + ("source", "Source", 220)): + self.knowledge_tree.heading(column, text=heading) + self.knowledge_tree.column(column, width=width) + self.knowledge_tree.grid(row=1, column=0, sticky="nsew") + + def _extract_knowledge(self) -> None: + self._run("Knowledge extraction", lambda: self.structured.extract_cached_documents()) + self._refresh_knowledge() + + def _refresh_knowledge(self) -> None: + rows = self.structured.list_records(self.knowledge_type.get()) + self._fill_tree(self.knowledge_tree, rows, + lambda row: (row["record_type"], row["canonical_name"], row["source_name"])) + + def _build_sync(self, frame: ttk.Frame) -> None: + frame.columnconfigure(1, weight=1) + self.sync_root = tk.StringVar() + self.sync_snapshot = tk.StringVar() + self.sync_target = tk.StringVar() + self.sync_dashboard = tk.StringVar(value="aurora") + self._path_row(frame, 0, "Local content root", self.sync_root, directory=True) + ttk.Label(frame, text="Console snapshot ID").grid(row=1, column=0, sticky="w", pady=5) + ttk.Entry(frame, textvariable=self.sync_snapshot).grid(row=1, column=1, sticky="ew", pady=5) + ttk.Label(frame, text="Dashboard").grid(row=2, column=0, sticky="w", pady=5) + ttk.Combobox(frame, textvariable=self.sync_dashboard, state="readonly", + values=tuple(DASHBOARD_PRESETS)).grid(row=2, column=1, sticky="w", pady=5) + ttk.Label(frame, text="Saved console target ID (optional)").grid( + row=3, column=0, sticky="w", pady=5 + ) + ttk.Entry(frame, textvariable=self.sync_target).grid( + row=3, column=1, sticky="ew", pady=5 + ) + controls = ttk.Frame(frame) + controls.grid(row=4, column=1, sticky="w", pady=10) + ttk.Button(controls, text="Create Sync Preview", command=self._create_sync_plan).pack( + side=tk.LEFT + ) + self.queue_sync_button = ttk.Button( + controls, text="Queue Previewed Uploads", command=self._queue_sync_plan, + state=tk.DISABLED, + ) + self.queue_sync_button.pack(side=tk.LEFT, padx=8) + self.sync_output = self._output(frame, 5, 2) + self.current_sync_plan_id: int | None = None + + def _create_sync_plan(self) -> None: + result = self._run("Sync preview", lambda: self.console_plans.create_plan( + self.sync_root.get(), int(self.sync_snapshot.get()), self.sync_dashboard.get() + )) + self._show_output(self.sync_output, result) + if result is not None: + self.current_sync_plan_id = int(result["plan_id"]) + self.queue_sync_button.configure( + state=tk.NORMAL if result["summary"]["uploads"] else tk.DISABLED + ) + + def _queue_sync_plan(self) -> None: + if self.current_sync_plan_id is None: + return + if not messagebox.askyesno( + "Queue console uploads", + "Queue every selected upload from this preview? Transfers will remain paused " + "in the normal console queue until you run them.", + parent=self.root, + ): + return + target = self.sync_target.get().strip() + result = self._run( + "Queue sync plan", + lambda: self.console_plans.queue_uploads( + self.current_sync_plan_id, int(target) if target else None + ), + ) + self._show_output(self.sync_output, result) + if result is not None: + self.queue_sync_button.configure(state=tk.DISABLED) + + def _build_profiles(self, frame: ttk.Frame) -> None: + frame.columnconfigure(1, weight=1) + self.profile_id = tk.StringVar() + ttk.Label(frame, text="Profile ID").grid(row=0, column=0, sticky="w", pady=4) + ttk.Entry(frame, textvariable=self.profile_id).grid(row=0, column=1, sticky="ew", pady=4) + ttk.Button(frame, text="Profile Dashboard", command=self._profile_dashboard).grid( + row=0, column=2, padx=(8, 0) + ) + self.package_path = tk.StringVar() + self._path_row(frame, 1, "STFS package", self.package_path, directory=False) + buttons = ttk.Frame(frame) + buttons.grid(row=2, column=1, sticky="w", pady=8) + ttk.Button(buttons, text="Inspect Package", command=self._inspect_package).pack(side=tk.LEFT) + ttk.Button(buttons, text="Create Read-only Workspace", command=self._package_workspace).pack( + side=tk.LEFT, padx=8 + ) + ttk.Button(buttons, text="Ownership Migration Preview", command=self._ownership_preview).pack( + side=tk.LEFT + ) + self.compare_left = tk.StringVar() + self.compare_right = tk.StringVar() + self._path_row(frame, 3, "Compare save A", self.compare_left, directory=False) + self._path_row(frame, 4, "Compare save B", self.compare_right, directory=False) + ttk.Button(frame, text="Compare Saves", command=self._compare_saves).grid( + row=5, column=1, sticky="w", pady=8 + ) + self.profile_output = self._output(frame, 6, 3) + + def _profile_dashboard(self) -> None: + self._show_output(self.profile_output, self._run( + "Profile dashboard", lambda: self.profiles.profile_dashboard(self.profile_id.get()))) + + def _inspect_package(self) -> None: + self._show_output(self.profile_output, self._run( + "Package inspection", lambda: self.packages.inspect(self.package_path.get()))) + + def _package_workspace(self) -> None: + destination = filedialog.askdirectory(parent=self.root, title="Choose package workspace") + if destination: + self._show_output(self.profile_output, self._run( + "Package workspace", lambda: {"manifest": str( + self.packages.create_workspace(self.package_path.get(), destination))})) + + def _ownership_preview(self) -> None: + self._show_output(self.profile_output, self._run( + "Ownership preview", lambda: self.profiles.preview_ownership_migration( + self.profile_id.get(), self.package_path.get()))) + + def _compare_saves(self) -> None: + self._show_output(self.profile_output, self._run( + "Save comparison", lambda: self.profiles.compare_save_files( + self.compare_left.get(), self.compare_right.get()))) + + def _build_preservation(self, frame: ttk.Frame) -> None: + frame.columnconfigure(1, weight=1) + self.art_titleid = tk.StringVar() + self.art_path = tk.StringVar() + ttk.Label(frame, text="Artwork TitleID").grid(row=0, column=0, sticky="w", pady=4) + ttk.Entry(frame, textvariable=self.art_titleid).grid(row=0, column=1, sticky="ew", pady=4) + self._path_row(frame, 1, "Preferred artwork", self.art_path, directory=False) + controls = ttk.Frame(frame) + controls.grid(row=2, column=1, sticky="w", pady=8) + ttk.Button(controls, text="Set Artwork", command=self._set_artwork).pack(side=tk.LEFT) + ttk.Button(controls, text="Export Artwork", command=self._export_artwork).pack(side=tk.LEFT, padx=8) + ttk.Button(controls, text="Audit Disc Sets", command=self._audit_discs).pack(side=tk.LEFT) + self.dedup_root = tk.StringVar() + self._path_row(frame, 3, "Duplicate scan root", self.dedup_root, directory=True) + dedup_controls = ttk.Frame(frame) + dedup_controls.grid(row=4, column=1, sticky="w", pady=8) + ttk.Button(dedup_controls, text="Create Dedup Preview", command=self._dedup).pack( + side=tk.LEFT + ) + self.dedup_action_id = tk.StringVar() + ttk.Label(dedup_controls, text="Action ID").pack(side=tk.LEFT, padx=(14, 4)) + ttk.Entry(dedup_controls, textvariable=self.dedup_action_id, width=8).pack(side=tk.LEFT) + self.dedup_mode = tk.StringVar(value="quarantine") + ttk.Combobox( + dedup_controls, textvariable=self.dedup_mode, state="readonly", width=11, + values=("quarantine", "hardlink"), + ).pack(side=tk.LEFT, padx=4) + ttk.Button(dedup_controls, text="Apply Safely", command=self._apply_dedup).pack( + side=tk.LEFT + ) + self.preservation_output = self._output(frame, 5, 3) + + def _set_artwork(self) -> None: + self._show_output(self.preservation_output, self._run( + "Artwork preference", lambda: self.artwork.set_preference( + self.art_titleid.get(), self.art_path.get()))) + + def _export_artwork(self) -> None: + destination = filedialog.askdirectory(parent=self.root, title="Choose artwork export folder") + if destination: + self._show_output(self.preservation_output, self._run( + "Artwork export", lambda: self.artwork.export(destination, "aurora"))) + + def _audit_discs(self) -> None: + self._show_output(self.preservation_output, self._run( + "Disc set audit", self.preservation.audit_disc_sets)) + + def _dedup(self) -> None: + self._show_output(self.preservation_output, self._run( + "Dedup preview", lambda: self.preservation.create_dedup_plan(self.dedup_root.get()))) + + def _apply_dedup(self) -> None: + if not messagebox.askyesno( + "Apply duplicate action", + "Revalidate this duplicate and move its original into the recovery quarantine?", + parent=self.root, + ): + return + self._show_output(self.preservation_output, self._run( + "Duplicate action", lambda: self.preservation.apply_dedup_action( + int(self.dedup_action_id.get()), self.dedup_mode.get()))) + + def _build_storage(self, frame: ttk.Frame) -> None: + frame.columnconfigure(1, weight=1) + self.storage_path = tk.StringVar() + self._path_row(frame, 0, "Mounted storage or image", self.storage_path, directory=False, + allow_directory=True) + ttk.Button(frame, text="Read-only Storage Audit", command=self._storage_audit).grid( + row=1, column=1, sticky="w", pady=8 + ) + self.xbox_root = tk.StringVar() + self._path_row(frame, 2, "Original Xbox games root", self.xbox_root, directory=True) + ttk.Button(frame, text="Scan Original Xbox Games", command=self._scan_xbox).grid( + row=3, column=1, sticky="w", pady=8 + ) + self.storage_output = self._output(frame, 4, 3) + + def _storage_audit(self) -> None: + self._show_output(self.storage_output, self._run( + "Storage audit", lambda: self.storage.audit_storage(self.storage_path.get()))) + + def _scan_xbox(self) -> None: + self._show_output(self.storage_output, self._run( + "Original Xbox scan", lambda: self.storage.scan_original_xbox(self.xbox_root.get()))) + + def _build_plugins(self, frame: ttk.Frame) -> None: + frame.columnconfigure(1, weight=1) + frame.rowconfigure(2, weight=1) + self.plugin_root = tk.StringVar(value=str(executable_root() / "plugins")) + self._path_row(frame, 0, "Plugin folder", self.plugin_root, directory=True) + controls = ttk.Frame(frame) + controls.grid(row=1, column=1, sticky="w", pady=8) + ttk.Button(controls, text="Discover", command=self._discover_plugins).pack(side=tk.LEFT) + ttk.Button(controls, text="Install or Update", command=self._install_plugin).pack(side=tk.LEFT, padx=8) + ttk.Button(controls, text="Enable", command=lambda: self._set_plugin(True)).pack(side=tk.LEFT) + ttk.Button(controls, text="Disable", command=lambda: self._set_plugin(False)).pack(side=tk.LEFT, padx=8) + self.plugin_tree = ttk.Treeview( + frame, columns=("id", "version", "enabled", "trusted", "permissions"), show="headings" + ) + for column, label, width in (("id", "Plugin", 180), ("version", "Version", 90), + ("enabled", "Enabled", 75), ("trusted", "Checksum", 90), + ("permissions", "Permissions", 280)): + self.plugin_tree.heading(column, text=label) + self.plugin_tree.column(column, width=width) + self.plugin_tree.grid(row=2, column=0, columnspan=3, sticky="nsew") + self.plugin_rows: dict[str, dict[str, Any]] = {} + + def _discover_plugins(self) -> None: + rows = self._run("Plugin discovery", lambda: self.plugins.discover(self.plugin_root.get())) + if rows is None: + return + self.plugin_rows.clear() + self.plugin_tree.delete(*self.plugin_tree.get_children()) + for row in rows: + plugin_id = row["id"] + self.plugin_rows[plugin_id] = row + self.plugin_tree.insert("", tk.END, iid=plugin_id, values=( + row.get("name", plugin_id), row.get("version", ""), + "Yes" if row.get("enabled") else "No", + "Trusted" if row.get("trusted") else "Review", + ", ".join(row.get("permissions", [])), + )) + + def _install_plugin(self) -> None: + archive = filedialog.askopenfilename( + parent=self.root, title="Choose a plugin ZIP", filetypes=(("ZIP archives", "*.zip"),) + ) + if archive: + self._run("Plugin installation", lambda: self.plugins.install_package( + archive, self.plugin_root.get())) + self._discover_plugins() + + def _set_plugin(self, enabled: bool) -> None: + selection = self.plugin_tree.selection() + if not selection: + messagebox.showinfo("Plugins", "Select a plugin first.", parent=self.root) + return + plugin_id = selection[0] + row = self.plugin_rows[plugin_id] + manifest_path = Path(self.plugin_root.get()) / plugin_id / "plugin.json" + try: + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + entry = manifest_path.parent / manifest["entrypoint"] + self.plugins.set_state(plugin_id, enabled, entry, row.get("permissions", [])) + except Exception as exc: + messagebox.showerror("Plugins", str(exc), parent=self.root) + return + self._discover_plugins() + + def _build_recovery(self, frame: ttk.Frame) -> None: + frame.columnconfigure(0, weight=1) + frame.rowconfigure(1, weight=1) + controls = ttk.Frame(frame) + controls.grid(row=0, column=0, sticky="ew", pady=(0, 8)) + ttk.Button(controls, text="Scan Recovery State", command=self._scan_recovery).pack(side=tk.LEFT) + ttk.Button(controls, text="Recover Selected", command=self._recover_selected).pack(side=tk.LEFT, padx=8) + self.recovery_tree = ttk.Treeview( + frame, columns=("type", "source", "detected"), show="headings" + ) + for column, label, width in (("type", "Issue", 160), ("source", "Source", 500), + ("detected", "Detected", 190)): + self.recovery_tree.heading(column, text=label) + self.recovery_tree.column(column, width=width) + self.recovery_tree.grid(row=1, column=0, sticky="nsew") + + def _scan_recovery(self) -> None: + roots = (DOWNLOADS_DIR, PROFILE_BACKUPS_DIR, DATABASE_PATH.parent) + result = self._run("Recovery scan", lambda: self.recovery.scan(roots)) + if result is not None: + self._refresh_recovery() + + def _refresh_recovery(self) -> None: + self.recovery_tree.delete(*self.recovery_tree.get_children()) + for row in self.recovery.list_open(): + self.recovery_tree.insert("", tk.END, iid=str(row["id"]), values=( + row["event_type"].replace("_", " ").title(), row["source"], row["detected_at"] + )) + + def _recover_selected(self) -> None: + selection = self.recovery_tree.selection() + if not selection: + messagebox.showinfo("Recovery", "Select a recovery item first.", parent=self.root) + return + result = self._run("Recovery", lambda: self.recovery.recover(int(selection[0]))) + if result: + messagebox.showinfo("Recovery", result["action"], parent=self.root) + self._refresh_recovery() + + def _build_compatibility(self, frame: ttk.Frame) -> None: + frame.columnconfigure(1, weight=1) + self.compat_dashboard = tk.StringVar(value="aurora") + self.compat_host = tk.StringVar() + self.compat_user = tk.StringVar(value="xbox") + self.compat_password = tk.StringVar(value="xbox") + for row, (label, variable, secret) in enumerate(( + ("Host", self.compat_host, False), ("Username", self.compat_user, False), + ("Password", self.compat_password, True), + )): + ttk.Label(frame, text=label).grid(row=row, column=0, sticky="w", pady=4) + ttk.Entry(frame, textvariable=variable, show="*" if secret else "").grid( + row=row, column=1, sticky="ew", pady=4 + ) + ttk.Combobox(frame, textvariable=self.compat_dashboard, state="readonly", + values=tuple(DASHBOARD_PRESETS)).grid(row=3, column=1, sticky="w", pady=4) + ttk.Button(frame, text="Run Read-only Compatibility Probe", command=self._probe).grid( + row=4, column=1, sticky="w", pady=8 + ) + self.compat_output = self._output(frame, 5, 2) + + def _probe(self) -> None: + target = FtpTarget(self.compat_host.get(), username=self.compat_user.get(), + password=self.compat_password.get()) + self._show_output(self.compat_output, self._run( + "Dashboard probe", lambda: self.compatibility.probe( + self.compat_dashboard.get(), target))) + + def _build_accessibility(self, frame: ttk.Frame) -> None: + values = self.accessibility.get() + self.access_vars: dict[str, tk.BooleanVar] = {} + labels = {"large_text": "Large text", "high_contrast": "High contrast", + "reduced_motion": "Reduced motion", "keyboard_hints": "Keyboard hints"} + for row, key in enumerate(labels): + variable = tk.BooleanVar(value=values[key]) + self.access_vars[key] = variable + ttk.Checkbutton(frame, text=labels[key], variable=variable, + command=lambda name=key, value=variable: self.accessibility.set( + name, value.get())).grid(row=row, column=0, sticky="w", pady=6) + ttk.Label(frame, text="Accessibility preferences are applied on the next application start.").grid( + row=len(labels), column=0, sticky="w", pady=(14, 0) + ) + + def _path_row(self, frame: ttk.Frame, row: int, label: str, variable: tk.StringVar, + *, directory: bool, allow_directory: bool = False) -> None: + ttk.Label(frame, text=label).grid(row=row, column=0, sticky="w", pady=4) + ttk.Entry(frame, textvariable=variable).grid(row=row, column=1, sticky="ew", pady=4) + def choose() -> None: + if allow_directory: + selected = filedialog.askopenfilename(parent=self.root, title=label) + if not selected: + selected = filedialog.askdirectory(parent=self.root, title=label) + elif directory: + selected = filedialog.askdirectory(parent=self.root, title=label) + else: + selected = filedialog.askopenfilename(parent=self.root, title=label) + if selected: + variable.set(selected) + ttk.Button(frame, text="Browse", command=choose).grid(row=row, column=2, padx=(8, 0), pady=4) + + @staticmethod + def _output(frame: ttk.Frame, row: int, columnspan: int) -> tk.Text: + frame.rowconfigure(row, weight=1) + output = tk.Text(frame, wrap=tk.WORD, height=12, background="#070b08", + foreground="#eef4ef", insertbackground="#75d34b") + output.grid(row=row, column=0, columnspan=columnspan, sticky="nsew", pady=(8, 0)) + output.configure(state=tk.DISABLED) + return output + + @staticmethod + def _show_output(widget: tk.Text, value: Any) -> None: + if value is None: + return + widget.configure(state=tk.NORMAL) + widget.delete("1.0", tk.END) + widget.insert(tk.END, json.dumps(value, indent=2, default=str)) + widget.configure(state=tk.DISABLED) + + def _run(self, title: str, callback: Callable[[], Any]) -> Any: + try: + return callback() + except Exception as exc: + messagebox.showerror(title, str(exc), parent=self.root) + return None + + @staticmethod + def _fill_tree(tree: ttk.Treeview, rows: list[dict], values: Callable) -> None: + tree.delete(*tree.get_children()) + for index, row in enumerate(rows): + tree.insert("", tk.END, iid=str(index), values=values(row)) diff --git a/community_services.py b/community_services.py new file mode 100644 index 0000000..d3414aa --- /dev/null +++ b/community_services.py @@ -0,0 +1,770 @@ +"""Safe operational services for UnityScraper's community roadmap.""" + +from __future__ import annotations + +import ftplib +import hashlib +import json +import shutil +import sqlite3 +import tempfile +import zipfile +from contextlib import contextmanager +from dataclasses import asdict, dataclass +from datetime import datetime, timezone +from pathlib import Path, PurePosixPath +from typing import Any, Iterable + +from PIL import Image + +from app_paths import DATABASE_PATH +from backup_manager import FtpTarget, inspect_stfs, inspect_xbe +from console_sync import ConsoleSyncService +from database_migrations import ensure_application_schema +from plugins import PluginManifest + + +def utc_now() -> str: + return datetime.now(timezone.utc).isoformat() + + +DASHBOARD_PRESETS: dict[str, dict[str, Any]] = { + "aurora": { + "name": "Aurora", + "content_root": "/Hdd1/Content/0000000000000000", + "games_root": "/Hdd1/Games", + "supports_resume": True, + "artwork_layout": "aurora-assets", + }, + "freestyle": { + "name": "Freestyle Dash", + "content_root": "/Hdd1/Content/0000000000000000", + "games_root": "/Hdd1/Games", + "supports_resume": True, + "artwork_layout": "freestyle-data", + }, + "xexmenu": { + "name": "XeXMenu", + "content_root": "/Hdd1/Content/0000000000000000", + "games_root": "/Hdd1/Games", + "supports_resume": False, + "artwork_layout": "none", + }, + "stock": { + "name": "Stock Content Layout", + "content_root": "/Hdd1/Content/0000000000000000", + "games_root": "/Hdd1/Content/0000000000000000", + "supports_resume": False, + "artwork_layout": "none", + }, +} + + +@dataclass(frozen=True) +class SyncAction: + action: str + local_path: str + remote_path: str + size: int + reason: str + + +class CommunityRepository: + def __init__(self, db_path: str | Path = DATABASE_PATH) -> None: + self.db_path = Path(db_path) + self.db_path.parent.mkdir(parents=True, exist_ok=True) + with self.connect() as connection: + ensure_application_schema(connection) + + @contextmanager + def connect(self): + connection = sqlite3.connect(self.db_path) + connection.row_factory = sqlite3.Row + try: + yield connection + connection.commit() + except Exception: + connection.rollback() + raise + finally: + connection.close() + + +class ConsolePlanService(CommunityRepository): + """Build explicit PC-to-console plans without touching the console.""" + + def create_plan( + self, + local_root: str | Path, + snapshot_id: int, + dashboard_slug: str = "aurora", + ) -> dict[str, Any]: + if dashboard_slug not in DASHBOARD_PRESETS: + raise ValueError(f"Unknown dashboard preset: {dashboard_slug}") + root = Path(local_root).expanduser().resolve() + if not root.is_dir(): + raise FileNotFoundError(root) + preset = DASHBOARD_PRESETS[dashboard_slug] + remote_root = str(preset["content_root"]) + local_files = { + path.relative_to(root).as_posix(): (path, path.stat().st_size) + for path in root.rglob("*") if path.is_file() + } + with self.connect() as connection: + remote_rows = connection.execute( + """ + SELECT remote_path, size FROM console_inventory_items + WHERE snapshot_id=? AND is_directory=0 + """, + (snapshot_id,), + ).fetchall() + if not remote_rows: + raise ValueError("The selected console snapshot contains no files") + remote_files: dict[str, tuple[str, int]] = {} + for row in remote_rows: + remote_path = str(row["remote_path"]) + prefix = remote_root.rstrip("/") + "/" + relative = remote_path[len(prefix):] if remote_path.startswith(prefix) else PurePosixPath(remote_path).name + remote_files[relative] = (remote_path, int(row["size"] or 0)) + actions: list[SyncAction] = [] + for relative, (path, size) in local_files.items(): + destination = remote_root.rstrip("/") + "/" + relative + if relative not in remote_files: + actions.append(SyncAction("upload", str(path), destination, size, "Missing on console")) + elif remote_files[relative][1] != size: + actions.append(SyncAction("upload", str(path), destination, size, "Size differs")) + for relative, (remote_path, size) in remote_files.items(): + if relative not in local_files: + actions.append(SyncAction("review_remote", "", remote_path, size, "Only on console")) + summary = { + "dashboard": dashboard_slug, + "uploads": sum(item.action == "upload" for item in actions), + "remote_only": sum(item.action == "review_remote" for item in actions), + "unchanged": len(set(local_files) & set(remote_files)) + - sum(item.reason == "Size differs" for item in actions), + } + cursor = connection.execute( + """ + INSERT INTO console_sync_plans( + dashboard_slug, local_root, remote_root, snapshot_id, + created_at, summary_json + ) VALUES (?, ?, ?, ?, ?, ?) + """, + (dashboard_slug, str(root), remote_root, snapshot_id, utc_now(), + json.dumps(summary, sort_keys=True)), + ) + plan_id = int(cursor.lastrowid or 0) + connection.executemany( + """ + INSERT INTO console_sync_actions( + plan_id, action, local_path, remote_path, size, reason, selected + ) VALUES (?, ?, ?, ?, ?, ?, ?) + """, + ((plan_id, item.action, item.local_path, item.remote_path, item.size, + item.reason, int(item.action == "upload")) for item in actions), + ) + return {"plan_id": plan_id, "summary": summary, + "actions": [asdict(item) for item in actions], "preset": preset} + + def list_plans(self) -> list[dict[str, Any]]: + with self.connect() as connection: + rows = connection.execute( + "SELECT * FROM console_sync_plans ORDER BY created_at DESC" + ).fetchall() + return [dict(row) for row in rows] + + def queue_uploads(self, plan_id: int, target_id: int | None = None) -> dict[str, Any]: + """Queue selected upload actions after revalidating their local files.""" + with self.connect() as connection: + plan = connection.execute( + "SELECT * FROM console_sync_plans WHERE id=?", (plan_id,) + ).fetchone() + if plan is None: + raise KeyError(f"Unknown sync plan: {plan_id}") + root = Path(plan["local_root"]).resolve() + rows = connection.execute( + """ + SELECT * FROM console_sync_actions + WHERE plan_id=? AND action='upload' AND selected=1 + AND status IN ('preview', 'failed') + ORDER BY id + """, + (plan_id,), + ).fetchall() + sync = ConsoleSyncService(self.db_path) + queued: list[int] = [] + failed: list[dict[str, Any]] = [] + for row in rows: + local = Path(row["local_path"]).resolve() + try: + local.relative_to(root) + if not local.is_file() or local.stat().st_size != int(row["size"]): + raise ValueError("Local file changed since the preview") + job_id = sync.enqueue( + "upload", local, row["remote_path"], target_id=target_id + ) + except (OSError, ValueError) as exc: + failed.append({"action_id": row["id"], "error": str(exc)}) + with self.connect() as connection: + connection.execute( + "UPDATE console_sync_actions SET status='failed' WHERE id=?", + (row["id"],), + ) + else: + queued.append(job_id) + with self.connect() as connection: + connection.execute( + "UPDATE console_sync_actions SET status='queued' WHERE id=?", + (row["id"],), + ) + with self.connect() as connection: + connection.execute( + "UPDATE console_sync_plans SET status=? WHERE id=?", + ("queued" if queued and not failed else "needs_review", plan_id), + ) + return {"plan_id": plan_id, "queued_job_ids": queued, "failed": failed} + + +class PackageWorkspaceService(CommunityRepository): + """Create auditable, read-only workspaces for user-supplied packages.""" + + def inspect(self, package_path: str | Path) -> dict[str, Any]: + package = inspect_stfs(package_path) + result = asdict(package) + result["path"] = str(package.path) + result["sha256"] = _sha256(package.path) + result["mutation_ready"] = False + result["required_before_rebuild"] = [ + "complete file-table extraction", "block/hash tree verification", + "rehash", "signature", "post-build verification", + ] + return result + + def create_workspace(self, package_path: str | Path, destination: str | Path) -> Path: + source = Path(package_path).expanduser().resolve() + details = self.inspect(source) + target = Path(destination).expanduser().resolve() + target.mkdir(parents=True, exist_ok=True) + package_copy = target / "original" / source.name + package_copy.parent.mkdir(parents=True, exist_ok=True) + if package_copy.exists() and _sha256(package_copy) != details["sha256"]: + raise FileExistsError(package_copy) + if not package_copy.exists(): + shutil.copy2(source, package_copy) + manifest = target / "unityscraper-package-workspace.json" + manifest.write_text( + json.dumps({"schema": 1, "created_at": utc_now(), "read_only": True, + "package": details}, indent=2), encoding="utf-8" + ) + return manifest + + +class ArtworkService(CommunityRepository): + PRESETS = { + "aurora": "Assets/{titleid}/cover{extension}", + "freestyle": "Data/GameData/{titleid}/boxart{extension}", + "archive": "Artwork/{titleid}/cover{extension}", + } + + def set_preference( + self, titleid: str, source_path: str | Path, *, region: str = "", language: str = "" + ) -> dict[str, Any]: + tid = titleid.strip().upper() + if len(tid) != 8 or any(ch not in "0123456789ABCDEF" for ch in tid): + raise ValueError("TitleID must be eight hexadecimal digits") + source = Path(source_path).expanduser().resolve() + with Image.open(source) as image: + image.verify() + with Image.open(source) as image: + width, height = image.size + digest = _sha256(source) + with self.connect() as connection: + connection.execute( + """ + INSERT INTO artwork_preferences( + titleid, source_path, region, language, width, height, sha256, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(titleid) DO UPDATE SET + source_path=excluded.source_path, region=excluded.region, + language=excluded.language, width=excluded.width, + height=excluded.height, sha256=excluded.sha256, + updated_at=excluded.updated_at + """, + (tid, str(source), region, language, width, height, digest, utc_now()), + ) + return {"titleid": tid, "path": str(source), "width": width, + "height": height, "sha256": digest} + + def export(self, destination: str | Path, preset: str = "aurora") -> dict[str, Any]: + if preset not in self.PRESETS: + raise ValueError(f"Unknown artwork preset: {preset}") + target = Path(destination).expanduser().resolve() + target.mkdir(parents=True, exist_ok=True) + exported: list[dict[str, str]] = [] + skipped: list[str] = [] + with self.connect() as connection: + rows = connection.execute("SELECT * FROM artwork_preferences ORDER BY titleid").fetchall() + for row in rows: + source = Path(row["source_path"]) + if not source.is_file() or _sha256(source) != row["sha256"]: + skipped.append(row["titleid"]) + continue + extension = source.suffix.casefold() or ".png" + relative = self.PRESETS[preset].format(titleid=row["titleid"], extension=extension) + output = target / Path(relative) + output.parent.mkdir(parents=True, exist_ok=True) + if output.exists() and _sha256(output) != row["sha256"]: + skipped.append(row["titleid"]) + continue + if not output.exists(): + shutil.copy2(source, output) + exported.append({"titleid": row["titleid"], "path": relative, + "sha256": row["sha256"]}) + manifest = target / "unityscraper-artwork-manifest.json" + manifest.write_text(json.dumps({"schema": 1, "preset": preset, + "generated_at": utc_now(), "artwork": exported}, indent=2), + encoding="utf-8") + cursor = connection.execute( + """INSERT INTO artwork_export_runs(destination, preset, created_at, + exported_count, skipped_count, manifest_path, status) + VALUES (?, ?, ?, ?, ?, ?, 'completed')""", + (str(target), preset, utc_now(), len(exported), len(skipped), str(manifest)), + ) + return {"run_id": int(cursor.lastrowid or 0), "exported": len(exported), + "skipped": skipped, "manifest": str(manifest)} + + +class PreservationPlanningService(CommunityRepository): + def audit_disc_sets(self, snapshot_id: int | None = None) -> list[dict[str, Any]]: + with self.connect() as connection: + selected = snapshot_id + if selected is None: + row = connection.execute( + "SELECT id FROM collection_snapshots WHERE status='completed' ORDER BY id DESC LIMIT 1" + ).fetchone() + if row is None: + return [] + selected = int(row["id"]) + rows = connection.execute( + """ + SELECT titleid, media_id, MAX(COALESCE(disc_count, 1)) expected_count, + GROUP_CONCAT(DISTINCT COALESCE(disc_number, 1)) present + FROM collection_items WHERE snapshot_id=? AND titleid IS NOT NULL + GROUP BY titleid, media_id + """, + (selected,), + ).fetchall() + audits = [] + for row in rows: + expected = max(1, int(row["expected_count"] or 1)) + present = sorted({int(value) for value in str(row["present"] or "1").split(",")}) + missing = sorted(set(range(1, expected + 1)) - set(present)) + status = "complete" if not missing else "incomplete" + connection.execute( + """ + INSERT INTO disc_set_audits(snapshot_id, titleid, media_id, + expected_count, present_json, missing_json, status, audited_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + (selected, row["titleid"], row["media_id"], expected, + json.dumps(present), json.dumps(missing), status, utc_now()), + ) + audits.append({"titleid": row["titleid"], "media_id": row["media_id"], + "expected": expected, "present": present, "missing": missing, + "status": status}) + return audits + + def create_dedup_plan(self, root: str | Path, max_files: int = 200_000) -> dict[str, Any]: + directory = Path(root).expanduser().resolve() + if not directory.is_dir(): + raise FileNotFoundError(directory) + by_size: dict[int, list[Path]] = {} + count = 0 + for path in directory.rglob("*"): + if path.is_file() and not path.is_symlink(): + count += 1 + if count > max_files: + raise ValueError(f"Dedup scan exceeds the {max_files}-file safety limit") + by_size.setdefault(path.stat().st_size, []).append(path) + groups: list[tuple[str, list[Path], int]] = [] + for size, paths in by_size.items(): + if len(paths) < 2: + continue + digests: dict[str, list[Path]] = {} + for path in paths: + digests.setdefault(_sha256(path), []).append(path) + groups.extend((digest, matches, size) for digest, matches in digests.items() + if len(matches) > 1) + reclaimable = sum(size * (len(paths) - 1) for _, paths, size in groups) + with self.connect() as connection: + cursor = connection.execute( + """INSERT INTO dedup_plans(root, created_at, duplicate_groups, + reclaimable_bytes) VALUES (?, ?, ?, ?)""", + (str(directory), utc_now(), len(groups), reclaimable), + ) + plan_id = int(cursor.lastrowid or 0) + for digest, paths, size in groups: + keeper = min(paths, key=lambda item: (len(str(item)), str(item).casefold())) + connection.executemany( + """INSERT INTO dedup_actions(plan_id, sha256, keeper_path, + duplicate_path, size) VALUES (?, ?, ?, ?, ?)""", + ((plan_id, digest, str(keeper), str(path), size) + for path in paths if path != keeper), + ) + return {"plan_id": plan_id, "groups": len(groups), + "reclaimable_bytes": reclaimable, "files_scanned": count} + + def apply_dedup_action(self, action_id: int, mode: str = "quarantine") -> dict[str, Any]: + """Apply one revalidated duplicate action with a recoverable quarantine copy.""" + if mode not in {"quarantine", "hardlink"}: + raise ValueError("mode must be quarantine or hardlink") + with self.connect() as connection: + row = connection.execute( + """ + SELECT a.*, p.root FROM dedup_actions a + JOIN dedup_plans p ON p.id=a.plan_id WHERE a.id=? + """, + (action_id,), + ).fetchone() + if row is None: + raise KeyError(action_id) + if row["status"] not in {"preview", "failed"}: + raise ValueError("This duplicate action has already been handled") + root = Path(row["root"]).resolve() + keeper = Path(row["keeper_path"]).resolve() + duplicate = Path(row["duplicate_path"]).resolve() + keeper.relative_to(root) + duplicate.relative_to(root) + if not keeper.is_file() or not duplicate.is_file(): + raise FileNotFoundError("One of the duplicate files is missing") + expected = row["sha256"] + if _sha256(keeper) != expected or _sha256(duplicate) != expected: + raise ValueError("A file changed since the duplicate preview") + quarantine_root = root / ".unityscraper-dedup-quarantine" / str(row["plan_id"]) + relative = duplicate.relative_to(root) + quarantined = quarantine_root / relative + quarantined.parent.mkdir(parents=True, exist_ok=True) + if quarantined.exists(): + raise FileExistsError(quarantined) + duplicate.replace(quarantined) + try: + if mode == "hardlink": + duplicate.hardlink_to(keeper) + except Exception: + quarantined.replace(duplicate) + raise + with self.connect() as connection: + connection.execute( + "UPDATE dedup_actions SET action=?, status='completed' WHERE id=?", + (mode, action_id), + ) + return {"action_id": action_id, "mode": mode, "keeper": str(keeper), + "duplicate": str(duplicate), "quarantine": str(quarantined)} + + +class StorageAndXboxService(CommunityRepository): + FATX_OFFSETS = (0, 0x80000, 0x130EB0000, 0x20000000) + + def audit_storage(self, source_path: str | Path) -> dict[str, Any]: + source = Path(source_path).expanduser().resolve() + if not source.exists(): + raise FileNotFoundError(source) + filesystem = "mounted-filesystem" if source.is_dir() else "unknown-image" + details: dict[str, Any] = {"size": source.stat().st_size} + if source.is_file(): + with source.open("rb") as handle: + for offset in self.FATX_OFFSETS: + if offset + 4 > source.stat().st_size: + continue + handle.seek(offset) + if handle.read(4) == b"XTAF": + filesystem = "FATX" + details["signature_offset"] = offset + break + status = "recognized" if filesystem != "unknown-image" else "unrecognized" + with self.connect() as connection: + cursor = connection.execute( + """INSERT INTO storage_source_audits(source_path, source_kind, + filesystem, detected_at, status, details_json) + VALUES (?, ?, ?, ?, ?, ?)""", + (str(source), "directory" if source.is_dir() else "image", filesystem, + utc_now(), status, json.dumps(details, sort_keys=True)), + ) + return {"audit_id": int(cursor.lastrowid or 0), "path": str(source), + "filesystem": filesystem, "access_mode": "read-only", "status": status, + "details": details} + + def scan_original_xbox(self, root: str | Path) -> list[dict[str, Any]]: + directory = Path(root).expanduser().resolve() + if not directory.is_dir(): + raise FileNotFoundError(directory) + records = [] + with self.connect() as connection: + for path in directory.rglob("default.xbe"): + try: + package = inspect_xbe(path) + except (OSError, ValueError): + continue + item = {"titleid": package.title_id, "title_name": package.title_name, + "xbe_path": str(package.path), "size": package.size} + connection.execute( + """ + INSERT INTO original_xbox_records( + titleid, title_name, xbe_path, metadata_json, scanned_at + ) VALUES (?, ?, ?, ?, ?) + ON CONFLICT(xbe_path) DO UPDATE SET titleid=excluded.titleid, + title_name=excluded.title_name, metadata_json=excluded.metadata_json, + scanned_at=excluded.scanned_at + """, + (package.title_id, package.title_name, str(package.path), + json.dumps(item, sort_keys=True), utc_now()), + ) + records.append(item) + return records + + +class PluginControlService(CommunityRepository): + def install_package(self, archive_path: str | Path, plugin_root: str | Path) -> dict[str, Any]: + archive = Path(archive_path).expanduser().resolve() + root = Path(plugin_root).expanduser().resolve() + root.mkdir(parents=True, exist_ok=True) + with zipfile.ZipFile(archive) as package: + members = [item for item in package.infolist() if not item.is_dir()] + if len(members) > 500 or sum(item.file_size for item in members) > 64 * 1024 * 1024: + raise ValueError("Plugin archive exceeds the safety limit") + for item in members: + parts = Path(item.filename).parts + if item.filename.startswith(("/", "\\")) or ".." in parts: + raise ValueError(f"Unsafe plugin archive path: {item.filename}") + with tempfile.TemporaryDirectory(dir=root) as temporary: + staging = Path(temporary) + package.extractall(staging) + manifests = list(staging.rglob("plugin.json")) + if len(manifests) != 1: + raise ValueError("Plugin archive must contain exactly one plugin.json") + manifest = PluginManifest.load(manifests[0]) + source_dir = manifests[0].parent + entry = source_dir / manifest.entrypoint + if not entry.is_file(): + raise ValueError("Plugin entrypoint is missing") + destination = root / manifest.plugin_id + backup = None + if destination.exists(): + backup = root / f".{manifest.plugin_id}.backup-{datetime.now().strftime('%Y%m%d%H%M%S')}" + destination.replace(backup) + try: + shutil.copytree(source_dir, destination) + except Exception: + if backup and backup.exists() and not destination.exists(): + backup.replace(destination) + raise + self.set_state(manifest.plugin_id, False, destination / manifest.entrypoint, + manifest.permissions) + return {"id": manifest.plugin_id, "name": manifest.name, + "version": manifest.version, "path": str(destination), + "enabled": False, "backup": str(backup) if backup else ""} + + def discover(self, plugin_root: str | Path) -> list[dict[str, Any]]: + root = Path(plugin_root).expanduser().resolve() + result = [] + if not root.is_dir(): + return result + with self.connect() as connection: + states = {row["plugin_id"]: dict(row) for row in + connection.execute("SELECT * FROM plugin_states").fetchall()} + for manifest_path in root.glob("*/plugin.json"): + try: + manifest = PluginManifest.load(manifest_path) + entry = manifest_path.parent / manifest.entrypoint + digest = _sha256(entry) if entry.is_file() else "" + state = states.get(manifest.plugin_id, {}) + result.append({"id": manifest.plugin_id, "name": manifest.name, + "version": manifest.version, + "permissions": list(manifest.permissions), + "entry_sha256": digest, + "enabled": bool(state.get("enabled", 0)), + "trusted": bool(digest and digest == state.get("trusted_sha256"))}) + except (OSError, ValueError, KeyError, json.JSONDecodeError) as exc: + result.append({"id": manifest_path.parent.name, "error": str(exc), + "enabled": False, "trusted": False}) + return result + + def set_state(self, plugin_id: str, enabled: bool, entry_path: str | Path, + permissions: Iterable[str] = ()) -> None: + entry = Path(entry_path).expanduser().resolve() + if enabled and not entry.is_file(): + raise FileNotFoundError(entry) + digest = _sha256(entry) if entry.is_file() else "" + with self.connect() as connection: + connection.execute( + """INSERT INTO plugin_states(plugin_id, enabled, trusted_sha256, + permissions_json, updated_at) VALUES (?, ?, ?, ?, ?) + ON CONFLICT(plugin_id) DO UPDATE SET enabled=excluded.enabled, + trusted_sha256=excluded.trusted_sha256, + permissions_json=excluded.permissions_json, + updated_at=excluded.updated_at""", + (plugin_id, int(enabled), digest, json.dumps(list(permissions)), utc_now()), + ) + + +class RecoveryService(CommunityRepository): + def scan(self, roots: Iterable[str | Path]) -> list[dict[str, Any]]: + events: list[dict[str, Any]] = [] + with self.connect() as connection: + for root in roots: + directory = Path(root).expanduser().resolve() + if not directory.is_dir(): + continue + for path in directory.rglob("*.partial"): + events.append(self._record(connection, "partial_file", path, True, + {"size": path.stat().st_size})) + queries = ( + ("failed_transfer", "SELECT id, local_path source, error_message details FROM console_transfer_jobs WHERE status='failed'"), + ("incomplete_snapshot", "SELECT id, snapshot_path source, status details FROM save_snapshots WHERE status<>'complete'"), + ("failed_operation", "SELECT id, source, error_message details FROM backup_operations WHERE status='failed'"), + ) + for event_type, sql in queries: + for row in connection.execute(sql).fetchall(): + events.append(self._record(connection, event_type, row["source"], True, + {"record_id": row["id"], "details": row["details"]})) + return events + + def list_open(self) -> list[dict[str, Any]]: + with self.connect() as connection: + rows = connection.execute( + "SELECT * FROM recovery_events WHERE status='open' ORDER BY detected_at DESC" + ).fetchall() + result = [] + for row in rows: + item = dict(row) + item["details"] = json.loads(item.pop("details_json")) + result.append(item) + return result + + def recover(self, event_id: int) -> dict[str, Any]: + with self.connect() as connection: + row = connection.execute( + "SELECT * FROM recovery_events WHERE id=? AND status='open'", (event_id,) + ).fetchone() + if row is None: + raise KeyError(event_id) + details = json.loads(row["details_json"]) + event_type = row["event_type"] + if event_type == "failed_transfer": + connection.execute( + """UPDATE console_transfer_jobs SET status='queued', error_message=NULL, + updated_at=? WHERE id=? AND status='failed'""", + (utc_now(), int(details["record_id"])), + ) + action = "Transfer returned to the queue" + elif event_type == "partial_file": + source = Path(row["source"]) + if not source.is_file(): + action = "Partial file was already removed" + else: + quarantine = source.parent / ".unityscraper-recovery" + quarantine.mkdir(exist_ok=True) + target = quarantine / source.name + if target.exists(): + target = quarantine / f"{source.stem}-{event_id}{source.suffix}" + source.replace(target) + action = f"Partial file quarantined at {target}" + else: + action = "Event acknowledged; inspect its source before retrying" + connection.execute( + "UPDATE recovery_events SET status='resolved', resolved_at=? WHERE id=?", + (utc_now(), event_id), + ) + return {"event_id": event_id, "action": action} + + @staticmethod + def _record(connection, event_type: str, source: str | Path, recoverable: bool, + details: dict[str, Any]) -> dict[str, Any]: + existing = connection.execute( + """SELECT id FROM recovery_events WHERE event_type=? AND source=? + AND status='open' LIMIT 1""", (event_type, str(source)) + ).fetchone() + if existing: + event_id = int(existing["id"]) + else: + cursor = connection.execute( + """INSERT INTO recovery_events(event_type, source, status, recoverable, + details_json, detected_at) VALUES (?, ?, 'open', ?, ?, ?)""", + (event_type, str(source), int(recoverable), json.dumps(details), utc_now()), + ) + event_id = int(cursor.lastrowid or 0) + return {"id": event_id, "event_type": event_type, "source": str(source), + "recoverable": recoverable, "details": details} + + +class DashboardCompatibilityService(CommunityRepository): + def probe(self, dashboard_slug: str, target: FtpTarget) -> dict[str, Any]: + if dashboard_slug not in DASHBOARD_PRESETS: + raise ValueError(f"Unknown dashboard preset: {dashboard_slug}") + results: dict[str, tuple[bool, str]] = {} + with ftplib.FTP() as ftp: + ftp.connect(target.host, target.port, timeout=target.timeout) + ftp.login(target.username, target.password) + results["connect"] = (True, ftp.getwelcome() or "Connected") + try: + features = ftp.sendcmd("FEAT") + except ftplib.all_errors as exc: + features = "" + results["feat"] = (False, str(exc)) + else: + results["feat"] = (True, features) + upper = features.upper() + results["resume"] = ("REST STREAM" in upper or "REST" in upper, + "Advertised by FEAT" if "REST" in upper else "Not advertised") + results["remote_hash"] = ("SHA-256" in upper or "XSHA256" in upper, + "Advertised by FEAT" if "HASH" in upper or "SHA" in upper else "Not advertised") + try: + ftp.cwd(str(DASHBOARD_PRESETS[dashboard_slug]["content_root"])) + except ftplib.all_errors as exc: + results["content_root"] = (False, str(exc)) + else: + results["content_root"] = (True, "Readable") + with self.connect() as connection: + for feature, (supported, details) in results.items(): + connection.execute( + """INSERT INTO dashboard_compatibility_results( + dashboard_slug, host_label, tested_at, feature, supported, details) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(dashboard_slug, host_label, feature) DO UPDATE SET + tested_at=excluded.tested_at, supported=excluded.supported, + details=excluded.details""", + (dashboard_slug, target.host, utc_now(), feature, int(supported), details), + ) + return {key: {"supported": value[0], "details": value[1]} + for key, value in results.items()} + + +class AccessibilityService(CommunityRepository): + DEFAULTS = {"large_text": "0", "high_contrast": "0", "reduced_motion": "0", + "keyboard_hints": "1"} + + def get(self) -> dict[str, bool]: + values = dict(self.DEFAULTS) + with self.connect() as connection: + values.update({row["key"]: row["value"] for row in + connection.execute("SELECT key, value FROM accessibility_preferences")}) + return {key: value == "1" for key, value in values.items()} + + def set(self, key: str, enabled: bool) -> None: + if key not in self.DEFAULTS: + raise ValueError(f"Unknown accessibility preference: {key}") + with self.connect() as connection: + connection.execute( + """INSERT INTO accessibility_preferences(key, value, updated_at) + VALUES (?, ?, ?) ON CONFLICT(key) DO UPDATE SET + value=excluded.value, updated_at=excluded.updated_at""", + (key, "1" if enabled else "0", utc_now()), + ) + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest().upper() diff --git a/database_migrations.py b/database_migrations.py index 947a439..667d129 100644 --- a/database_migrations.py +++ b/database_migrations.py @@ -8,7 +8,7 @@ from pathlib import Path -SCHEMA_VERSION = 7 +SCHEMA_VERSION = 8 def _now() -> str: @@ -68,6 +68,7 @@ def ensure_application_schema(connection: sqlite3.Connection) -> int: (5, "XboxUnity title catalog", _migration_xboxunity_catalog), (6, "profile and save management", _migration_profiles_and_saves), (7, "profile intelligence and knowledge controls", _migration_roadmap), + (8, "community roadmap workspaces", _migration_community_roadmap), ) for version, name, migration in migrations: if version in applied: @@ -499,3 +500,205 @@ def _migration_roadmap(connection: sqlite3.Connection) -> None: ADD COLUMN verify_remote_hash INTEGER NOT NULL DEFAULT 0 """ ) + + +def _migration_community_roadmap(connection: sqlite3.Connection) -> None: + """Add durable records for the community-facing roadmap workspaces.""" + connection.executescript( + """ + CREATE TABLE IF NOT EXISTS structured_knowledge_records ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + document_id INTEGER NOT NULL, + source_id INTEGER NOT NULL, + record_type TEXT NOT NULL, + canonical_name TEXT NOT NULL, + normalized_name TEXT NOT NULL, + properties_json TEXT NOT NULL, + confidence REAL NOT NULL DEFAULT 0.75, + extracted_at TEXT NOT NULL, + UNIQUE(document_id, record_type, normalized_name), + FOREIGN KEY(document_id) REFERENCES source_documents(id), + FOREIGN KEY(source_id) REFERENCES knowledge_sources(id) + ); + CREATE INDEX IF NOT EXISTS idx_structured_knowledge_lookup + ON structured_knowledge_records(record_type, normalized_name); + + CREATE TABLE IF NOT EXISTS console_sync_plans ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + dashboard_slug TEXT NOT NULL, + local_root TEXT NOT NULL, + remote_root TEXT NOT NULL, + snapshot_id INTEGER, + created_at TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'preview', + summary_json TEXT NOT NULL, + FOREIGN KEY(snapshot_id) REFERENCES console_inventory_snapshots(id) + ); + CREATE TABLE IF NOT EXISTS console_sync_actions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + plan_id INTEGER NOT NULL, + action TEXT NOT NULL, + local_path TEXT, + remote_path TEXT, + size INTEGER NOT NULL DEFAULT 0, + reason TEXT NOT NULL, + selected INTEGER NOT NULL DEFAULT 1, + status TEXT NOT NULL DEFAULT 'preview', + FOREIGN KEY(plan_id) REFERENCES console_sync_plans(id) + ); + + CREATE TABLE IF NOT EXISTS profile_migration_previews ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + profile_id TEXT NOT NULL, + source_path TEXT NOT NULL, + target_profile_id TEXT, + target_device_id TEXT, + target_console_id TEXT, + created_at TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'preview', + warnings_json TEXT NOT NULL, + changes_json TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS profile_gpd_titles ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + gpd_file_id INTEGER NOT NULL, + entry_id INTEGER NOT NULL, + titleid TEXT NOT NULL, + title TEXT, + achievements_earned INTEGER NOT NULL DEFAULT 0, + achievements_possible INTEGER NOT NULL DEFAULT 0, + gamerscore_earned INTEGER NOT NULL DEFAULT 0, + gamerscore_possible INTEGER NOT NULL DEFAULT 0, + last_played_at TEXT, + UNIQUE(gpd_file_id, entry_id), + FOREIGN KEY(gpd_file_id) REFERENCES profile_gpd_files(id) + ); + CREATE TABLE IF NOT EXISTS profile_gpd_images ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + gpd_file_id INTEGER NOT NULL, + entry_id INTEGER NOT NULL, + image_format TEXT NOT NULL, + size INTEGER NOT NULL, + sha256 TEXT NOT NULL, + UNIQUE(gpd_file_id, entry_id), + FOREIGN KEY(gpd_file_id) REFERENCES profile_gpd_files(id) + ); + CREATE TABLE IF NOT EXISTS save_comparison_runs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + left_path TEXT NOT NULL, + right_path TEXT NOT NULL, + created_at TEXT NOT NULL, + identical INTEGER NOT NULL, + summary_json TEXT NOT NULL + ); + + CREATE TABLE IF NOT EXISTS artwork_preferences ( + titleid TEXT PRIMARY KEY, + source_path TEXT NOT NULL, + artwork_type TEXT NOT NULL DEFAULT 'cover', + region TEXT, + language TEXT, + width INTEGER, + height INTEGER, + sha256 TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS artwork_export_runs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + destination TEXT NOT NULL, + preset TEXT NOT NULL, + created_at TEXT NOT NULL, + exported_count INTEGER NOT NULL DEFAULT 0, + skipped_count INTEGER NOT NULL DEFAULT 0, + manifest_path TEXT, + status TEXT NOT NULL + ); + + CREATE TABLE IF NOT EXISTS disc_set_audits ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + snapshot_id INTEGER, + titleid TEXT NOT NULL, + media_id TEXT, + expected_count INTEGER NOT NULL, + present_json TEXT NOT NULL, + missing_json TEXT NOT NULL, + status TEXT NOT NULL, + audited_at TEXT NOT NULL, + FOREIGN KEY(snapshot_id) REFERENCES collection_snapshots(id) + ); + CREATE TABLE IF NOT EXISTS dedup_plans ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + root TEXT NOT NULL, + created_at TEXT NOT NULL, + duplicate_groups INTEGER NOT NULL DEFAULT 0, + reclaimable_bytes INTEGER NOT NULL DEFAULT 0, + status TEXT NOT NULL DEFAULT 'preview' + ); + CREATE TABLE IF NOT EXISTS dedup_actions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + plan_id INTEGER NOT NULL, + sha256 TEXT NOT NULL, + keeper_path TEXT NOT NULL, + duplicate_path TEXT NOT NULL, + size INTEGER NOT NULL, + action TEXT NOT NULL DEFAULT 'review', + status TEXT NOT NULL DEFAULT 'preview', + FOREIGN KEY(plan_id) REFERENCES dedup_plans(id) + ); + + CREATE TABLE IF NOT EXISTS storage_source_audits ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + source_path TEXT NOT NULL, + source_kind TEXT NOT NULL, + filesystem TEXT, + access_mode TEXT NOT NULL DEFAULT 'read-only', + detected_at TEXT NOT NULL, + status TEXT NOT NULL, + details_json TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS original_xbox_records ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + titleid TEXT, + title_name TEXT NOT NULL, + xbe_path TEXT NOT NULL UNIQUE, + region_flags TEXT, + version TEXT, + compatibility TEXT, + metadata_json TEXT NOT NULL, + scanned_at TEXT NOT NULL + ); + + CREATE TABLE IF NOT EXISTS plugin_states ( + plugin_id TEXT PRIMARY KEY, + enabled INTEGER NOT NULL DEFAULT 0, + trusted_sha256 TEXT, + permissions_json TEXT NOT NULL DEFAULT '[]', + updated_at TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS recovery_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + event_type TEXT NOT NULL, + source TEXT NOT NULL, + status TEXT NOT NULL, + recoverable INTEGER NOT NULL DEFAULT 1, + details_json TEXT NOT NULL, + detected_at TEXT NOT NULL, + resolved_at TEXT + ); + CREATE TABLE IF NOT EXISTS dashboard_compatibility_results ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + dashboard_slug TEXT NOT NULL, + host_label TEXT, + tested_at TEXT NOT NULL, + feature TEXT NOT NULL, + supported INTEGER NOT NULL, + details TEXT, + UNIQUE(dashboard_slug, host_label, feature) + ); + CREATE TABLE IF NOT EXISTS accessibility_preferences ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + """ + ) diff --git a/desktop_app.py b/desktop_app.py index dc80777..131d43c 100644 --- a/desktop_app.py +++ b/desktop_app.py @@ -3,8 +3,18 @@ from __future__ import annotations import sys +import traceback -from app_paths import ensure_app_dirs, ensure_user_titleids_file +from app_paths import LOG_DIR, ensure_app_dirs, ensure_user_titleids_file + + +def _write_startup_report(exc: BaseException) -> str: + details = "".join(traceback.format_exception(type(exc), exc, exc.__traceback__)) + try: + (LOG_DIR / "fatal_startup.log").write_text(details, encoding="utf-8") + except OSError: + pass + return details def main() -> int: @@ -16,23 +26,39 @@ def main() -> int: import tkinter as tk from modern_gui import main as gui_main except ImportError as exc: + details = _write_startup_report(exc) print( "UnityScraper requires Tkinter. On Debian/Ubuntu install python3-tk; " "on Fedora install python3-tkinter.", file=sys.stderr, ) - print(f"Details: {exc}", file=sys.stderr) + print(details, file=sys.stderr) return 1 try: gui_main() except tk.TclError as exc: + details = _write_startup_report(exc) print( "UnityScraper could not connect to a graphical desktop. " "Start it from an X11 or Wayland session.", file=sys.stderr, ) - print(f"Details: {exc}", file=sys.stderr) + print(details, file=sys.stderr) + return 1 + except Exception as exc: + report = LOG_DIR / "fatal_startup.log" + details = _write_startup_report(exc) + try: + from tkinter import messagebox + + messagebox.showerror( + "UnityScraper could not start", + f"A startup report was written to:\n{report}\n\n{exc}", + ) + except tk.TclError: + pass + print(details, file=sys.stderr) return 1 return 0 diff --git a/gpd_parser.py b/gpd_parser.py index f08176a..c117fef 100644 --- a/gpd_parser.py +++ b/gpd_parser.py @@ -82,6 +82,26 @@ class GpdSetting: value: Any +@dataclass(frozen=True) +class GpdTitleHistory: + entry_id: int + title_id: str + title: str + achievements_earned: int + achievements_possible: int + gamerscore_earned: int + gamerscore_possible: int + last_played_at: str + + +@dataclass(frozen=True) +class GpdImage: + entry_id: int + image_format: str + size: int + sha256: str + + @dataclass(frozen=True) class GpdReport: path: Path @@ -92,6 +112,8 @@ class GpdReport: entry_count: int achievements: tuple[GpdAchievement, ...] settings: tuple[GpdSetting, ...] + titles: tuple[GpdTitleHistory, ...] + images: tuple[GpdImage, ...] namespace_counts: dict[str, int] warnings: tuple[str, ...] @@ -120,6 +142,55 @@ def parse_gpd(path: str | Path, title_id: str = "") -> GpdReport: return parse_gpd_bytes(data, path=source, title_id=title_id) +def export_gpd_image( + path: str | Path, entry_id: int, destination: str | Path +) -> Path: + """Export one validated embedded image without changing its GPD.""" + source = Path(path).expanduser().resolve() + data = source.read_bytes() + if len(data) > MAX_FILE_SIZE: + raise GpdError(f"GPD exceeds the {MAX_FILE_SIZE}-byte safety limit") + if len(data) < XDBF_HEADER.size: + raise GpdError("GPD is smaller than the XDBF header") + magic, _version, entry_max, entry_count, free_max, _free_count = XDBF_HEADER.unpack_from(data) + if magic != XDBF_MAGIC or entry_max > MAX_TABLE_ENTRIES or entry_count > entry_max: + raise GpdError("GPD has an invalid XDBF header") + header_size = XDBF_HEADER.size + entry_max * XDBF_ENTRY.size + free_max * XDBF_FREE_ENTRY_SIZE + for index in range(entry_count): + namespace, current_id, offset, size = XDBF_ENTRY.unpack_from( + data, XDBF_HEADER.size + index * XDBF_ENTRY.size + ) + if namespace != 2 or current_id != entry_id: + continue + absolute = header_size + offset + if offset < 0 or size < 0 or absolute > len(data) or size > len(data) - absolute: + raise GpdError("Image entry extends beyond the end of the GPD") + payload = data[absolute:absolute + size] + image_format = _image_format(payload) + if not image_format or size > 16 * 1024 * 1024: + raise GpdError("Embedded image is unsupported or exceeds the safety limit") + target = Path(destination).expanduser().resolve() + expected_suffix = ".jpg" if image_format == "jpeg" else f".{image_format}" + if target.suffix.casefold() != expected_suffix: + target = target.with_suffix(expected_suffix) + target.parent.mkdir(parents=True, exist_ok=True) + temporary = target.with_suffix(target.suffix + ".partial") + temporary.write_bytes(payload) + try: + from PIL import Image + with Image.open(temporary) as image: + image.verify() + except Exception as exc: + temporary.unlink(missing_ok=True) + raise GpdError(f"Embedded image failed validation: {exc}") from exc + if target.exists() and hashlib.sha256(target.read_bytes()).digest() != hashlib.sha256(payload).digest(): + temporary.unlink(missing_ok=True) + raise FileExistsError(target) + temporary.replace(target) + return target + raise KeyError(f"GPD image entry {entry_id} was not found") + + def parse_gpd_bytes( data: bytes, *, @@ -174,6 +245,8 @@ def parse_gpd_bytes( achievements: list[GpdAchievement] = [] settings: list[GpdSetting] = [] + titles: list[GpdTitleHistory] = [] + images: list[GpdImage] = [] warnings: list[str] = [] for entry in entries: payload = data[ @@ -189,6 +262,24 @@ def parse_gpd_bytes( settings.append(_parse_setting(entry, payload)) except GpdError as exc: warnings.append(f"Setting {entry.entry_id}: {exc}") + elif entry.namespace == 4 and entry.entry_id not in {-1, -2}: + try: + titles.append(_parse_title_history(entry, payload)) + except GpdError as exc: + warnings.append(f"Title {entry.entry_id}: {exc}") + elif entry.namespace == 2 and entry.entry_id not in {-1, -2}: + image_format = _image_format(payload) + if image_format: + images.append( + GpdImage( + entry.entry_id, + image_format, + len(payload), + hashlib.sha256(payload).hexdigest().upper(), + ) + ) + else: + warnings.append(f"Image {entry.entry_id}: unsupported or malformed image") source = Path(path) inferred = source.stem.upper() if TITLE_ID_RE.fullmatch(source.stem) else "" @@ -204,6 +295,8 @@ def parse_gpd_bytes( entry_count, tuple(sorted(achievements, key=lambda item: item.achievement_id)), tuple(sorted(settings, key=lambda item: item.setting_id)), + tuple(sorted(titles, key=lambda item: item.last_played_at, reverse=True)), + tuple(sorted(images, key=lambda item: item.entry_id)), namespace_counts, tuple(warnings), ) @@ -266,6 +359,37 @@ def _parse_setting(entry: XdbfEntry, payload: bytes) -> GpdSetting: return GpdSetting(entry.entry_id, setting_id, type_name, value) +def _parse_title_history(entry: XdbfEntry, payload: bytes) -> GpdTitleHistory: + if len(payload) < 0x28: + raise GpdError("record is smaller than the title-history header") + title_id, possible_count, earned_count, possible_score, earned_score = ( + struct.unpack_from(">IIIII", payload, 0) + ) + filetime = struct.unpack_from(">q", payload, 0x20)[0] + title_bytes = payload[0x28:] + title = title_bytes.decode("utf-16-be", errors="replace").split("\0", 1)[0] + return GpdTitleHistory( + entry.entry_id, + f"{title_id:08X}", + title, + earned_count, + possible_count, + earned_score, + possible_score, + _filetime_iso(filetime), + ) + + +def _image_format(payload: bytes) -> str: + if payload.startswith(b"\x89PNG\r\n\x1a\n"): + return "png" + if payload.startswith(b"\xff\xd8\xff"): + return "jpeg" + if payload.startswith(b"BM"): + return "bmp" + return "" + + def _split_utf16be_strings(data: bytes, limit: int) -> list[str]: result: list[str] = [] current = bytearray() diff --git a/i18n.py b/i18n.py index b257bd6..100a289 100644 --- a/i18n.py +++ b/i18n.py @@ -240,6 +240,26 @@ } +def _repair_legacy_text(value: str) -> str: + """Repair translations that were historically saved with the wrong encoding.""" + if not any(marker in value for marker in ("Ã", "Â", "â", "ã", "ç", "é")): + return value + for encoding in ("cp1252", "latin-1"): + try: + repaired = value.encode(encoding).decode("utf-8") + except (UnicodeEncodeError, UnicodeDecodeError): + continue + if repaired != value: + return repaired + return value + + +for _language, _strings in TRANSLATIONS.items(): + TRANSLATIONS[_language] = { + key: _repair_legacy_text(value) for key, value in _strings.items() + } + + class Translator: """Language translator for GUI""" diff --git a/modern_gui.py b/modern_gui.py index da51916..ab89eaa 100644 --- a/modern_gui.py +++ b/modern_gui.py @@ -36,6 +36,8 @@ from backup_gui import BackupPage from collection_gui import CollectionPage from collection_intelligence import CollectionIntelligenceService +from community_gui import CommunityHubPage +from community_services import AccessibilityService from backup_service import BackupService from database import DatabaseManager from database_migrations import create_database_backup, restore_database_backup @@ -333,6 +335,8 @@ def __init__(self, root: tk.Tk) -> None: self.collections = CollectionIntelligenceService() self.database = DatabaseManager() self.title_catalog = XboxUnityTitleCatalog() + self.accessibility = AccessibilityService() + self.accessibility_preferences = self.accessibility.get() self.current_game: str | None = None self._catalog_syncing = False self._catalog_events: queue.Queue[tuple[str, Any]] = queue.Queue() @@ -343,6 +347,8 @@ def __init__(self, root: tk.Tk) -> None: self.root.title(f"UnityScraper {APP_VERSION}") config = self._read_config() scale = max(0.8, min(2.0, float(config.get("ui_scale", 1.0)))) + if self.accessibility_preferences["large_text"]: + scale = min(2.0, scale + 0.25) self.root.tk.call("tk", "scaling", scale) self.root.geometry("1220x780") self.root.minsize(980, 640) @@ -434,11 +440,38 @@ def _configure_style(self) -> None: foreground=[("selected", ACCENT), ("active", TEXT)]) style.configure("Vertical.TScrollbar", background="#111a13", troughcolor="#050806", arrowcolor=ACCENT, bordercolor=BORDER) + + if self.accessibility_preferences["high_contrast"]: + contrast_bg = "#000000" + contrast_text = "#ffffff" + contrast_accent = "#b6ff00" + self.root.configure(background=contrast_bg) + style.configure(".", background=contrast_bg, foreground=contrast_text, + fieldbackground=contrast_bg, bordercolor=contrast_text, + troughcolor=contrast_bg, selectbackground=contrast_accent, + selectforeground=contrast_bg) + for name in ("TFrame", "Content.TFrame", "TLabel", "TLabelframe", + "TLabelframe.Label", "TNotebook"): + style.configure(name, background=contrast_bg, foreground=contrast_text) + style.configure("Sidebar.TFrame", background=contrast_bg) + style.configure("Brand.TLabel", background=contrast_bg, foreground=contrast_text) + style.configure("AccentBrand.TLabel", background=contrast_bg, + foreground=contrast_accent) + style.configure("TButton", background=contrast_bg, foreground=contrast_text, + bordercolor=contrast_text) + style.configure("Nav.TButton", background=contrast_bg, foreground=contrast_text, + bordercolor=contrast_text) + style.configure("Treeview", background=contrast_bg, fieldbackground=contrast_bg, + foreground=contrast_text, bordercolor=contrast_text) + style.configure("Treeview.Heading", background=contrast_bg, + foreground=contrast_accent, bordercolor=contrast_text) + style.configure("TNotebook.Tab", background=contrast_bg, foreground=contrast_text) + def _build_shell(self) -> None: self._wallpaper_source: Image.Image | None = None self._wallpaper_photo: ImageTk.PhotoImage | None = None wallpaper = resource_path("assets", "backgrounds", "unityscraper_full_background.png") - if wallpaper.exists(): + if wallpaper.exists() and not self.accessibility_preferences["high_contrast"]: try: self._wallpaper_source = Image.open(wallpaper).convert("RGB") except OSError: @@ -467,12 +500,17 @@ def _build_shell(self) -> None: ("EXTERNAL TOOLS", self.show_external_tools), ("COLLECTIONS", self.show_collections), ("KNOWLEDGE", self.show_knowledge), + ("COMMUNITY HUB", self.show_community_hub), ("ARCHIVE HEALTH", self.show_health), ("SETTINGS", self.show_settings), ("HELP & ABOUT", self.show_about), ) - for label, callback in pages: - ttk.Button(nav, text=label, command=callback, style="Nav.TButton", width=22).pack( + for index, (label, callback) in enumerate(pages, start=1): + shortcut = navigation_shortcut(index) + display_label = label + if shortcut and self.accessibility_preferences["keyboard_hints"]: + display_label = f"{label} Alt+{shortcut}" + ttk.Button(nav, text=display_label, command=callback, style="Nav.TButton", width=22).pack( fill=tk.X, pady=3 ) @@ -495,6 +533,7 @@ def _build_shell(self) -> None: lambda _event, action=callback: action(), ) self.root.bind("", lambda _event: self.show_library()) + self.root.bind("", lambda _event: self.show_community_hub(focus_search=True)) self.show_library() def _resize_shell(self, _event: tk.Event[Any] | None) -> None: @@ -1058,6 +1097,14 @@ def show_knowledge(self) -> None: self.knowledge, ) + def show_community_hub(self, focus_search: bool = False) -> None: + self._clear_content() + self.community_hub_page = CommunityHubPage( + self.root, self.content, self._page_header + ) + if focus_search: + self.root.after_idle(self.community_hub_page.focus_search) + def show_health(self) -> None: self._clear_content() self._page_header( diff --git a/packaging/linux/io.github.trapemall.UnityScraper.metainfo.xml b/packaging/linux/io.github.trapemall.UnityScraper.metainfo.xml index a3e8484..5ef4994 100644 --- a/packaging/linux/io.github.trapemall.UnityScraper.metainfo.xml +++ b/packaging/linux/io.github.trapemall.UnityScraper.metainfo.xml @@ -22,6 +22,6 @@ https://github.com/TrapEmAll/UnityScraper/issues - + diff --git a/packaging/macos/Info.plist b/packaging/macos/Info.plist new file mode 100644 index 0000000..e08c366 --- /dev/null +++ b/packaging/macos/Info.plist @@ -0,0 +1,16 @@ + + + + + CFBundleDisplayNameUnityScraper + CFBundleExecutableUnityScraper + CFBundleIdentifierio.github.trapemall.UnityScraper + CFBundleInfoDictionaryVersion6.0 + CFBundleNameUnityScraper + CFBundlePackageTypeAPPL + CFBundleShortVersionString1.1.0 + CFBundleVersion2 + LSMinimumSystemVersion11.0 + NSHighResolutionCapable + + diff --git a/profile_gui.py b/profile_gui.py index 611854c..5233f19 100644 --- a/profile_gui.py +++ b/profile_gui.py @@ -10,6 +10,7 @@ from tkinter import filedialog, messagebox, simpledialog, ttk from typing import Any, Callable +from gpd_parser import export_gpd_image from platform_support import open_path from profile_intelligence import ProfileIntelligenceService from profile_manager import ProfileSaveManager, mask_identifier @@ -96,16 +97,19 @@ def _build(self) -> None: inventory = ttk.Frame(notebook, padding=10) snapshots = ttk.Frame(notebook, padding=10) achievements = ttk.Frame(notebook, padding=10) + history = ttk.Frame(notebook, padding=10) compare = ttk.Frame(notebook, padding=10) xenia = ttk.Frame(notebook, padding=10) notebook.add(inventory, text="Inventory") notebook.add(snapshots, text="Snapshots") notebook.add(achievements, text="Achievements") + notebook.add(history, text="Played Titles") notebook.add(compare, text="Compare") notebook.add(xenia, text="Xenia") self._build_inventory(inventory) self._build_snapshots(snapshots) self._build_achievements(achievements) + self._build_history(history) self._build_compare(compare) self._build_xenia(xenia) @@ -381,6 +385,73 @@ def _build_compare(self, parent: ttk.Frame) -> None: ) self.compare_text.configure(state=tk.DISABLED) + def _build_history(self, parent: ttk.Frame) -> None: + parent.columnconfigure(0, weight=1) + parent.rowconfigure(1, weight=1) + parent.rowconfigure(3, weight=1) + self.history_summary_var = tk.StringVar( + value="Import an extracted dashboard GPD to view played-title history." + ) + ttk.Label(parent, textvariable=self.history_summary_var, + style="Subheader.TLabel").grid(row=0, column=0, sticky="ew", pady=(0, 8)) + self.history_tree = ttk.Treeview( + parent, + columns=("game", "titleid", "achievements", "score", "last_played"), + show="headings", + ) + for column, label, width in ( + ("game", "Game", 300), ("titleid", "TitleID", 90), + ("achievements", "Achievements", 110), ("score", "Gamerscore", 110), + ("last_played", "Last Played", 190), + ): + self.history_tree.heading(column, text=label) + self.history_tree.column(column, width=width, anchor=tk.W) + self.history_tree.grid(row=1, column=0, sticky="nsew") + image_toolbar = ttk.Frame(parent) + image_toolbar.grid(row=2, column=0, sticky="ew", pady=(10, 6)) + ttk.Label(image_toolbar, text="Embedded artwork", style="CardTitle.TLabel").pack( + side=tk.LEFT + ) + ttk.Button( + image_toolbar, text="Export Selected Image", command=self.export_embedded_image + ).pack(side=tk.RIGHT) + self.gpd_image_tree = ttk.Treeview( + parent, columns=("format", "size", "source"), show="headings", height=5 + ) + for column, label, width in ( + ("format", "Format", 90), ("size", "Size", 90), ("source", "Source GPD", 620) + ): + self.gpd_image_tree.heading(column, text=label) + self.gpd_image_tree.column(column, width=width, anchor=tk.W) + self.gpd_image_tree.grid(row=3, column=0, sticky="nsew") + self.gpd_images: dict[str, dict[str, Any]] = {} + + def export_embedded_image(self) -> None: + selection = self.gpd_image_tree.selection() + if not selection: + messagebox.showinfo( + "Embedded artwork", "Select an image first.", parent=self.root + ) + return + row = self.gpd_images[selection[0]] + suffix = ".jpg" if row["image_format"] == "jpeg" else f".{row['image_format']}" + destination = filedialog.asksaveasfilename( + parent=self.root, + title="Export embedded artwork", + defaultextension=suffix, + initialfile=f"gpd-image-{row['entry_id']}{suffix}", + ) + if not destination: + return + try: + exported = export_gpd_image(row["source_path"], int(row["entry_id"]), destination) + except Exception as exc: + messagebox.showerror("Image export failed", str(exc), parent=self.root) + return + messagebox.showinfo( + "Embedded artwork", f"Exported to {exported}", parent=self.root + ) + def _build_xenia(self, parent: ttk.Frame) -> None: parent.columnconfigure(1, weight=1) parent.rowconfigure(3, weight=1) @@ -598,6 +669,7 @@ def refresh(self) -> None: self._refresh_profiles() self._refresh_snapshots() self._refresh_gpd_files() + self._refresh_history() def _refresh_profiles(self) -> None: selected = self._selected_profile_id() @@ -680,6 +752,41 @@ def _refresh_gpd_files(self) -> None: elif self.gpd_tree.get_children(): self.gpd_tree.selection_set(self.gpd_tree.get_children()[0]) self._refresh_achievements() + self._refresh_history() + + def _refresh_history(self) -> None: + if not hasattr(self, "history_tree"): + return + self.history_tree.delete(*self.history_tree.get_children()) + rows = self.intelligence.list_title_history(self._selected_profile_id()) + for row in rows: + self.history_tree.insert( + "", tk.END, + values=( + row.get("title") or "Unknown game", row["titleid"], + f"{row['achievements_earned']} / {row['achievements_possible']}", + f"{row['gamerscore_earned']} / {row['gamerscore_possible']}", + row.get("last_played_at") or "Unknown", + ), + ) + self.gpd_image_tree.delete(*self.gpd_image_tree.get_children()) + self.gpd_images.clear() + image_count = 0 + for gpd in self.gpd_files.values(): + for image in self.intelligence.list_images(int(gpd["id"])): + image_count += 1 + item_id = f"gpd-image-{gpd['id']}-{image['entry_id']}" + image["source_path"] = gpd["source_path"] + self.gpd_images[item_id] = image + self.gpd_image_tree.insert( + "", tk.END, iid=item_id, + values=(image["image_format"].upper(), _size(image["size"]), + gpd["source_path"]), + ) + self.history_summary_var.set( + f"{len(rows)} played titles and {image_count} validated embedded images. " + "Source GPD files remain unchanged." + ) def _refresh_achievements(self) -> None: if not hasattr(self, "achievement_tree"): diff --git a/profile_intelligence.py b/profile_intelligence.py index bb9fb32..b05a174 100644 --- a/profile_intelligence.py +++ b/profile_intelligence.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import hashlib import sqlite3 from contextlib import contextmanager from dataclasses import asdict @@ -120,6 +121,8 @@ def import_gpd( connection.execute( "DELETE FROM profile_achievements WHERE gpd_file_id=?", (gpd_id,) ) + connection.execute("DELETE FROM profile_gpd_titles WHERE gpd_file_id=?", (gpd_id,)) + connection.execute("DELETE FROM profile_gpd_images WHERE gpd_file_id=?", (gpd_id,)) connection.executemany( """ INSERT INTO profile_achievements( @@ -144,6 +147,35 @@ def import_gpd( for item in report.achievements ), ) + connection.executemany( + """ + INSERT INTO profile_gpd_titles( + gpd_file_id, entry_id, titleid, title, achievements_earned, + achievements_possible, gamerscore_earned, gamerscore_possible, + last_played_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + ( + gpd_id, item.entry_id, item.title_id, item.title, + item.achievements_earned, item.achievements_possible, + item.gamerscore_earned, item.gamerscore_possible, + item.last_played_at, + ) + for item in report.titles + ), + ) + connection.executemany( + """ + INSERT INTO profile_gpd_images( + gpd_file_id, entry_id, image_format, size, sha256 + ) VALUES (?, ?, ?, ?, ?) + """, + ( + (gpd_id, item.entry_id, item.image_format, item.size, item.sha256) + for item in report.images + ), + ) return gpd_id def scan_gpd_directory( @@ -212,6 +244,160 @@ def list_achievements( ).fetchall() return [dict(row) for row in rows] + def list_title_history(self, profile_id: str = "") -> list[dict[str, Any]]: + values: tuple[Any, ...] = () + where = "" + if profile_id: + where = "WHERE g.profile_id=?" + values = (profile_id.strip().upper(),) + with self._connect() as connection: + rows = connection.execute( + f""" + SELECT t.*, g.profile_id, g.source_path + FROM profile_gpd_titles t + JOIN profile_gpd_files g ON g.id=t.gpd_file_id + {where} + ORDER BY t.last_played_at DESC, t.title + """, + values, + ).fetchall() + return [dict(row) for row in rows] + + def list_images(self, gpd_file_id: int) -> list[dict[str, Any]]: + with self._connect() as connection: + rows = connection.execute( + "SELECT * FROM profile_gpd_images WHERE gpd_file_id=? ORDER BY entry_id", + (gpd_file_id,), + ).fetchall() + return [dict(row) for row in rows] + + def profile_dashboard(self, profile_id: str) -> dict[str, Any]: + profile = profile_id.strip().upper() + if not PROFILE_ID_RE.fullmatch(profile): + raise ProfileIntelligenceError("Profile ID must be 16 hexadecimal digits") + with self._connect() as connection: + row = connection.execute( + """ + SELECT + (SELECT COUNT(*) FROM profile_saves WHERE profile_id=?) save_count, + (SELECT COUNT(DISTINCT titleid) FROM profile_saves + WHERE profile_id=?) save_titles, + (SELECT COALESCE(SUM(size), 0) FROM profile_saves + WHERE profile_id=?) save_bytes, + (SELECT COUNT(*) FROM profile_gpd_files WHERE profile_id=?) gpd_count, + (SELECT COALESCE(SUM(unlocked_count), 0) FROM profile_gpd_files + WHERE profile_id=?) unlocked_count, + (SELECT COALESCE(SUM(achievement_count), 0) FROM profile_gpd_files + WHERE profile_id=?) achievement_count, + (SELECT COALESCE(SUM(gamerscore_earned), 0) FROM profile_gpd_files + WHERE profile_id=?) gamerscore_earned, + (SELECT COALESCE(SUM(gamerscore_possible), 0) FROM profile_gpd_files + WHERE profile_id=?) gamerscore_possible + """, + (profile,) * 8, + ).fetchone() + recent = connection.execute( + """ + SELECT titleid, title, last_played_at, gamerscore_earned, + gamerscore_possible FROM profile_gpd_titles t + JOIN profile_gpd_files g ON g.id=t.gpd_file_id + WHERE g.profile_id=? ORDER BY last_played_at DESC LIMIT 10 + """, + (profile,), + ).fetchall() + result = dict(row or {}) + possible = int(result.get("gamerscore_possible") or 0) + earned = int(result.get("gamerscore_earned") or 0) + result.update( + { + "profile_id": profile, + "completion_percent": round((earned / possible) * 100, 1) if possible else 0.0, + "recent_titles": [dict(item) for item in recent], + } + ) + return result + + def preview_ownership_migration( + self, + profile_id: str, + source_path: str | Path, + *, + target_profile_id: str = "", + target_device_id: str = "", + target_console_id: str = "", + ) -> dict[str, Any]: + profile = profile_id.strip().upper() + target_profile = target_profile_id.strip().upper() + if not PROFILE_ID_RE.fullmatch(profile): + raise ProfileIntelligenceError("Source profile ID must be 16 hexadecimal digits") + if target_profile and not PROFILE_ID_RE.fullmatch(target_profile): + raise ProfileIntelligenceError("Target profile ID must be 16 hexadecimal digits") + source = Path(source_path).expanduser().resolve() + if not source.is_file(): + raise FileNotFoundError(source) + for label, value in (("Device ID", target_device_id), ("Console ID", target_console_id)): + if value and (len(value) % 2 or any(ch not in "0123456789abcdefABCDEF" for ch in value)): + raise ProfileIntelligenceError(f"{label} must contain complete hexadecimal bytes") + changes = { + "profile_id": {"from": profile, "to": target_profile or profile}, + "device_id": {"to": target_device_id.upper()}, + "console_id": {"to": target_console_id.upper()}, + } + warnings = [ + "Preview only: no package bytes were changed.", + "A signed CON package requires complete STFS rehashing and signing after ownership changes.", + "Create and verify a snapshot before using any external resigning tool.", + ] + with self._connect() as connection: + cursor = connection.execute( + """ + INSERT INTO profile_migration_previews( + profile_id, source_path, target_profile_id, target_device_id, + target_console_id, created_at, warnings_json, changes_json + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + profile, str(source), target_profile, target_device_id.upper(), + target_console_id.upper(), utc_now(), json.dumps(warnings), + json.dumps(changes, sort_keys=True), + ), + ) + preview_id = int(cursor.lastrowid or 0) + return {"preview_id": preview_id, "source_path": str(source), + "changes": changes, "warnings": warnings} + + def compare_save_files(self, left_path: str | Path, right_path: str | Path) -> dict[str, Any]: + left = Path(left_path).expanduser().resolve() + right = Path(right_path).expanduser().resolve() + if not left.is_file() or not right.is_file(): + raise FileNotFoundError("Both comparison paths must be files") + left_hash, left_blocks = _hash_blocks(left) + right_hash, right_blocks = _hash_blocks(right) + changed = [ + index for index in range(max(len(left_blocks), len(right_blocks))) + if (left_blocks[index] if index < len(left_blocks) else None) + != (right_blocks[index] if index < len(right_blocks) else None) + ] + summary = { + "left": str(left), "right": str(right), + "left_size": left.stat().st_size, "right_size": right.stat().st_size, + "left_sha256": left_hash, "right_sha256": right_hash, + "identical": left_hash == right_hash, + "block_size": 64 * 1024, "changed_blocks": changed, + } + with self._connect() as connection: + cursor = connection.execute( + """ + INSERT INTO save_comparison_runs( + left_path, right_path, created_at, identical, summary_json + ) VALUES (?, ?, ?, ?, ?) + """, + (str(left), str(right), utc_now(), int(summary["identical"]), + json.dumps(summary, sort_keys=True)), + ) + summary["comparison_id"] = int(cursor.lastrowid or 0) + return summary + def compare_profiles(self, left_profile_id: str, right_profile_id: str) -> dict[str, Any]: left = left_profile_id.strip().upper() right = right_profile_id.strip().upper() @@ -403,3 +589,13 @@ def _comparison_summary( "achievements_only_right": sorted(unlocked[right] - unlocked[left]), "achievements_shared": len(unlocked[left] & unlocked[right]), } + + +def _hash_blocks(path: Path, block_size: int = 64 * 1024) -> tuple[str, list[str]]: + digest = hashlib.sha256() + blocks: list[str] = [] + with path.open("rb") as handle: + for block in iter(lambda: handle.read(block_size), b""): + digest.update(block) + blocks.append(hashlib.sha256(block).hexdigest()) + return digest.hexdigest(), blocks diff --git a/pyproject.toml b/pyproject.toml index 3292f74..d83f111 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "unityscraper" -version = "1.0.0b1" +version = "1.1.0b1" description = "Xbox 360 library, knowledge, preservation, and backup manager" readme = "README.md" requires-python = ">=3.10" diff --git a/structured_knowledge.py b/structured_knowledge.py new file mode 100644 index 0000000..51c2d01 --- /dev/null +++ b/structured_knowledge.py @@ -0,0 +1,179 @@ +"""Conservative structured extraction from locally cached reference articles.""" + +from __future__ import annotations + +import html +import json +import re +import sqlite3 +from contextlib import contextmanager +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from app_paths import DATABASE_PATH +from database_migrations import ensure_application_schema + + +TAG_RE = re.compile(r"<[^>]+>") +ROW_RE = re.compile( + r"]*>\s*<(?:th|td)[^>]*>(.*?)\s*" + r"]*>(.*?)\s*", + re.IGNORECASE | re.DOTALL, +) +PAIR_RE = re.compile(r"^([A-Za-z][A-Za-z0-9 /_+#().-]{1,48})\s*[:=-]\s*(.{1,500})$") + +RECORD_TYPES = ( + ("motherboard", ("motherboard", "xenon", "zephyr", "falcon", "jasper", "trinity", "corona")), + ("dvd_drive", ("dvd drive", "lite-on", "hitachi", "benq", "samsung")), + ("dashboard", ("dashboard", "kernel", "system update")), + ("exploit", ("jtag", "rgh", "reset glitch", "exploit")), + ("error_code", ("error code", "secondary error", "red ring")), + ("file_format", ("file format", "stfs", "xex", "xbe", "fatx", "xcontent")), + ("repair", ("repair", "reflow", "replace", "solder")), + ("tool", ("tool", "utility", "homebrew", "application")), +) + + +def utc_now() -> str: + return datetime.now(timezone.utc).isoformat() + + +class StructuredKnowledgeService: + """Extract useful fields while preserving the original source document.""" + + def __init__(self, db_path: str | Path = DATABASE_PATH) -> None: + self.db_path = Path(db_path) + with self._connect() as connection: + ensure_application_schema(connection) + + @contextmanager + def _connect(self): + connection = sqlite3.connect(self.db_path) + connection.row_factory = sqlite3.Row + try: + yield connection + connection.commit() + finally: + connection.close() + + def extract_cached_documents(self, limit: int = 0) -> dict[str, Any]: + sql = """ + SELECT d.id, d.source_id, d.title, d.url, d.cache_path, d.metadata, + s.name source_name + FROM source_documents d JOIN knowledge_sources s ON s.id=d.source_id + WHERE d.cache_path IS NOT NULL AND d.cache_path <> '' + ORDER BY d.id + """ + values: tuple[Any, ...] = () + if limit > 0: + sql += " LIMIT ?" + values = (limit,) + extracted = 0 + skipped = 0 + errors: list[str] = [] + with self._connect() as connection: + rows = connection.execute(sql, values).fetchall() + for row in rows: + try: + path = Path(row["cache_path"]) + if not path.is_file() or path.stat().st_size > 16 * 1024 * 1024: + skipped += 1 + continue + raw = path.read_text(encoding="utf-8", errors="replace") + title = (row["title"] or Path(row["url"]).name).strip() + properties = extract_properties(raw) + properties.update( + {"source": row["source_name"], "source_url": row["url"]} + ) + record_type = infer_record_type(title, raw[:20_000]) + connection.execute( + """ + INSERT INTO structured_knowledge_records( + document_id, source_id, record_type, canonical_name, + normalized_name, properties_json, confidence, extracted_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(document_id, record_type, normalized_name) DO UPDATE SET + canonical_name=excluded.canonical_name, + properties_json=excluded.properties_json, + confidence=excluded.confidence, + extracted_at=excluded.extracted_at + """, + ( + row["id"], row["source_id"], record_type, title, + normalize(title), json.dumps(properties, sort_keys=True), + 0.80 if properties else 0.60, utc_now(), + ), + ) + extracted += 1 + except (OSError, UnicodeError, sqlite3.Error, ValueError) as exc: + errors.append(f"{row['url']}: {exc}") + return {"extracted": extracted, "skipped": skipped, "errors": errors} + + def list_records(self, record_type: str = "", query: str = "") -> list[dict]: + clauses: list[str] = [] + values: list[Any] = [] + if record_type: + clauses.append("r.record_type=?") + values.append(record_type) + if query.strip(): + clauses.append("(r.canonical_name LIKE ? OR r.properties_json LIKE ?)") + term = f"%{query.strip()}%" + values.extend((term, term)) + where = " WHERE " + " AND ".join(clauses) if clauses else "" + with self._connect() as connection: + rows = connection.execute( + """ + SELECT r.*, d.url source_url, s.name source_name + FROM structured_knowledge_records r + JOIN source_documents d ON d.id=r.document_id + JOIN knowledge_sources s ON s.id=r.source_id + """ + where + " ORDER BY r.record_type, r.canonical_name LIMIT 2000", + values, + ).fetchall() + result = [] + for row in rows: + item = dict(row) + item["properties"] = json.loads(item.pop("properties_json")) + result.append(item) + return result + + +def extract_properties(raw: str) -> dict[str, str]: + properties: dict[str, str] = {} + for key, value in ROW_RE.findall(raw): + _add_property(properties, clean_markup(key), clean_markup(value)) + plain = clean_markup(raw) + for line in plain.splitlines(): + match = PAIR_RE.match(" ".join(line.split())) + if match: + _add_property(properties, match.group(1), match.group(2)) + if len(properties) >= 40: + break + return properties + + +def _add_property(properties: dict[str, str], key: str, value: str) -> None: + normalized = normalize(key).replace(" ", "_") + cleaned = " ".join(value.split()) + if normalized and cleaned and normalized not in properties and len(cleaned) <= 500: + properties[normalized] = cleaned + + +def clean_markup(value: str) -> str: + value = re.sub(r"<(?:br|/p|/li|/tr|/h\d)>\s*", "\n", value, flags=re.I) + return html.unescape(TAG_RE.sub(" ", value)).replace("\r", "") + + +def infer_record_type(title: str, body: str) -> str: + haystack = f"{title}\n{clean_markup(body)}".casefold() + scores = [ + (sum(haystack.count(keyword) for keyword in keywords), kind) + for kind, keywords in RECORD_TYPES + ] + score, kind = max(scores) + return kind if score else "reference_article" + + +def normalize(value: str) -> str: + return " ".join(value.casefold().split()) diff --git a/tests.py b/tests.py index 4bfbc22..64883d3 100644 --- a/tests.py +++ b/tests.py @@ -64,6 +64,12 @@ class TestPlatformSupport(unittest.TestCase): """Test cross-platform storage and desktop integration.""" + def test_legacy_translations_are_repaired_at_load_time(self): + from i18n import TRANSLATIONS + + self.assertEqual(TRANSLATIONS["es"]["settings"], "Configuraci\u00f3n") + self.assertEqual(TRANSLATIONS["ja"]["browse"], "\u53c2\u7167") + def test_linux_uses_xdg_directories(self): home = Path("/home/tester") paths = resolve_storage_paths( @@ -83,6 +89,17 @@ def test_linux_uses_xdg_directories(self): self.assertEqual(paths.cache, Path("/xdg/cache/unityscraper")) self.assertEqual(paths.logs, Path("/xdg/state/unityscraper/logs")) + def test_macos_uses_native_user_directories(self): + home = Path("/Users/tester") + paths = resolve_storage_paths( + os_name="posix", platform_name="darwin", environ={}, home=home + ) + self.assertEqual( + paths.base, home / "Library" / "Application Support" / "UnityScraper" + ) + self.assertEqual(paths.cache, home / "Library" / "Caches" / "UnityScraper") + self.assertEqual(paths.logs, home / "Library" / "Logs" / "UnityScraper") + def test_linux_xdg_defaults_follow_home(self): home = Path("/home/tester") paths = resolve_storage_paths( @@ -1631,7 +1648,7 @@ 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, 5, 6, 7]) + self.assertEqual([row[0] for row in versions], [1, 2, 3, 4, 5, 6, 7, 8]) self.assertIn("collection_snapshots", tables) self.assertIn("preservation_matches", tables) self.assertIn("console_transfer_jobs", tables) @@ -1737,6 +1754,217 @@ def test_updater_selects_platform_asset_and_ignores_checksum(self): self.assertEqual(selected["name"], "UnityScraper-Windows-x64.zip") +class TestCommunityRoadmap(unittest.TestCase): + def setUp(self): + self.temp_dir = Path(tempfile.mkdtemp()) + self.db_path = self.temp_dir / "community.db" + self.database = DatabaseManager(str(self.db_path)) + + def tearDown(self): + shutil.rmtree(self.temp_dir) + + def test_unified_search_spans_games_profiles_and_achievements(self): + import sqlite3 + from contextlib import closing + from unified_search import UnifiedSearchService + + self.database.add_titleid("53510804", "Hitman: Absolution", "Square Enix") + with closing(sqlite3.connect(self.db_path)) as connection: + connection.execute( + """INSERT INTO xbox_profiles(profile_id, gamertag, source_path, + profile_kind, package_status, first_seen_at, last_seen_at) + VALUES ('E000000000000001', 'Agent47', 'profile', 'retail', + 'header-valid', 'now', 'now')""" + ) + connection.commit() + service = UnifiedSearchService(self.db_path) + self.assertEqual(service.search("Hitman")[0]["identifier"], "53510804") + self.assertEqual(service.search("Agent47")[0]["category"], "profile") + + def test_structured_knowledge_extracts_cached_hardware_article(self): + import sqlite3 + from contextlib import closing + from structured_knowledge import StructuredKnowledgeService + + cache = self.temp_dir / "jasper.html" + cache.write_text( + "

Jasper Motherboard

" + "" + "
CPU65 nm
NAND16 MB
", + encoding="utf-8", + ) + with closing(sqlite3.connect(self.db_path)) as connection: + source = connection.execute( + """INSERT INTO knowledge_sources(slug, name) + VALUES ('test-source', 'Test Source')""" + ) + source_id = source.lastrowid + connection.execute( + """INSERT INTO source_documents(source_id, url, title, cache_path) + VALUES (?, 'https://example.test/jasper', 'Jasper Motherboard', ?)""", + (source_id, str(cache)), + ) + connection.commit() + service = StructuredKnowledgeService(self.db_path) + self.assertEqual(service.extract_cached_documents()["extracted"], 1) + record = service.list_records()[0] + self.assertEqual(record["record_type"], "motherboard") + self.assertEqual(record["properties"]["cpu"], "65 nm") + + def test_gpd_title_history_and_safe_image_metadata(self): + import struct + from gpd_parser import parse_gpd_bytes + + title = bytearray(0x28) + struct.pack_into(">IIIII", title, 0, 0x53510804, 50, 25, 1000, 500) + title.extend("Hitman: Absolution".encode("utf-16-be") + b"\0\0") + image = b"\x89PNG\r\n\x1a\n" + b"image payload" + header = struct.pack(">4sIIIII", b"XDBF", 1, 2, 2, 0, 0) + entries = ( + struct.pack(">Hqii", 4, 0x53510804, 0, len(title)) + + struct.pack(">Hqii", 2, 42, len(title), len(image)) + ) + report = parse_gpd_bytes(header + entries + title + image) + self.assertEqual(report.titles[0].title_id, "53510804") + self.assertEqual(report.titles[0].gamerscore_earned, 500) + self.assertEqual(report.images[0].image_format, "png") + + def test_console_sync_plan_can_queue_revalidated_uploads(self): + import sqlite3 + from contextlib import closing + from community_services import ConsolePlanService + + local = self.temp_dir / "content" + game = local / "53510804" / "00000001" / "save.bin" + game.parent.mkdir(parents=True) + game.write_bytes(b"new save") + with closing(sqlite3.connect(self.db_path)) as connection: + snapshot = connection.execute( + """INSERT INTO console_inventory_snapshots(root, captured_at, status) + VALUES ('/Hdd1/Content/0000000000000000', 'now', 'completed')""" + ) + connection.execute( + """INSERT INTO console_inventory_items(snapshot_id, remote_path, size) + VALUES (?, '/Hdd1/Content/0000000000000000/old.bin', 3)""", + (snapshot.lastrowid,), + ) + connection.commit() + snapshot_id = snapshot.lastrowid + service = ConsolePlanService(self.db_path) + plan = service.create_plan(local, snapshot_id) + self.assertEqual(plan["summary"]["uploads"], 1) + self.assertTrue(any(item["action"] == "review_remote" for item in plan["actions"])) + queued = service.queue_uploads(plan["plan_id"]) + self.assertEqual(len(queued["queued_job_ids"]), 1) + with closing(sqlite3.connect(self.db_path)) as connection: + status = connection.execute( + "SELECT status FROM console_transfer_jobs WHERE id=?", + (queued["queued_job_ids"][0],), + ).fetchone()[0] + self.assertEqual(status, "queued") + + def test_artwork_disc_dedup_and_storage_plans(self): + import sqlite3 + from PIL import Image + from community_services import ArtworkService, PreservationPlanningService, StorageAndXboxService + + artwork = self.temp_dir / "cover.png" + Image.new("RGB", (64, 96), "green").save(artwork) + art = ArtworkService(self.db_path) + art.set_preference("53510804", artwork) + exported = art.export(self.temp_dir / "art-export") + self.assertEqual(exported["exported"], 1) + + duplicates = self.temp_dir / "duplicates" + duplicates.mkdir() + (duplicates / "a.bin").write_bytes(b"same") + (duplicates / "b.bin").write_bytes(b"same") + plan = PreservationPlanningService(self.db_path).create_dedup_plan(duplicates) + self.assertEqual(plan["groups"], 1) + self.assertEqual((duplicates / "a.bin").read_bytes(), b"same") + from contextlib import closing + with closing(sqlite3.connect(self.db_path)) as connection: + action_id = connection.execute( + "SELECT id FROM dedup_actions WHERE plan_id=?", (plan["plan_id"],) + ).fetchone()[0] + applied = PreservationPlanningService(self.db_path).apply_dedup_action(action_id) + self.assertTrue(Path(applied["quarantine"]).is_file()) + self.assertFalse(Path(applied["duplicate"]).exists()) + + fatx = self.temp_dir / "drive.img" + fatx.write_bytes(b"XTAF" + bytes(64)) + audit = StorageAndXboxService(self.db_path).audit_storage(fatx) + self.assertEqual(audit["filesystem"], "FATX") + self.assertEqual(audit["access_mode"], "read-only") + + def test_plugin_recovery_and_accessibility_controls(self): + from community_services import AccessibilityService, PluginControlService, RecoveryService + + plugin = self.temp_dir / "plugins" / "sample" + plugin.mkdir(parents=True) + (plugin / "plugin.py").write_text("value = 1\n", encoding="utf-8") + (plugin / "plugin.json").write_text(json.dumps({ + "id": "sample", "name": "Sample", "version": "1.0", + "api_version": 1, "entrypoint": "plugin.py", "permissions": ["metadata"], + }), encoding="utf-8") + control = PluginControlService(self.db_path) + discovered = control.discover(plugin.parent) + self.assertEqual(discovered[0]["id"], "sample") + control.set_state("sample", True, plugin / "plugin.py", ["metadata"]) + self.assertTrue(control.discover(plugin.parent)[0]["trusted"]) + + partial = self.temp_dir / "download.partial" + partial.write_bytes(b"partial") + events = RecoveryService(self.db_path).scan([self.temp_dir]) + partial_event = next(item for item in events if item["event_type"] == "partial_file") + recovered = RecoveryService(self.db_path).recover(partial_event["id"]) + self.assertIn("quarantined", recovered["action"]) + self.assertFalse(partial.exists()) + + access = AccessibilityService(self.db_path) + access.set("large_text", True) + access.set("high_contrast", True) + access.set("reduced_motion", True) + self.assertTrue(access.get()["large_text"]) + self.assertTrue(access.get()["high_contrast"]) + self.assertTrue(access.get()["reduced_motion"]) + + def test_package_workspace_is_read_only_and_profile_tools_are_audited(self): + from community_services import PackageWorkspaceService + from profile_intelligence import ProfileIntelligenceService + + package = self.temp_dir / "save.bin" + header = bytearray(0x1791) + header[:4] = b"CON " + header[0x344:0x348] = (1).to_bytes(4, "big") + header[0x354:0x358] = bytes.fromhex("12345678") + header[0x360:0x364] = bytes.fromhex("53510804") + header[0x371:0x379] = bytes.fromhex("E000000000000001") + header[0x3FD:0x411] = bytes.fromhex("11" * 20) + title = "Hitman: Absolution".encode("utf-16-be") + header[0x411:0x411 + len(title)] = title + header[0x1691:0x1691 + len(title)] = title + package.write_bytes(header + b"payload") + + service = PackageWorkspaceService(self.db_path) + details = service.inspect(package) + self.assertFalse(details["mutation_ready"]) + manifest = service.create_workspace(package, self.temp_dir / "workspace") + self.assertTrue(manifest.is_file()) + self.assertTrue(json.loads(manifest.read_text(encoding="utf-8"))["read_only"]) + + intelligence = ProfileIntelligenceService(self.db_path) + preview = intelligence.preview_ownership_migration( + "E000000000000001", package, target_profile_id="E000000000000002" + ) + self.assertGreater(preview["preview_id"], 0) + self.assertIn("Preview only", preview["warnings"][0]) + other = self.temp_dir / "other.bin" + other.write_bytes(package.read_bytes()[:-1] + b"x") + comparison = intelligence.compare_save_files(package, other) + self.assertFalse(comparison["identical"]) + + def run_tests(): """Run all tests""" # Create test suite @@ -1762,6 +1990,7 @@ def run_tests(): suite.addTests(loader.loadTestsFromTestCase(TestProfileSaveManager)) suite.addTests(loader.loadTestsFromTestCase(TestRoadmapFeatures)) suite.addTests(loader.loadTestsFromTestCase(TestUnifiedV1Foundation)) + suite.addTests(loader.loadTestsFromTestCase(TestCommunityRoadmap)) # Run tests runner = unittest.TextTestRunner(verbosity=2) diff --git a/unified_search.py b/unified_search.py new file mode 100644 index 0000000..f87150a --- /dev/null +++ b/unified_search.py @@ -0,0 +1,229 @@ +"""Fast, local search across UnityScraper's library and knowledge domains.""" + +from __future__ import annotations + +import sqlite3 +from contextlib import contextmanager +from dataclasses import dataclass, asdict +from pathlib import Path +from typing import Any, Iterable + +from app_paths import DATABASE_PATH +from database_migrations import ensure_application_schema + + +@dataclass(frozen=True) +class SearchResult: + category: str + title: str + subtitle: str + identifier: str + target: str + score: int + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + +class UnifiedSearchService: + """Query every user-facing domain without requiring a network request.""" + + def __init__(self, db_path: str | Path = DATABASE_PATH) -> None: + self.db_path = Path(db_path) + self.db_path.parent.mkdir(parents=True, exist_ok=True) + with self._connect() as connection: + ensure_application_schema(connection) + + @contextmanager + def _connect(self): + connection = sqlite3.connect(self.db_path) + connection.row_factory = sqlite3.Row + try: + yield connection + connection.commit() + finally: + connection.close() + + def search( + self, + query: str, + *, + categories: Iterable[str] = (), + limit: int = 100, + ) -> list[dict[str, Any]]: + needle = " ".join(query.strip().split()) + if len(needle) < 2: + return [] + wanted = {item.casefold() for item in categories} + term = f"%{needle}%" + results: list[SearchResult] = [] + with self._connect() as connection: + if not wanted or "game" in wanted: + results.extend(self._games(connection, needle, term)) + if not wanted or "knowledge" in wanted: + results.extend(self._knowledge(connection, needle, term)) + if not wanted or "profile" in wanted: + results.extend(self._profiles(connection, term)) + if not wanted or "save" in wanted: + results.extend(self._saves(connection, term)) + if not wanted or "achievement" in wanted: + results.extend(self._achievements(connection, term)) + if not wanted or "file" in wanted: + results.extend(self._files(connection, term)) + if not wanted or "tool" in wanted: + results.extend(self._structured(connection, term)) + unique: dict[tuple[str, str], SearchResult] = {} + for result in results: + key = (result.category, result.target) + previous = unique.get(key) + if previous is None or result.score > previous.score: + unique[key] = result + ordered = sorted( + unique.values(), key=lambda row: (-row.score, row.title.casefold(), row.category) + ) + return [item.to_dict() for item in ordered[: max(1, min(limit, 500))]] + + @staticmethod + def _rank(needle: str, title: str, identifier: str = "") -> int: + query = needle.casefold() + name = title.casefold() + key = identifier.casefold() + if query == key or query == name: + return 100 + if name.startswith(query) or key.startswith(query): + return 80 + return 50 + + def _games(self, connection, needle: str, term: str) -> list[SearchResult]: + rows = connection.execute( + """ + SELECT titleid, name, COALESCE(publisher, '') publisher + FROM titleids + WHERE titleid LIKE ? OR name LIKE ? OR publisher LIKE ? + UNION + SELECT titleid, name, '' publisher FROM xboxunity_title_catalog + WHERE titleid LIKE ? OR name LIKE ? + LIMIT 200 + """, + (term, term, term, term, term), + ).fetchall() + return [ + SearchResult( + "game", + row["name"] or row["titleid"], + " | ".join(value for value in (row["titleid"], row["publisher"]) if value), + row["titleid"], + f"game:{row['titleid']}", + self._rank(needle, row["name"] or "", row["titleid"]), + ) + for row in rows + ] + + def _knowledge(self, connection, needle: str, term: str) -> list[SearchResult]: + rows = connection.execute( + """ + SELECT DISTINCT e.id, e.entity_type, e.canonical_name, + GROUP_CONCAT(DISTINCT i.identifier_value) identifiers + FROM knowledge_entities e + LEFT JOIN entity_identifiers i ON i.entity_id=e.id + LEFT JOIN knowledge_facts f ON f.entity_id=e.id + WHERE e.canonical_name LIKE ? OR i.identifier_value LIKE ? OR f.value LIKE ? + GROUP BY e.id LIMIT 200 + """, + (term, term, term), + ).fetchall() + return [ + SearchResult( + "knowledge", + row["canonical_name"], + f"{row['entity_type']} | {row['identifiers'] or 'source-attributed record'}", + str(row["id"]), + f"knowledge:{row['id']}", + self._rank(needle, row["canonical_name"], row["identifiers"] or ""), + ) + for row in rows + ] + + @staticmethod + def _profiles(connection, term: str) -> list[SearchResult]: + rows = connection.execute( + """ + SELECT profile_id, COALESCE(gamertag, 'Unknown gamertag') gamertag, + source_path FROM xbox_profiles + WHERE profile_id LIKE ? OR gamertag LIKE ? OR source_path LIKE ? + LIMIT 100 + """, + (term, term, term), + ).fetchall() + return [ + SearchResult("profile", row["gamertag"], row["profile_id"], row["profile_id"], + f"profile:{row['profile_id']}", 60) + for row in rows + ] + + @staticmethod + def _saves(connection, term: str) -> list[SearchResult]: + rows = connection.execute( + """ + SELECT id, titleid, name, profile_id, source_path FROM profile_saves + WHERE titleid LIKE ? OR name LIKE ? OR profile_id LIKE ? OR source_path LIKE ? + LIMIT 150 + """, + (term, term, term, term), + ).fetchall() + return [ + SearchResult("save", row["name"], f"{row['titleid']} | {row['profile_id']}", + str(row["id"]), f"save:{row['id']}", 55) + for row in rows + ] + + @staticmethod + def _achievements(connection, term: str) -> list[SearchResult]: + rows = connection.execute( + """ + SELECT a.id, a.title, a.gamerscore, a.unlock_state, g.titleid + FROM profile_achievements a + JOIN profile_gpd_files g ON g.id=a.gpd_file_id + WHERE a.title LIKE ? OR a.locked_description LIKE ? + OR a.unlocked_description LIKE ? OR g.titleid LIKE ? + LIMIT 150 + """, + (term, term, term, term), + ).fetchall() + return [ + SearchResult("achievement", row["title"] or "Untitled achievement", + f"{row['titleid']} | {row['gamerscore']}G | {row['unlock_state']}", + str(row["id"]), f"achievement:{row['id']}", 50) + for row in rows + ] + + @staticmethod + def _files(connection, term: str) -> list[SearchResult]: + rows = connection.execute( + """ + SELECT id, path, size, COALESCE(sha256, '') sha256 FROM local_file_hashes + WHERE path LIKE ? OR sha256 LIKE ? ORDER BY calculated_at DESC LIMIT 100 + """, + (term, term), + ).fetchall() + return [ + SearchResult("file", Path(row["path"]).name, row["path"], str(row["id"]), + f"file:{row['path']}", 40) + for row in rows + ] + + @staticmethod + def _structured(connection, term: str) -> list[SearchResult]: + rows = connection.execute( + """ + SELECT id, record_type, canonical_name, properties_json + FROM structured_knowledge_records + WHERE canonical_name LIKE ? OR properties_json LIKE ? LIMIT 150 + """, + (term, term), + ).fetchall() + return [ + SearchResult(row["record_type"], row["canonical_name"], "Structured knowledge", + str(row["id"]), f"structured:{row['id']}", 45) + for row in rows + ]