From 042d47e3d44d8471f74a62eb8a10578ee50b9e2d Mon Sep 17 00:00:00 2001 From: Golumpa Date: Wed, 15 Jul 2026 19:06:47 +0100 Subject: [PATCH] Added mappings updater Implemented a mappings updater script that uses the Waifus-Grace/NTE_Assets repo to export all the required mappings. --- .github/workflows/update-mappings.yml | 136 ++++++ README.md | 20 + docs/mapping-updates.md | 95 +++++ mappings/arcs.json | 36 +- mappings/characters.json | 1 + mappings/items.json | 41 +- src/nte_history_exporter/mapping_update.py | 387 ++++++++++++++++++ tests/fixtures/README.md | 9 + .../Appearance/DT_AppearanceData.json | 1 + .../DataTable/Character/DT_Character.json | 1 + .../DataTable/Fork/DT_ForkItemData.json | 1 + .../DataTable/Gacha/GachaIllustrate.json | 1 + .../Inventory/DT_CapitalItemConfig.json | 1 + .../DataTable/Inventory/DT_ItemConfig.json | 1 + .../nte_assets/Localization/en/game.json | 18 + tests/test_mapping_update.py | 184 +++++++++ tools/update_mappings.py | 86 ++++ 17 files changed, 980 insertions(+), 39 deletions(-) create mode 100644 .github/workflows/update-mappings.yml create mode 100644 docs/mapping-updates.md create mode 100644 src/nte_history_exporter/mapping_update.py create mode 100644 tests/fixtures/nte_assets/DataTable/Character/Appearance/DT_AppearanceData.json create mode 100644 tests/fixtures/nte_assets/DataTable/Character/DT_Character.json create mode 100644 tests/fixtures/nte_assets/DataTable/Fork/DT_ForkItemData.json create mode 100644 tests/fixtures/nte_assets/DataTable/Gacha/GachaIllustrate.json create mode 100644 tests/fixtures/nte_assets/DataTable/Inventory/DT_CapitalItemConfig.json create mode 100644 tests/fixtures/nte_assets/DataTable/Inventory/DT_ItemConfig.json create mode 100644 tests/fixtures/nte_assets/Localization/en/game.json create mode 100644 tests/test_mapping_update.py create mode 100644 tools/update_mappings.py diff --git a/.github/workflows/update-mappings.yml b/.github/workflows/update-mappings.yml new file mode 100644 index 0000000..8760bb3 --- /dev/null +++ b/.github/workflows/update-mappings.yml @@ -0,0 +1,136 @@ +name: Update reward mappings + +on: + workflow_dispatch: + inputs: + source_ref: + description: NTE_Assets branch, tag, or commit + required: true + default: main + type: string + +permissions: + contents: write + pull-requests: write + +concurrency: + group: update-reward-mappings + cancel-in-progress: false + +jobs: + update: + name: Rebuild mappings from NTE_Assets + runs-on: ubuntu-latest + + env: + AUTOMATION_BRANCH: dev/automated-mapping-update + NTE_ASSETS_REF: ${{ inputs.source_ref }} + GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} + + steps: + - name: Checkout source code + uses: actions/checkout@v5 + with: + fetch-depth: 0 + ref: ${{ github.event.repository.default_branch }} + + - name: Get Python + uses: actions/setup-python@v6 + with: + python-version: "3.12" + + - name: Install project + run: python -m pip install -e . + + - name: Rebuild reward mappings + run: python tools/update_mappings.py --source-ref "$NTE_ASSETS_REF" --apply + + - name: Run tests + run: python -m unittest discover -s tests + + - name: Summarize mapping changes + id: changes + shell: bash + run: | + python - <<'PY' + import json + import os + from pathlib import Path + + report = json.loads(Path("build/mapping-update/mapping-update-report.json").read_text()) + changes = report["changes"] + changed = sum(changes.values()) > 0 + with Path(os.environ["GITHUB_OUTPUT"]).open("a", encoding="utf-8") as output: + output.write(f"changed={'true' if changed else 'false'}\n") + with Path(os.environ["GITHUB_STEP_SUMMARY"]).open("a", encoding="utf-8") as summary: + summary.write("## NTE_Assets mapping update\n\n") + summary.write(f"Source ref: `{report['source_ref']}`\n\n") + summary.write(f"- Additions: {changes['additions']}\n") + summary.write(f"- Updates: {changes['updates']}\n") + summary.write(f"- Deletions: {changes['deletions']}\n") + PY + + - name: Upload review report + uses: actions/upload-artifact@v6 + with: + name: mapping-update-report + path: build/mapping-update/mapping-update-report.json + + - name: Commit and push mapping changes + if: steps.changes.outputs.changed == 'true' + shell: bash + run: | + git config user.name "github-actions[bot]" + git config user.email "41898282+github-actions[bot]@users.noreply.github.com" + git checkout -B "$AUTOMATION_BRANCH" + git add mappings/arcs.json mappings/characters.json mappings/items.json + git commit -m "chore: update reward mappings from NTE_Assets" + git push --force origin "HEAD:refs/heads/$AUTOMATION_BRANCH" + + - name: Open or update pull request + if: steps.changes.outputs.changed == 'true' + shell: bash + env: + BASE_BRANCH: ${{ github.event.repository.default_branch }} + RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }} + run: | + python - <<'PY' + import json + import os + from pathlib import Path + + report = json.loads(Path("build/mapping-update/mapping-update-report.json").read_text()) + changes = report["changes"] + body = ( + "Automated authoritative reward-mapping snapshot from " + "[Waifus-Grace/NTE_Assets](https://github.com/Waifus-Grace/NTE_Assets).\n\n" + f"Source ref: `{report['source_ref']}`\n\n" + f"- {changes['additions']} additions\n" + f"- {changes['updates']} updates\n" + f"- {changes['deletions']} deletions\n\n" + "The workflow rebuilt all reward mappings, resolved English names through " + "`Localization/en/game.json`, and passed the complete test suite. Pool mapping " + "files and UID inputs were not changed.\n\n" + f"[Workflow run]({os.environ['RUN_URL']})\n" + ) + Path("build/mapping-update/pull-request-body.md").write_text(body, encoding="utf-8") + PY + + pr_number="$(gh pr list \ + --head "$AUTOMATION_BRANCH" \ + --base "$BASE_BRANCH" \ + --state open \ + --json number \ + --jq '.[0].number // empty')" + + if [ -n "$pr_number" ]; then + gh pr edit "$pr_number" \ + --title "chore: update reward mappings from NTE_Assets" \ + --body-file build/mapping-update/pull-request-body.md + else + gh pr create \ + --head "$AUTOMATION_BRANCH" \ + --base "$BASE_BRANCH" \ + --title "chore: update reward mappings from NTE_Assets" \ + --body-file build/mapping-update/pull-request-body.md + fi diff --git a/README.md b/README.md index 7a50127..2ffd974 100644 --- a/README.md +++ b/README.md @@ -163,6 +163,26 @@ The exporter automatically includes the shareable NTE user UID when it appears i > [!TIP] > For reliable deduplication, start from page 1 and scroll through the pages. If you only want pages 1–5, scroll through to page 6 as well just to be on the safe side. +## Mapping maintenance + +Reward metadata can be rebuilt directly from the latest NTE_Assets tables and +English translation files. The reward snapshot may change IDs while UID inputs +remain untouched: + +Run **Update reward mappings** from the GitHub Actions tab to generate and test +the snapshot in a reviewable pull request, or run it locally: + +```powershell +python tools/update_mappings.py +``` + +This rebuilds reviewable reward-map candidates directly from NTE_Assets under +`build/mapping-update/`; committed mappings are untouched unless `--apply` is +explicitly supplied. The snapshot may include additions, updates, and removals, +while pool mappings and UID inputs remain untouched. See +[Reward mapping updates](docs/mapping-updates.md) for the source rules and +review workflow. + ## Privacy > [!CAUTION] diff --git a/docs/mapping-updates.md b/docs/mapping-updates.md new file mode 100644 index 0000000..9978af9 --- /dev/null +++ b/docs/mapping-updates.md @@ -0,0 +1,95 @@ +# Reward mapping updates + +`tools/update_mappings.py` rebuilds `arcs.json`, `characters.json`, and +`items.json` directly from the current +[`Waifus-Grace/NTE_Assets`](https://github.com/Waifus-Grace/NTE_Assets) +tables. No other exporter or intermediate mapping format is used. + +English display names come from `Localization/en/game.json`. The data tables +provide localization namespace/key references and quality metadata; their +embedded `LocalizedString` values are deliberately ignored. + +The generated files are an authoritative snapshot, not an additive merge. +When NTE_Assets removes an ID, changes its casing, or updates its metadata, the +staged mapping reflects that change. This is intentional for a live-service +game whose asset tables evolve over time. + +## Review workflow + +### GitHub Actions + +Run **Update reward mappings** from the repository's Actions tab. The optional +`source_ref` input accepts an NTE_Assets branch, tag, or commit and defaults to +`main`. + +The workflow rebuilds the mappings, runs the complete test suite, uploads the +JSON review report, and opens a pull request from +`dev/automated-mapping-update`. Running it again updates the same open pull +request. If the authoritative snapshot has not changed, it reports zero changes +and does not create a commit or pull request. + +The repository setting **Allow GitHub Actions to create and approve pull +requests** must be enabled for automatic pull-request creation. The workflow +uses only the repository-scoped `GITHUB_TOKEN` and grants it `contents: write` +and `pull-requests: write` permissions. + +### Local + +Stage the latest candidate files without touching the committed mappings: + +```powershell +python tools/update_mappings.py +``` + +The required JSON tables are downloaded from the `main` branch by default. +Pin a tag, branch, or commit for reproducible review: + +```powershell +python tools/update_mappings.py --source-ref e9752e5963103529f0d683fd0aed4752b7dfad78 +``` + +To use an existing local checkout without network access: + +```powershell +python tools/update_mappings.py --assets-root path\to\NTE_Assets +``` + +The default output is `build/mapping-update/` and contains the three candidate +mapping files plus `mapping-update-report.json`. The report lists additions, +updates, and deletions by file and records SHA-256 hashes for every source +table. + +After reviewing the report and diff, apply the validated snapshot explicitly: + +```powershell +python tools/update_mappings.py --apply +python -m pytest -q +``` + +For CI, `--check` exits with status 1 whenever the generated snapshot differs +from the committed reward mappings. It still writes the staged artifacts. + +## Source rules + +- Every Arc in `DT_ForkItemData` is emitted to `arcs.json`. +- Every character in `DT_Character` is emitted to `characters.json`. +- Other pull rewards are selected by `GachaIllustrate` and resolved to the + inventory tables for quality and localization keys. Appearance-table IDs + provide canonical casing for glider rewards. +- Names are resolved strictly through `Localization/en/game.json`. A missing or + ambiguous key fails the update instead of falling back to a DT value. +- Character-awakening illustration entries are not independent pull rewards + and are excluded. +- Orange, purple, and blue item qualities map to `S`, `A`, and `B`. + +## UID compatibility boundary + +This updater reads and writes only the three reward mapping files. It never +reads or writes the permanent, limited, beginner, or Arc pool mapping files. +Banner IDs, timestamps, record ordering, format version, and all UID inputs are +therefore unchanged by mapping synchronization. + +Reward IDs and display metadata may change or disappear when the authoritative +NTE_Assets snapshot changes. Existing exported history files remain unchanged; +new exports describe rewards using the current asset snapshot. UID stability is +independent of reward display mappings. diff --git a/mappings/arcs.json b/mappings/arcs.json index c31ea08..1a23a05 100644 --- a/mappings/arcs.json +++ b/mappings/arcs.json @@ -1,38 +1,36 @@ { - "fork_dustbin": {"name": "Dangerous Game", "rank": "B"}, - "fork_vine": {"name": "Be Happy", "rank": "B"}, - "fork_nonos": {"name": "First Step to Success", "rank": "B"}, "fork_appliance": {"name": "\"Real Music\"", "rank": "B"}, - "fork_Prokaryon": {"name": "Us.", "rank": "B"}, - "fork_PaperPlane": {"name": "Clear Skies", "rank": "A"}, - "fork_mofeikesi": {"name": "Good Boy's Grand Adventure", "rank": "S"}, - "fork_wuhuakuang": {"name": "The Forgotten", "rank": "A"}, - "fork_whuakuang": {"name": "The Forgotten", "rank": "A"}, - "fork_jingmotingyuan": {"name": "Camellia Society", "rank": "S"}, - "fork_Kite": {"name": "Watch Your Heads!", "rank": "A"}, - "fork_yuren": {"name": "Umbrella", "rank": "A"}, - "fork_yaodao": {"name": "Drawn Blade", "rank": "A"}, "fork_Arachne": {"name": "Eternal Waltz", "rank": "S"}, "fork_BitGame": {"name": "Blow up the Crowd", "rank": "S"}, "fork_BitterCake": {"name": "The Good, The Bad, The Bitter", "rank": "A"}, "fork_BlackBook": {"name": "Youthful Fantasy", "rank": "S"}, - "fork_Baozhatang": {"name": "Fluff of Fearlessness", "rank": "S"}, + "fork_BlastCandy": {"name": "Fluff of Fearlessness", "rank": "S"}, "fork_bopu": {"name": "Cosmos Daze, Wild Reverie", "rank": "A"}, - "fork_Quanjitang": {"name": "Fluff of Fortitude", "rank": "S"}, + "fork_BoxingCandy": {"name": "Fluff of Fortitude", "rank": "S"}, "fork_Butterfly": {"name": "Reality Refuge", "rank": "S"}, "fork_Castle": {"name": "Call of the Twisted City", "rank": "A"}, - "fork_Qiaoqiao": {"name": "Time Bandit", "rank": "A"}, + "fork_Crowbar": {"name": "Time Bandit", "rank": "A"}, + "fork_Door": {"name": "The Wrong Gate", "rank": "S"}, + "fork_dustbin": {"name": "Dangerous Game", "rank": "B"}, + "fork_GoldWool": {"name": "What's Desired", "rank": "S"}, "fork_jiaojuan": {"name": "Shiny Days", "rank": "A"}, + "fork_jingmotingyuan": {"name": "Camellia Society", "rank": "S"}, + "fork_Kite": {"name": "Watch Your Heads!", "rank": "A"}, "fork_KnightCandy": {"name": "Fluff of Ferocity", "rank": "S"}, "fork_koinobori": {"name": "A Time Will Come", "rank": "A"}, "fork_lingganzhongjiezhe": {"name": "Mind Royale", "rank": "A"}, + "fork_LunarPhase": {"name": "Blushing Mirage", "rank": "S"}, "fork_mamen": {"name": "Contemplative Cat", "rank": "S"}, + "fork_mofeikesi": {"name": "Good Boy's Grand Adventure", "rank": "S"}, "fork_moon": {"name": "Stellar Veil", "rank": "S"}, "fork_MotorCandy": {"name": "Fluff of Fleetness", "rank": "S"}, "fork_Nakupeda": {"name": "Your Happiness is Priceless", "rank": "S"}, "fork_NestBird": {"name": "Tears Beneath the Mask", "rank": "S"}, + "fork_nonos": {"name": "First Step to Success", "rank": "B"}, "fork_oulaquantao": {"name": "Oraora!", "rank": "A"}, + "fork_PaperPlane": {"name": "Clear Skies", "rank": "A"}, "fork_PoliceRat": {"name": "Hethereau's Keeper", "rank": "S"}, + "fork_Prokaryon": {"name": "Us.", "rank": "B"}, "fork_rishi": {"name": "Day Off", "rank": "S"}, "fork_Rose": {"name": "The Last Rose", "rank": "S"}, "fork_snowman": {"name": "The Fools' Spring", "rank": "A"}, @@ -41,9 +39,11 @@ "fork_TigerTally": {"name": "Ready-Ready", "rank": "S"}, "fork_Time": {"name": "Marching Beyond Time", "rank": "S"}, "fork_tuansanlang": {"name": "The Great Thief", "rank": "A"}, + "fork_vine": {"name": "Be Happy", "rank": "B"}, "fork_Whale": {"name": "Song of the Whale", "rank": "S"}, "fork_worldrain": {"name": "The Rain That Shook the World", "rank": "S"}, - "fork_Wushoutieyu": {"name": "Raging Flames", "rank": "S"}, - "fork_GoldWool": {"name": "What's Desired", "rank": "S"}, - "fork_LunarPhase": {"name": "Blushing Mirage", "rank": "S"} + "fork_wuhuakuang": {"name": "The Forgotten", "rank": "A"}, + "fork_wushoutieyu": {"name": "Raging Flames", "rank": "S"}, + "fork_yaodao": {"name": "Drawn Blade", "rank": "A"}, + "fork_yuren": {"name": "Umbrella", "rank": "A"} } diff --git a/mappings/characters.json b/mappings/characters.json index 842b603..b037890 100644 --- a/mappings/characters.json +++ b/mappings/characters.json @@ -19,5 +19,6 @@ "1070": {"name": "Aurelia", "rank": "A"}, "1071": {"name": "Chaos", "rank": "S"}, "1073": {"name": "Chiz", "rank": "S"}, + "1075": {"name": "Iroi", "rank": "S"}, "1076": {"name": "Shinku", "rank": "S"} } diff --git a/mappings/items.json b/mappings/items.json index c795db2..f8daf7e 100644 --- a/mappings/items.json +++ b/mappings/items.json @@ -1,26 +1,25 @@ { - "DiceNormal": {"type": "item", "name": "Fabricated Dice", "rank": "S"}, "Dice_ticket_01": {"type": "item", "name": "Warp Piece", "rank": "S"}, "Dice_ticket_02": {"type": "item", "name": "Lost Piece", "rank": "A"}, "Dicelimite": {"type": "item", "name": "Solid Dice", "rank": "S"}, - - "Fashion_Glide_1010": {"type": "cosmetic", "name": "Glider - Underboss-of-the-Underboss", "rank": "A"}, - "Fashion_vehicle_1010_V008": {"type": "cosmetic", "name": "Mod Parts - Tiger Incoming! - Livery", "rank": "S"}, - "Fashion_character_1010": {"type": "cosmetic", "name": "Character Skin - Phoenix Kick", "rank": "S"}, - - "Fashion_Glide_1052": {"type": "cosmetic", "name": "Glider - Orchid Breeze", "rank": "A"}, - "Fashion_vehicle_1052_V024": {"type": "cosmetic", "name": "Mod Parts - Autumn Haze - Livery", "rank": "S"}, - "Fashion_character_1052_01": {"type": "cosmetic", "name": "Character Skin - Priceless Orchid", "rank": "S"}, - - "Fashion_Glide_1004": {"type": "cosmetic", "name": "Glider - Tomato Duo", "rank": "A"}, - "Fashion_vehicle_1004_V021": {"type": "cosmetic", "name": "Mod Parts - Tomato Cruise - Livery", "rank": "S"}, - "Fashion_character_1004_01": {"type": "cosmetic", "name": "Character Skin - Gilded Rhapsody", "rank": "S"}, - - "Fashion_Glide_1071": {"type": "cosmetic", "name": "Glider - Skyrider", "rank": "A"}, - "Fashion_vehicle_1071_V010": {"type": "cosmetic", "name": "Mod Parts - Novis ST-X 950 - Livery", "rank": "S"}, - "Fashion_character_1071_01": {"type": "cosmetic", "name": "Character Skin - Clear Skies", "rank": "S"}, - - "Fashion_Glide_1076": {"type": "cosmetic", "name": "Glider - Overcast Canopy", "rank": "A"}, - "Fashion_vehicle_1076_V024": {"type": "cosmetic", "name": "Mod Parts - Hidden Dragon - Livery", "rank": "S"}, - "Fashion_character_1076_01": {"type": "cosmetic", "name": "Character Skin - Student of Terrasea", "rank": "S"} + "DiceNormal": {"type": "item", "name": "Fabricated Dice", "rank": "S"}, + "Fashion_character_1004_01": {"type": "cosmetic", "name": "Gilded Rhapsody", "rank": "S"}, + "Fashion_character_1010": {"type": "cosmetic", "name": "Phoenix Kick", "rank": "S"}, + "Fashion_character_1052_01": {"type": "cosmetic", "name": "Priceless Orchid", "rank": "S"}, + "Fashion_character_1071_01": {"type": "cosmetic", "name": "Clear Skies", "rank": "S"}, + "Fashion_character_1075_01": {"type": "cosmetic", "name": "Summer Dream", "rank": "S"}, + "Fashion_character_1076_01": {"type": "cosmetic", "name": "Student of Terrasea", "rank": "S"}, + "Fashion_Glide_1004": {"type": "cosmetic", "name": "Tomato Duo", "rank": "A"}, + "Fashion_Glide_1010": {"type": "cosmetic", "name": "Underboss-of-the-Underboss", "rank": "A"}, + "Fashion_Glide_1052": {"type": "cosmetic", "name": "Orchid Breeze", "rank": "A"}, + "Fashion_Glide_1071": {"type": "cosmetic", "name": "Skyrider", "rank": "A"}, + "Fashion_glide_1075": {"type": "cosmetic", "name": "Skyrider", "rank": "A"}, + "Fashion_Glide_1076": {"type": "cosmetic", "name": "Overcast Canopy", "rank": "A"}, + "Fashion_vehicle_1004_V021": {"type": "cosmetic", "name": "Tomato Cruise - Livery", "rank": "S"}, + "Fashion_vehicle_1010_V008": {"type": "cosmetic", "name": "Tiger Incoming! - Livery", "rank": "S"}, + "Fashion_vehicle_1052_V024": {"type": "cosmetic", "name": "Autumn Haze - Livery", "rank": "S"}, + "Fashion_vehicle_1071_V010": {"type": "cosmetic", "name": "Hound Blitz - Livery", "rank": "S"}, + "Fashion_vehicle_1075_V006": {"type": "cosmetic", "name": "Hidden Dragon - Livery", "rank": "S"}, + "Fashion_vehicle_1076_V024": {"type": "cosmetic", "name": "Hidden Dragon - Livery", "rank": "S"}, + "Fashion_vehicleSkin_1052_V016": {"type": "cosmetic", "name": "Golden Breeze - Livery", "rank": "S"} } diff --git a/src/nte_history_exporter/mapping_update.py b/src/nte_history_exporter/mapping_update.py new file mode 100644 index 0000000..e231c34 --- /dev/null +++ b/src/nte_history_exporter/mapping_update.py @@ -0,0 +1,387 @@ +from __future__ import annotations + +import hashlib +import json +import os +import tempfile +import urllib.request +from dataclasses import dataclass +from pathlib import Path +from typing import Any + + +NTE_ASSETS_REPOSITORY = "https://github.com/Waifus-Grace/NTE_Assets" +DEFAULT_SOURCE_REF = "main" +ASSET_TABLES = { + "characters": "DataTable/Character/DT_Character.json", + "arcs": "DataTable/Fork/DT_ForkItemData.json", + "inventory": "DataTable/Inventory/DT_ItemConfig.json", + "capital_inventory": "DataTable/Inventory/DT_CapitalItemConfig.json", + "appearances": "DataTable/Character/Appearance/DT_AppearanceData.json", + "illustrations": "DataTable/Gacha/GachaIllustrate.json", + "localization": "Localization/en/game.json", +} +REWARD_MAPPING_FILES = ("arcs.json", "characters.json", "items.json") +RANK_BY_QUALITY = { + "EItemQuality::ITEM_QUALITY_ORANGE": "S", + "EItemQuality::ITEM_QUALITY_PURPLE": "A", + "EItemQuality::ITEM_QUALITY_BLUE": "B", +} + + +class MappingUpdateError(ValueError): + pass + + +@dataclass(frozen=True) +class AssetSource: + tables: dict[str, dict[str, Any]] + source: str + source_ref: str | None + sha256: str + file_sha256: dict[str, str] + + +@dataclass(frozen=True) +class MappingUpdateResult: + mappings: dict[str, dict[str, dict[str, Any]]] + report: dict[str, Any] + + @property + def change_count(self) -> int: + changes = self.report["changes"] + return changes["additions"] + changes["updates"] + changes["deletions"] + + +def load_assets(*, assets_root: Path | None = None, source_ref: str = DEFAULT_SOURCE_REF) -> AssetSource: + tables: dict[str, dict[str, Any]] = {} + file_hashes: dict[str, str] = {} + combined = hashlib.sha256() + + for label, relative_path in ASSET_TABLES.items(): + if assets_root is not None: + path = assets_root / Path(relative_path) + try: + raw = path.read_bytes() + except OSError as exc: + raise MappingUpdateError(f"cannot read NTE_Assets table {path}: {exc}") from exc + else: + url = f"https://raw.githubusercontent.com/Waifus-Grace/NTE_Assets/{source_ref}/{relative_path}" + request = urllib.request.Request(url, headers={"User-Agent": "nte-history-exporter-mapping-update"}) + try: + with urllib.request.urlopen(request, timeout=60) as response: + raw = response.read() + except OSError as exc: + raise MappingUpdateError(f"cannot download NTE_Assets table {relative_path}: {exc}") from exc + + try: + document = json.loads(raw.decode("utf-8-sig")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise MappingUpdateError(f"invalid NTE_Assets JSON in {relative_path}: {exc}") from exc + tables[label] = ( + _extract_localization(document, relative_path) + if label == "localization" + else _extract_rows(document, relative_path) + ) + digest = hashlib.sha256(raw).hexdigest() + file_hashes[relative_path] = digest + combined.update(relative_path.encode("utf-8")) + combined.update(b"\0") + combined.update(raw) + + source = str(assets_root.resolve()) if assets_root is not None else NTE_ASSETS_REPOSITORY + return AssetSource( + tables=tables, + source=source, + source_ref=None if assets_root is not None else source_ref, + sha256=combined.hexdigest(), + file_sha256=file_hashes, + ) + + +def load_current_mappings(directory: Path) -> dict[str, dict[str, dict[str, Any]]]: + mappings = {} + for filename in REWARD_MAPPING_FILES: + path = directory / filename + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise MappingUpdateError(f"cannot read {path}: {exc}") from exc + if not isinstance(value, dict): + raise MappingUpdateError(f"{path} must contain an object") + mappings[filename] = value + validate_mappings(mappings) + return mappings + + +def build_mapping_update( + current: dict[str, dict[str, dict[str, Any]]], + assets: AssetSource, +) -> MappingUpdateResult: + validate_mappings(current) + translations = _build_translation_index(assets.tables["localization"]) + output = { + "arcs.json": _build_primary_mapping(assets.tables["arcs"], "arc", translations), + "characters.json": _build_primary_mapping(assets.tables["characters"], "character", translations), + "items.json": _build_item_mapping(assets.tables, translations), + } + validate_mappings(output) + + additions: dict[str, list[str]] = {} + updates: dict[str, list[str]] = {} + deletions: dict[str, list[str]] = {} + for filename in REWARD_MAPPING_FILES: + old = current[filename] + new = output[filename] + additions[filename] = sorted(new.keys() - old.keys(), key=str.casefold) + deletions[filename] = sorted(old.keys() - new.keys(), key=str.casefold) + updates[filename] = sorted( + (item_id for item_id in old.keys() & new.keys() if old[item_id] != new[item_id]), + key=str.casefold, + ) + + report = { + "schema_version": 2, + "source": assets.source, + "source_ref": assets.source_ref, + "source_sha256": assets.sha256, + "source_file_sha256": assets.file_sha256, + "safety": { + "authoritative_snapshot": True, + "deletions_allowed": True, + "pool_mappings_touched": False, + "uid_inputs_touched": False, + }, + "changes": { + "additions": sum(map(len, additions.values())), + "updates": sum(map(len, updates.values())), + "deletions": sum(map(len, deletions.values())), + }, + "additions_by_file": additions, + "updates_by_file": updates, + "deletions_by_file": deletions, + "output_counts": {filename: len(entries) for filename, entries in output.items()}, + } + return MappingUpdateResult(output, report) + + +def write_update(result: MappingUpdateResult, directory: Path) -> None: + directory.mkdir(parents=True, exist_ok=True) + for filename in REWARD_MAPPING_FILES: + _atomic_write(directory / filename, _dump_mapping(result.mappings[filename])) + report_text = json.dumps(result.report, ensure_ascii=False, indent=2, sort_keys=True) + "\n" + _atomic_write(directory / "mapping-update-report.json", report_text) + + +def apply_update(result: MappingUpdateResult, directory: Path) -> None: + """Replace all reward maps as one transaction, rolling back on failure.""" + directory.mkdir(parents=True, exist_ok=True) + originals: dict[Path, bytes | None] = {} + written: list[Path] = [] + for filename in REWARD_MAPPING_FILES: + path = directory / filename + originals[path] = path.read_bytes() if path.exists() else None + try: + for filename in REWARD_MAPPING_FILES: + path = directory / filename + _atomic_write(path, _dump_mapping(result.mappings[filename])) + written.append(path) + except OSError: + for path in reversed(written): + original = originals[path] + if original is None: + path.unlink(missing_ok=True) + else: + _atomic_write_bytes(path, original) + raise + + +def validate_mappings(mappings: dict[str, dict[str, dict[str, Any]]]) -> None: + if set(mappings) != set(REWARD_MAPPING_FILES): + raise MappingUpdateError(f"mapping set must be exactly {REWARD_MAPPING_FILES}") + locations: dict[str, str] = {} + folded: dict[str, str] = {} + for filename in REWARD_MAPPING_FILES: + entries = mappings[filename] + if not isinstance(entries, dict): + raise MappingUpdateError(f"{filename} must contain an object") + for item_id, meta in entries.items(): + if not isinstance(item_id, str) or not item_id: + raise MappingUpdateError(f"{filename} contains an invalid item ID") + if item_id in locations: + raise MappingUpdateError(f"item ID {item_id} appears in {locations[item_id]} and {filename}") + locations[item_id] = filename + case_key = item_id.casefold() + if case_key in folded and folded[case_key] != item_id: + raise MappingUpdateError(f"case-insensitive duplicate IDs: {folded[case_key]} and {item_id}") + folded[case_key] = item_id + if not isinstance(meta, dict): + raise MappingUpdateError(f"{filename}:{item_id} must be an object") + name = meta.get("name") + if not isinstance(name, str) or not name.strip(): + raise MappingUpdateError(f"{filename}:{item_id}.name must be a non-empty string") + rank = meta.get("rank") + if filename == "arcs.json" and rank not in {"S", "A", "B"}: + raise MappingUpdateError(f"{filename}:{item_id}.rank must be S, A, or B") + if filename == "characters.json" and rank not in {"S", "A"}: + raise MappingUpdateError(f"{filename}:{item_id}.rank must be S or A") + if filename == "items.json": + if meta.get("type") not in {"item", "cosmetic"}: + raise MappingUpdateError(f"{filename}:{item_id}.type must be item or cosmetic") + if rank not in {"S", "A", "B"}: + raise MappingUpdateError(f"{filename}:{item_id}.rank must be S, A, or B") + + +def _extract_rows(document: Any, relative_path: str) -> dict[str, Any]: + candidates = document if isinstance(document, list) else [document] + for candidate in candidates: + if not isinstance(candidate, dict): + continue + rows = candidate.get("Rows") + if isinstance(rows, dict): + return rows + if all(isinstance(key, str) for key in candidate) and candidate: + return candidate + raise MappingUpdateError(f"NTE_Assets table {relative_path} does not contain a Rows object") + + +def _extract_localization(document: Any, relative_path: str) -> dict[str, Any]: + if not isinstance(document, dict) or not all(isinstance(value, dict) for value in document.values()): + raise MappingUpdateError(f"NTE_Assets localization {relative_path} must contain namespace objects") + return document + + +def _build_translation_index(localization: dict[str, Any]) -> dict[str, list[tuple[str, str]]]: + result: dict[str, list[tuple[str, str]]] = {} + for namespace, entries in localization.items(): + for key, value in entries.items(): + if not isinstance(key, str) or not isinstance(value, str): + raise MappingUpdateError(f"invalid English localization entry {namespace}:{key}") + result.setdefault(key, []).append((namespace, value)) + return result + + +def _build_primary_mapping( + rows: dict[str, Any], + kind: str, + translations: dict[str, list[tuple[str, str]]], +) -> dict[str, dict[str, Any]]: + result = {} + for item_id in sorted(rows, key=lambda value: (value.casefold(), value)): + meta = _normalise_row(item_id, rows[item_id], translations) + if kind == "character" and meta["rank"] not in {"S", "A"}: + raise MappingUpdateError(f"unexpected character quality for {item_id}") + result[item_id] = {"name": meta["name"], "rank": meta["rank"]} + return result + + +def _build_item_mapping( + tables: dict[str, dict[str, Any]], + translations: dict[str, list[tuple[str, str]]], +) -> dict[str, dict[str, Any]]: + inventory = _casefold_index(tables["inventory"], "inventory") + capital = _casefold_index(tables["capital_inventory"], "capital inventory") + appearances = _casefold_index(tables["appearances"], "appearances") + result: dict[str, dict[str, Any]] = {} + for illustrated_id in tables["illustrations"]: + folded = illustrated_id.casefold() + if illustrated_id.isdigit() or folded.startswith("fork_") or folded.startswith("characterawaken_"): + continue + sources = (inventory, capital, appearances) + match = next((source.get(folded) for source in sources if folded in source), None) + if match is None: + raise MappingUpdateError(f"illustrated reward {illustrated_id} is missing from item tables") + canonical_id, row = match + if folded.startswith("fashion_glide_") and folded in appearances: + canonical_id = appearances[folded][0] + meta = _normalise_row(canonical_id, row, translations) + result[canonical_id] = { + "type": "cosmetic" if canonical_id.casefold().startswith("fashion_") else "item", + "name": meta["name"], + "rank": meta["rank"], + } + return dict(sorted(result.items(), key=lambda pair: (pair[0].casefold(), pair[0]))) + + +def _casefold_index(rows: dict[str, Any], label: str) -> dict[str, tuple[str, Any]]: + result: dict[str, tuple[str, Any]] = {} + for item_id, row in rows.items(): + folded = item_id.casefold() + if folded in result and result[folded][0] != item_id: + raise MappingUpdateError(f"{label} has case-insensitive duplicate IDs: {result[folded][0]} and {item_id}") + result[folded] = (item_id, row) + return result + + +def _normalise_row( + item_id: str, + row: Any, + translations: dict[str, list[tuple[str, str]]], +) -> dict[str, str]: + if not isinstance(row, dict): + raise MappingUpdateError(f"NTE_Assets row {item_id} must be an object") + item_name = row.get("ItemName") or row.get("Name") + if not isinstance(item_name, dict): + raise MappingUpdateError(f"NTE_Assets row {item_id} has no ItemName") + name = _translate_name(item_id, item_name, translations) + quality = row.get("ItemQuality") or row.get("Quality") + rank = RANK_BY_QUALITY.get(quality) + if rank is None: + raise MappingUpdateError(f"NTE_Assets row {item_id} has unsupported quality {quality!r}") + return {"name": name.strip(), "rank": rank} + + +def _translate_name( + item_id: str, + string_reference: dict[str, Any], + translations: dict[str, list[tuple[str, str]]], +) -> str: + key = string_reference.get("Key") + if not isinstance(key, str) or not key: + raise MappingUpdateError(f"NTE_Assets row {item_id} has no localization key") + table_id = string_reference.get("TableId") + expected_namespace = None + if isinstance(table_id, str) and table_id: + expected_namespace = table_id.rsplit("/", 1)[-1].split(".", 1)[0] + matches = translations.get(key, []) + preferred = [value for namespace, value in matches if namespace == expected_namespace] + if len(preferred) == 1: + name = preferred[0] + elif len(matches) == 1: + name = matches[0][1] + elif not matches: + raise MappingUpdateError(f"English localization is missing {item_id} key {key}") + else: + namespaces = ", ".join(namespace for namespace, _value in matches) + raise MappingUpdateError(f"English localization key {key} for {item_id} is ambiguous: {namespaces}") + if not name.strip(): + raise MappingUpdateError(f"English localization key {key} for {item_id} is empty") + return name + + +def _dump_mapping(mapping: dict[str, dict[str, Any]]) -> str: + lines = ["{"] + entries = list(mapping.items()) + for index, (item_id, meta) in enumerate(entries): + comma = "," if index < len(entries) - 1 else "" + key = json.dumps(item_id, ensure_ascii=False) + value = json.dumps(meta, ensure_ascii=False, separators=(", ", ": ")) + lines.append(f" {key}: {value}{comma}") + lines.append("}") + return "\n".join(lines) + "\n" + + +def _atomic_write(path: Path, text: str) -> None: + _atomic_write_bytes(path, text.encode("utf-8")) + + +def _atomic_write_bytes(path: Path, content: bytes) -> None: + with tempfile.NamedTemporaryFile( + mode="wb", dir=path.parent, prefix=f".{path.name}.", suffix=".tmp", delete=False + ) as handle: + handle.write(content) + temporary = Path(handle.name) + try: + os.replace(temporary, path) + finally: + temporary.unlink(missing_ok=True) diff --git a/tests/fixtures/README.md b/tests/fixtures/README.md index 7c54fd3..abdedc9 100644 --- a/tests/fixtures/README.md +++ b/tests/fixtures/README.md @@ -21,3 +21,12 @@ points gifts, chase rewards, batched pages, and bit-packed responses. Do not replace this file with a real `.pcap`, `.flows`, or exported account history. Add new cases by constructing the smallest relevant payload, replacing all timestamps and endpoints, and extending the privacy assertions. + +## Synthetic NTE_Assets fixture + +`nte_assets/` mirrors only the six table paths and English localization file +consumed by the mapping updater. Its rows, IDs, and names are deliberately +synthetic and contain no account or personal data. The fixture exercises +authoritative additions, metadata updates, deletions, translation lookup, +illustration filtering, and canonical ID casing without depending on the +network or copying a live game-data snapshot into the repository. diff --git a/tests/fixtures/nte_assets/DataTable/Character/Appearance/DT_AppearanceData.json b/tests/fixtures/nte_assets/DataTable/Character/Appearance/DT_AppearanceData.json new file mode 100644 index 0000000..d513855 --- /dev/null +++ b/tests/fixtures/nte_assets/DataTable/Character/Appearance/DT_AppearanceData.json @@ -0,0 +1 @@ +[{"Rows":{"Fashion_Glide_2000":{"Name":{"TableId":"/Game/Text/ST_Appearance.ST_Appearance","Key":"glider_2000","LocalizedString":"Wrong Embedded Appearance"},"Quality":"EItemQuality::ITEM_QUALITY_PURPLE"}}}] diff --git a/tests/fixtures/nte_assets/DataTable/Character/DT_Character.json b/tests/fixtures/nte_assets/DataTable/Character/DT_Character.json new file mode 100644 index 0000000..9866018 --- /dev/null +++ b/tests/fixtures/nte_assets/DataTable/Character/DT_Character.json @@ -0,0 +1 @@ +[{"Rows":{"1003":{"ItemName":{"TableId":"/Game/Text/ST_Player.ST_Player","Key":"character_1003","LocalizedString":"Wrong Embedded Character"},"ItemQuality":"EItemQuality::ITEM_QUALITY_ORANGE"},"1099":{"ItemName":{"TableId":"/Game/Text/ST_Player.ST_Player","Key":"character_1099","LocalizedString":"Wrong Embedded New Character"},"ItemQuality":"EItemQuality::ITEM_QUALITY_PURPLE"}}}] diff --git a/tests/fixtures/nte_assets/DataTable/Fork/DT_ForkItemData.json b/tests/fixtures/nte_assets/DataTable/Fork/DT_ForkItemData.json new file mode 100644 index 0000000..fe59f68 --- /dev/null +++ b/tests/fixtures/nte_assets/DataTable/Fork/DT_ForkItemData.json @@ -0,0 +1 @@ +[{"Rows":{"fork_alpha":{"ItemName":{"TableId":"/Game/Text/ST_Fork.ST_Fork","Key":"fork_alpha_name","LocalizedString":"Wrong Embedded Alpha"},"ItemQuality":"EItemQuality::ITEM_QUALITY_BLUE"},"fork_new":{"ItemName":{"TableId":"/Game/Text/ST_Fork.ST_Fork","Key":"fork_new_name","LocalizedString":"Wrong Embedded New Arc"},"ItemQuality":"EItemQuality::ITEM_QUALITY_ORANGE"}}}] diff --git a/tests/fixtures/nte_assets/DataTable/Gacha/GachaIllustrate.json b/tests/fixtures/nte_assets/DataTable/Gacha/GachaIllustrate.json new file mode 100644 index 0000000..bd2fdf3 --- /dev/null +++ b/tests/fixtures/nte_assets/DataTable/Gacha/GachaIllustrate.json @@ -0,0 +1 @@ +[{"Rows":{"1003":{},"1099":{},"fork_alpha":{},"fork_new":{},"DIceNormal":{},"Dice_ticket_03":{},"Fashion_glide_2000":{},"Characterawaken_1003":{}}}] diff --git a/tests/fixtures/nte_assets/DataTable/Inventory/DT_CapitalItemConfig.json b/tests/fixtures/nte_assets/DataTable/Inventory/DT_CapitalItemConfig.json new file mode 100644 index 0000000..edf0e46 --- /dev/null +++ b/tests/fixtures/nte_assets/DataTable/Inventory/DT_CapitalItemConfig.json @@ -0,0 +1 @@ +[{"Rows":{}}] diff --git a/tests/fixtures/nte_assets/DataTable/Inventory/DT_ItemConfig.json b/tests/fixtures/nte_assets/DataTable/Inventory/DT_ItemConfig.json new file mode 100644 index 0000000..a98e2f8 --- /dev/null +++ b/tests/fixtures/nte_assets/DataTable/Inventory/DT_ItemConfig.json @@ -0,0 +1 @@ +[{"Rows":{"DiceNormal":{"ItemName":{"TableId":"/Game/Text/ST_Item.ST_Item","Key":"dice_normal","LocalizedString":"Wrong Embedded Dice"},"ItemQuality":"EItemQuality::ITEM_QUALITY_ORANGE"},"Dice_ticket_03":{"ItemName":{"TableId":"/Game/Text/ST_Item.ST_Item","Key":"ticket_03","LocalizedString":"Wrong Embedded Ticket"},"ItemQuality":"EItemQuality::ITEM_QUALITY_BLUE"},"Fashion_glide_2000":{"ItemName":{"TableId":"/Game/Text/ST_Appearance.ST_Appearance","Key":"glider_2000","LocalizedString":"Wrong Embedded Glider"},"ItemQuality":"EItemQuality::ITEM_QUALITY_PURPLE"},"UnusedItem":{"ItemName":{"TableId":"/Game/Text/ST_Item.ST_Item","Key":"unused_item","LocalizedString":"Wrong Embedded Unused"},"ItemQuality":"EItemQuality::ITEM_QUALITY_BLUE"}}}] diff --git a/tests/fixtures/nte_assets/Localization/en/game.json b/tests/fixtures/nte_assets/Localization/en/game.json new file mode 100644 index 0000000..5bda83c --- /dev/null +++ b/tests/fixtures/nte_assets/Localization/en/game.json @@ -0,0 +1,18 @@ +{ + "ST_Player": { + "character_1003": "Old Character Renamed", + "character_1099": "New Character" + }, + "ST_Fork": { + "fork_alpha_name": "Alpha Arc", + "fork_new_name": "New Arc" + }, + "ST_Item": { + "dice_normal": "Fabricated Dice", + "ticket_03": "New Ticket", + "unused_item": "Not A Pull Reward" + }, + "ST_Appearance": { + "glider_2000": "Inventory Glider" + } +} diff --git a/tests/test_mapping_update.py b/tests/test_mapping_update.py new file mode 100644 index 0000000..a715f16 --- /dev/null +++ b/tests/test_mapping_update.py @@ -0,0 +1,184 @@ +from __future__ import annotations + +import json +import unittest +from copy import deepcopy +from dataclasses import replace +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest.mock import patch + +from nte_history_exporter.mapping_update import ( + MappingUpdateError, + apply_update, + build_mapping_update, + load_assets, + load_current_mappings, + write_update, +) +from tests.support import FIXTURES + + +SAMPLE_ASSETS = FIXTURES / "nte_assets" + + +def sample_current(): + return { + "arcs.json": { + "fork_alpha": {"name": "Stale Alpha Name", "rank": "B"}, + "fork_retired": {"name": "Retired Arc", "rank": "A"}, + }, + "characters.json": { + "1003": {"name": "Stale Character Name", "rank": "S"}, + "1001": {"name": "Retired Character", "rank": "A"}, + }, + "items.json": { + "DiceNormal": {"type": "item", "name": "Fabricated Dice", "rank": "S"}, + "OldTicket": {"type": "item", "name": "Retired Ticket", "rank": "B"}, + }, + } + + +def build_sample_update(): + current = sample_current() + return current, build_mapping_update(current, load_assets(assets_root=SAMPLE_ASSETS)) + + +class MappingUpdateTests(unittest.TestCase): + def test_mapping_update_is_an_authoritative_snapshot(self): + current, result = build_sample_update() + + self.assertEqual(result.mappings["characters.json"]["1003"], {"name": "Old Character Renamed", "rank": "S"}) + self.assertEqual(result.mappings["characters.json"]["1099"], {"name": "New Character", "rank": "A"}) + self.assertNotIn("1001", result.mappings["characters.json"]) + self.assertNotIn("fork_retired", result.mappings["arcs.json"]) + self.assertNotIn("OldTicket", result.mappings["items.json"]) + self.assertNotEqual(result.mappings["characters.json"]["1003"], current["characters.json"]["1003"]) + self.assertEqual(result.report["changes"], {"additions": 4, "updates": 2, "deletions": 3}) + self.assertIs(result.report["safety"]["authoritative_snapshot"], True) + self.assertIs(result.report["safety"]["deletions_allowed"], True) + self.assertIs(result.report["safety"]["uid_inputs_touched"], False) + self.assertIs(result.report["safety"]["pool_mappings_touched"], False) + + def test_items_use_illustrations_as_filter_and_asset_tables_as_authority(self): + _current, result = build_sample_update() + + self.assertEqual( + result.mappings["items.json"]["Fashion_Glide_2000"], + {"type": "cosmetic", "name": "Inventory Glider", "rank": "A"}, + ) + self.assertIn("DiceNormal", result.mappings["items.json"]) + self.assertNotIn("DIceNormal", result.mappings["items.json"]) + self.assertNotIn("UnusedItem", result.mappings["items.json"]) + self.assertFalse(any(key.startswith("Characterawaken_") for key in result.mappings["items.json"])) + + def test_embedded_localized_strings_are_ignored(self): + _current, result = build_sample_update() + + self.assertEqual(result.mappings["characters.json"]["1003"]["name"], "Old Character Renamed") + self.assertNotEqual(result.mappings["characters.json"]["1003"]["name"], "Wrong Embedded Character") + + def test_mapping_update_rejects_missing_english_translation(self): + assets = load_assets(assets_root=SAMPLE_ASSETS) + tables = deepcopy(assets.tables) + del tables["localization"]["ST_Player"]["character_1003"] + + with self.assertRaisesRegex(MappingUpdateError, "English localization is missing"): + build_mapping_update(sample_current(), replace(assets, tables=tables)) + + def test_mapping_update_rejects_ambiguous_translation_without_namespace_match(self): + assets = load_assets(assets_root=SAMPLE_ASSETS) + tables = deepcopy(assets.tables) + tables["inventory"]["Fashion_glide_2000"]["ItemName"]["TableId"] = "/Game/Text/Unknown.Unknown" + tables["localization"]["OtherNamespace"] = {"glider_2000": "Other Glider"} + + with self.assertRaisesRegex(MappingUpdateError, "is ambiguous"): + build_mapping_update(sample_current(), replace(assets, tables=tables)) + + def test_local_asset_source_has_reproducible_provenance(self): + first = load_assets(assets_root=SAMPLE_ASSETS) + second = load_assets(assets_root=SAMPLE_ASSETS) + + self.assertEqual(first.sha256, second.sha256) + self.assertEqual(first.file_sha256, second.file_sha256) + self.assertEqual(first.source, str(SAMPLE_ASSETS.resolve())) + self.assertIsNone(first.source_ref) + + def test_mapping_update_writes_deterministic_review_artifacts(self): + _current, result = build_sample_update() + with TemporaryDirectory() as first_tmp, TemporaryDirectory() as second_tmp: + first = Path(first_tmp) + second = Path(second_tmp) + write_update(result, first) + write_update(result, second) + + for filename in (*result.mappings, "mapping-update-report.json"): + self.assertEqual((first / filename).read_bytes(), (second / filename).read_bytes()) + report = json.loads((first / "mapping-update-report.json").read_text(encoding="utf-8")) + self.assertEqual(report["changes"]["deletions"], 3) + + def test_mapping_update_does_not_mutate_inputs(self): + current = sample_current() + original = deepcopy(current) + + build_mapping_update(current, load_assets(assets_root=SAMPLE_ASSETS)) + + self.assertEqual(current, original) + + def test_mapping_update_apply_touches_only_reward_mapping_files(self): + _current, result = build_sample_update() + with TemporaryDirectory() as tmp: + directory = Path(tmp) + for filename, mapping in sample_current().items(): + (directory / filename).write_text(json.dumps(mapping), encoding="utf-8") + protected = directory / "permanent_board.json" + protected.write_text('{"protected": true}\n', encoding="utf-8") + + apply_update(result, directory) + + self.assertEqual(load_current_mappings(directory), result.mappings) + self.assertEqual(protected.read_text(encoding="utf-8"), '{"protected": true}\n') + self.assertFalse((directory / "mapping-update-report.json").exists()) + + def test_mapping_update_apply_rolls_back_partial_write(self): + _current, result = build_sample_update() + with TemporaryDirectory() as tmp: + directory = Path(tmp) + originals = {} + for filename, mapping in sample_current().items(): + destination = directory / filename + destination.write_text(json.dumps(mapping), encoding="utf-8") + originals[filename] = destination.read_bytes() + + from nte_history_exporter import mapping_update + + real_write = mapping_update._atomic_write + calls = 0 + + def fail_second_write(path, text): + nonlocal calls + calls += 1 + if calls == 2: + raise OSError("synthetic write failure") + return real_write(path, text) + + with ( + patch("nte_history_exporter.mapping_update._atomic_write", side_effect=fail_second_write), + self.assertRaisesRegex(OSError, "synthetic write failure"), + ): + apply_update(result, directory) + + for filename, original in originals.items(): + self.assertEqual((directory / filename).read_bytes(), original) + + def test_mapping_update_rejects_unresolved_illustrated_rewards(self): + assets = load_assets(assets_root=SAMPLE_ASSETS) + tables = deepcopy(assets.tables) + tables["illustrations"]["MissingReward"] = {} + + with self.assertRaisesRegex(MappingUpdateError, "MissingReward"): + build_mapping_update(sample_current(), replace(assets, tables=tables)) + + def test_load_assets_rejects_missing_tables(self): + with TemporaryDirectory() as tmp, self.assertRaisesRegex(MappingUpdateError, "cannot read"): + load_assets(assets_root=Path(tmp)) diff --git a/tools/update_mappings.py b/tools/update_mappings.py new file mode 100644 index 0000000..9d71a80 --- /dev/null +++ b/tools/update_mappings.py @@ -0,0 +1,86 @@ +from __future__ import annotations + +import argparse +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +SRC = ROOT / "src" +if str(SRC) not in sys.path: + sys.path.insert(0, str(SRC)) + +from nte_history_exporter.mapping_update import ( # noqa: E402 + DEFAULT_SOURCE_REF, + MappingUpdateError, + apply_update, + build_mapping_update, + load_assets, + load_current_mappings, + write_update, +) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Synchronise reward mappings from the authoritative NTE_Assets tables." + ) + parser.add_argument( + "--assets-root", + type=Path, + help="local NTE_Assets checkout (downloads the required tables when omitted)", + ) + parser.add_argument( + "--source-ref", + default=DEFAULT_SOURCE_REF, + help="NTE_Assets branch, tag, or commit to download (default: main)", + ) + parser.add_argument( + "--output-dir", + type=Path, + default=ROOT / "build" / "mapping-update", + help="directory for staged mappings and the review report", + ) + parser.add_argument( + "--apply", + action="store_true", + help="replace reward mapping files after staging and validation", + ) + parser.add_argument( + "--check", + action="store_true", + help="exit 1 when the generated snapshot differs from local mappings", + ) + return parser + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + try: + assets = load_assets(assets_root=args.assets_root, source_ref=args.source_ref) + current = load_current_mappings(ROOT / "mappings") + result = build_mapping_update(current, assets) + write_update(result, args.output_dir) + if args.apply: + apply_update(result, ROOT / "mappings") + except (MappingUpdateError, OSError) as exc: + print(f"mapping update failed: {exc}", file=sys.stderr) + return 2 + + changes = result.report["changes"] + print(f"source: {assets.source}") + if assets.source_ref: + print(f"source ref: {assets.source_ref}") + print(f"staged: {args.output_dir}") + print( + "changes: " + f"{changes['additions']} additions, " + f"{changes['updates']} updates, " + f"{changes['deletions']} deletions" + ) + if args.apply: + print("applied: mappings/arcs.json, mappings/characters.json, mappings/items.json") + return 1 if args.check and result.change_count else 0 + + +if __name__ == "__main__": + raise SystemExit(main())