diff --git a/.github/workflows/update-mappings.yml b/.github/workflows/update-mappings.yml index d24223a..38068bf 100644 --- a/.github/workflows/update-mappings.yml +++ b/.github/workflows/update-mappings.yml @@ -1,4 +1,4 @@ -name: Update mappings +name: Update reward mappings on: workflow_dispatch: @@ -14,7 +14,7 @@ permissions: pull-requests: write concurrency: - group: update-mappings + group: update-reward-mappings cancel-in-progress: false jobs: @@ -23,7 +23,7 @@ jobs: runs-on: ubuntu-latest env: - AUTOMATION_BRANCH: dev/automated-mapping-update-${{ github.ref_name }} + AUTOMATION_BRANCH: dev/automated-mapping-update NTE_ASSETS_REF: ${{ inputs.source_ref }} GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} @@ -32,7 +32,7 @@ jobs: uses: actions/checkout@v5 with: fetch-depth: 0 - ref: ${{ github.ref_name }} + ref: ${{ github.event.repository.default_branch }} - name: Get Python uses: actions/setup-python@v6 @@ -42,7 +42,7 @@ jobs: - name: Install project run: python -m pip install -e . - - name: Rebuild mappings + - name: Rebuild reward mappings run: python tools/update_mappings.py --source-ref "$NTE_ASSETS_REF" --apply - name: Run tests @@ -91,7 +91,7 @@ jobs: if: steps.changes.outputs.changed == 'true' shell: bash env: - BASE_BRANCH: ${{ github.ref_name }} + BASE_BRANCH: ${{ github.event.repository.default_branch }} RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }} run: | python - <<'PY' @@ -108,7 +108,7 @@ jobs: f"- {changes['additions']} additions\n" f"- {changes['updates']} updates\n" f"- {changes['deletions']} deletions\n\n" - "The workflow rebuilt all reward and achievement mappings, resolved English text through " + "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" @@ -125,12 +125,12 @@ jobs: if [ -n "$pr_number" ]; then gh pr edit "$pr_number" \ - --title "chore: update mappings from NTE_Assets" \ + --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 mappings from NTE_Assets" \ + --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 b15f290..95c91c1 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ # NTE History Exporter -Prototype CLI exporter for **Neverness to Everness** pull history and achievements — decodes your own game traffic into sanitized JSON. +Prototype CLI exporter for **Neverness to Everness** pull history — decodes your own game traffic into sanitized JSON ready for tracker import. ![Python](https://img.shields.io/badge/python-3.10+-blue) ![Platform](https://img.shields.io/badge/platform-Windows%20%7C%20Linux%20%7C%20macOS-0078D6) @@ -23,17 +23,10 @@ Prototype CLI exporter for **Neverness to Everness** pull history and achievemen ## What It Does -The exporter decodes Permanent Board, Limited Character Board, Arc Miracle Box, -and Mystery Box history pages from captured UDP data. It also captures tracked -achievements during login, including completed and in-progress records. Pull -history is written as tracker-ready JSON and achievements use a separate -versioned JSON format. +The exporter decodes Permanent Board, Limited Character Board, Arc Miracle Box, and Mystery Box history pages from captured UDP data, applies conservative timestamp-boundary handling, and writes sanitized JSON suitable for tracker import. > [!NOTE] -> Export JSON contains decoded history or achievement records and the shareable -> NTE user UID when it can be detected. It does **not** export tokens, account -> IDs, role IDs, device IDs, server IPs, raw packets, cookies, session data, or -> other capture metadata. +> The import JSON contains decoded history rows and the shareable NTE user UID when it can be detected. It does **not** export tokens, account IDs, role IDs, device IDs, server IPs, raw packets, cookies, session data, or other capture metadata. ## Requirements @@ -97,11 +90,7 @@ From source: Or simply double-click **`run-exporter.cmd`**. > [!IMPORTANT] -> Start the exporter **before pressing Start on the game's main menu**. This is -> required for achievement capture and enables automatic user UID and -> account-server detection. If already in game, pull-history capture still -> works, but achievements require returning to the main menu and logging in -> again; missing account details are requested before saving. +> For automatic user UID and account-server detection, launch the tool **before pressing Start on the game's main menu**. If you are already in game, history capture can still work; the tool will ask for missing account details before saving. Once running, open any supported history board in game. The tool keeps listening until you press any key. After it prints and saves the results, press any key again to close the exporter. Exports are written under `exports\` as: @@ -109,7 +98,6 @@ Once running, open any supported history board in game. The tool keeps listening - `_Limited_.json` - `_Arc_.json` - `_MysteryBox_.json` -- `_Achievements_.json` If the user UID is not detected automatically, the console asks for it before saving. Leaving it blank saves as `unknown__.json`, but may prevent import on some trackers. @@ -124,22 +112,19 @@ If a page response is missed, the exporter reports the missing page number while ### Achievement export -Start the live exporter before pressing Start on the game's main menu. The -tracked achievement state is loaded during login, captured automatically, and written to +Start the live exporter before launching the game. The completed achievement +list is loaded during login, captured automatically, and written to `exports/_Achievements_*.json` as a versioned achievement export with account metadata, completion totals, and records grouped by achievement category. -Both completed and in-progress achievements are included with their current +Both completed and incomplete achievements are included with their current progress, status, and completion time when available. Generated metadata adds the localized title and description, target, quality, and rewards from the current `NTE_Assets` achievement table. Newly captured IDs that are not yet in the bundled mapping still export normally; only their optional display metadata is omitted until the mappings are refreshed. -PlayStation trophy entries are retained and identified separately +Completed PlayStation trophy entries are retained and identified separately even though they are not shown in the in-game achievement list. The console -reports completed and in-progress totals for in-game and PlayStation records as -soon as the list has been decoded. Every observed record without a completion -timestamp is classified as `in_progress`, even when its numeric progress is -zero; compound achievements can track hidden checklist progress this way. +reports the visible and PlayStation counts as soon as the list has been decoded. #### Linux/macOS @@ -206,22 +191,22 @@ The exporter automatically includes the shareable NTE user UID when it appears i ## Mapping maintenance -Reward and achievement metadata can be rebuilt directly from the latest -NTE_Assets tables and English translation files. Mapping snapshots may change -IDs while UID inputs remain untouched: +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 mappings** from the GitHub Actions tab to generate and test +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 mapping candidates directly from NTE_Assets under +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 -[Mapping updates](docs/mapping-updates.md) for the source rules and +[Reward mapping updates](docs/mapping-updates.md) for the source rules and review workflow. ## Privacy diff --git a/docs/export-format.md b/docs/export-format.md index cccc121..6004738 100644 --- a/docs/export-format.md +++ b/docs/export-format.md @@ -15,7 +15,7 @@ decoder-only offsets. "capture_source": "npcap", "exporter": { "name": "nte-history-exporter", - "version": "0.3.0" + "version": "0.2.1" }, "banner": { "id": "Lottery_Permanent", @@ -84,12 +84,6 @@ Current stable pool IDs: Avoid using `banner.name`, `reward_name`, `reward_type` or `reward_rank` as primary IDs. They are useful display fields, but may change when mapping files are updated. -Known `reward_id` values are matched case-insensitively and emitted using the -canonical casing from the bundled mapping. This prevents packet/asset casing -differences from dropping display metadata or producing different downstream -IDs. An unknown reward keeps the casing decoded from the packet. Reward IDs are -not UID inputs, so canonicalizing their casing does not change pull UIDs. - ## Stability notes Every JSON record is exported with a stable `uid`. Re-scanning deeper history can @@ -182,79 +176,6 @@ Mystery Box: } ``` -## Achievement export - -Achievements are written separately as -`_Achievements_.json`. The document contains only records -observed in the account's login response; definitions found only in the asset -mapping are not synthesized into the export. - -```json -{ - "format": "nte-achievement-export", - "format_version": 1, - "game": "Neverness to Everness", - "source": "live_capture", - "capture_source": "npcap", - "exporter": { - "name": "nte-history-exporter", - "version": "0.3.0" - }, - "scan": { - "in_game": { - "total_achievements": 367, - "completed_achievements": 310, - "in_progress_achievements": 57 - }, - "playstation": { - "total_achievements": 35, - "completed_achievements": 31, - "in_progress_achievements": 4 - } - }, - "user_uid": "218216016349", - "server_id": "23003", - "account_region": "EU", - "categories": { - "battle": [ - { - "id": "Battle_25", - "platform": "in_game", - "status": "in_progress", - "progress": 20, - "completed": false, - "completed_at": null, - "name": "Devil Within II", - "description": "Trigger Hexed ×50.", - "quality": "high", - "target": 50, - "rewards": [ - {"item_id": "Annulith", "amount": 10} - ] - } - ] - } -} -``` - -Achievement fields: - -- `scan.in_game` and `scan.playstation` separate records visible in game from - PlayStation trophy records and report completed/in-progress totals. -- `categories` groups records by the prefix of their captured ID. The category - is therefore not repeated on every record. -- `status` is `completed` when completion ticks are present and `in_progress` - otherwise. An observed compound achievement can be in progress while its - numeric `progress` is zero because individual checklist state is server-side. -- `completed_at` is a decoded UTC time for completed records and `null` for - in-progress records. -- `name`, `description`, `quality`, `target`, and `rewards` are optional display - metadata from `mappings/achievements.json`. -- A newly captured ID missing from the bundled mapping still exports its core - captured fields; only optional display metadata is omitted. -- Asset definitions absent from the network response are not exported because - absence cannot distinguish unstarted, locked, or unavailable content. - ## CSV diagnostics CSV exports are mainly for debugging the decoder. Tools should prefer JSON diff --git a/docs/limitations.md b/docs/limitations.md index 0054239..bfd5c13 100644 --- a/docs/limitations.md +++ b/docs/limitations.md @@ -21,13 +21,5 @@ Known limitations: - The file adapter reads mitmproxy `.flows` captures for research and testing. - Npcap is Windows-only and is not redistributed with this project; Linux and macOS use their system libpcap. - Reward keys decode to their reward id string, so unknown rewards still export a usable `reward_id`; display names/ranks come from the mapping JSON files and should be expanded as new rewards appear. -- Achievement exports contain only records observed in the login response. - Asset-only definitions are not synthesized because absence cannot distinguish - unstarted, locked, or unavailable content. Unknown captured achievement IDs - still export without optional display metadata. -- Observed achievements have only `completed` and `in_progress` states. Some - compound achievements expose numeric progress `0` while the server tracks - hidden checklist progress, so every observed record without completion ticks - is classified as `in_progress`. Privacy guardrail: raw packet data must not be included in sanitized exports. diff --git a/docs/mapping-updates.md b/docs/mapping-updates.md index 80c7ae4..a4856bc 100644 --- a/docs/mapping-updates.md +++ b/docs/mapping-updates.md @@ -1,4 +1,4 @@ -# Mapping updates +# Reward mapping updates `tools/update_mappings.py` rebuilds `arcs.json`, `characters.json`, `items.json`, and `achievements.json` directly from the current @@ -19,17 +19,15 @@ game whose asset tables evolve over time. ### GitHub Actions -Run **Update mappings** from the repository's Actions tab. **Use workflow -from** selects the exporter branch whose updater is run and the target branch -for the resulting pull request. The `source_ref` input separately selects an -NTE_Assets branch, tag, or commit and defaults to `main`. +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 for the same -selected branch updates that branch's open pull request. If the authoritative -snapshot has not changed, it reports zero changes and does not create a commit -or pull request. +`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 @@ -70,7 +68,7 @@ python -m pytest -q ``` For CI, `--check` exits with status 1 whenever the generated snapshot differs -from the committed mappings. It still writes the staged artifacts. +from the committed reward mappings. It still writes the staged artifacts. ## Source rules @@ -82,9 +80,6 @@ from the committed mappings. It still writes the staged artifacts. - 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. -- Vehicle liveries are resolved from their illustration name and icon. Rows - whose icon identifies a different reward are treated as unreleased upstream - placeholders and skipped until the assets are corrected. - Mystery Box rewards are selected from every row in `DT_GashaponLotteryGlobal`. No event IDs are hard-coded, so future rotations are included automatically. Rewards resolve through the inventory, capital, @@ -94,8 +89,8 @@ from the committed mappings. It still writes the staged artifacts. currencies, are categorized as items. - 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. +- Character-awakening and character-vehicle 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 @@ -105,7 +100,7 @@ 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 and achievement IDs or display metadata may change or disappear when -the authoritative NTE_Assets snapshot changes. Existing exports remain -unchanged; new exports use the current asset snapshot. UID stability is -independent of display mappings. +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/docs/packet-format.md b/docs/packet-format.md index ad8937f..290094b 100644 --- a/docs/packet-format.md +++ b/docs/packet-format.md @@ -1,21 +1,6 @@ # Packet Format Notes -This prototype supports separate Monopoly, Arc/Gashapon, Mystery Box history, -and achievement decoders. - -## Achievements - -The account's achievement component arrives in an inbound TCP game-data block -during login. The block is LZ4-decoded and contains `AchievementRecord` entries. -Each record provides an achievement ID, an unsigned numeric progress value, and -a .NET completion timestamp. A non-zero timestamp marks a completed record. -Every observed record without completion ticks is treated as in progress; -compound achievements can have server-side checklist progress while exposing a -numeric value of zero. - -Only records present in the network response are exported. Display metadata is -looked up case-insensitively in `mappings/achievements.json`, because captured -and asset-table ID casing can differ. Missing metadata never prevents export. +This prototype supports separate Monopoly, Arc/Gashapon, and Mystery Box history decoders. ## Decoder strategy @@ -101,8 +86,6 @@ Decoded fields: Arc/Gashapon history uses the same scheme at `ASCII * 2`. - The decoder decodes the key to its id string and looks up display metadata in `mappings/arcs.json`, `mappings/characters.json`, and `mappings/items.json`. - Lookups are case-insensitive across Monopoly, Arc, and Mystery Box decoders; - known IDs are emitted using mapping-canonical casing. Achievement display metadata is generated separately in `mappings/achievements.json`. Unknown rewards still export their decoded id with empty name/rank. diff --git a/mappings/arcs.json b/mappings/arcs.json index a83a4b4..1a23a05 100644 --- a/mappings/arcs.json +++ b/mappings/arcs.json @@ -10,10 +10,8 @@ "fork_Butterfly": {"name": "Reality Refuge", "rank": "S"}, "fork_Castle": {"name": "Call of the Twisted City", "rank": "A"}, "fork_Crowbar": {"name": "Time Bandit", "rank": "A"}, - "fork_DemonBlade": {"name": "Ravenous Blade", "rank": "S"}, "fork_Door": {"name": "The Wrong Gate", "rank": "S"}, "fork_dustbin": {"name": "Dangerous Game", "rank": "B"}, - "fork_GoldRecord": {"name": "Voice of the Voyager", "rank": "S"}, "fork_GoldWool": {"name": "What's Desired", "rank": "S"}, "fork_jiaojuan": {"name": "Shiny Days", "rank": "A"}, "fork_jingmotingyuan": {"name": "Camellia Society", "rank": "S"}, diff --git a/mappings/characters.json b/mappings/characters.json index 539c719..b037890 100644 --- a/mappings/characters.json +++ b/mappings/characters.json @@ -9,7 +9,6 @@ "1023": {"name": "Baicang", "rank": "S"}, "1025": {"name": "Hathor", "rank": "S"}, "1033": {"name": "Adler", "rank": "A"}, - "1036": {"name": "Zankou", "rank": "S"}, "1039": {"name": "Fadia", "rank": "S"}, "1046": {"name": "Zero", "rank": "S"}, "1051": {"name": "Zero", "rank": "S"}, @@ -19,9 +18,7 @@ "1056": {"name": "Lacrimosa", "rank": "S"}, "1070": {"name": "Aurelia", "rank": "A"}, "1071": {"name": "Chaos", "rank": "S"}, - "1072": {"name": "Linko", "rank": "S"}, "1073": {"name": "Chiz", "rank": "S"}, "1075": {"name": "Iroi", "rank": "S"}, - "1076": {"name": "Shinku", "rank": "S"}, - "1091": {"name": "Lacrimosa", "rank": "S"} + "1076": {"name": "Shinku", "rank": "S"} } diff --git a/mappings/items.json b/mappings/items.json index 42c8a9f..263f9ef 100644 --- a/mappings/items.json +++ b/mappings/items.json @@ -1,51 +1,40 @@ { - "BussinessCard_MangHe_bsj": {"type": "cosmetic", "name": "Trackside Shadow", "rank": "S"}, - "CharacterUpMaterial_lv3": {"type": "item", "name": "Elite Hunter Guide", "rank": "A"}, - "CityAbility_UpMaterial": {"type": "item", "name": "Dreamless Seed", "rank": "B"}, "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"}, "DiceNormal": {"type": "item", "name": "Fabricated Dice", "rank": "S"}, - "Fashion_1051_rpg_level3_Base": {"type": "cosmetic", "name": "Glory Days", "rank": "S"}, - "Fashion_1051_rpg_level3_Dye1": {"type": "cosmetic", "name": "Glory Days - Braveheart", "rank": "S"}, - "Fashion_1051_rpg_level3_Dye2": {"type": "cosmetic", "name": "Glory Days - Starfall", "rank": "S"}, - "Fashion_1075_rpg_level2_Base": {"type": "cosmetic", "name": "Archmage", "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_1010_swimsuit": {"type": "cosmetic", "name": "Sunny Surf", "rank": "S"}, - "Fashion_character_1019_swimsuit": {"type": "cosmetic", "name": "Saltwater Taffy", "rank": "S"}, - "Fashion_character_1036_03": {"type": "cosmetic", "name": "Autumn Vignette", "rank": "S"}, - "Fashion_character_1051_07": {"type": "cosmetic", "name": "Perfect Partners", "rank": "S"}, - "Fashion_character_1051_swimsuit": {"type": "cosmetic", "name": "Cloudless Summer", "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_1072_01": {"type": "cosmetic", "name": "Surfing Among Stars", "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_1036": {"type": "cosmetic", "name": "Scarlet Sash", "rank": "A"}, "Fashion_Glide_1052": {"type": "cosmetic", "name": "Orchid Breeze", "rank": "A"}, "Fashion_Glide_1071": {"type": "cosmetic", "name": "Skyrider", "rank": "A"}, - "Fashion_Glide_1072": {"type": "cosmetic", "name": "Sheepcopter", "rank": "A"}, - "Fashion_Glide_1075": {"type": "cosmetic", "name": "Sheepcopter", "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_1036_V043": {"type": "cosmetic", "name": "Sandy Mirage - 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": "Three's Company - 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"}, - "Fons": {"type": "item", "name": "Fons", "rank": "A"}, - "Frame_MangHe_bsj": {"type": "cosmetic", "name": "Heartbeat Moment", "rank": "S"}, - "Frame_MangHe_shinku": {"type": "cosmetic", "name": "Ember Trail", "rank": "S"}, - "Frame_MangHe_summer_01": {"type": "cosmetic", "name": "Endless Summer", "rank": "S"}, - "Gold": {"type": "item", "name": "Beetle Coin", "rank": "B"}, - "SpecialGift_letter": {"type": "item", "name": "A Handwritten Letter", "rank": "S"}, - "SpecialGift_ticket": {"type": "item", "name": "Floe Cinema Ticket", "rank": "A"}, - "vehicle026": {"type": "item", "name": "Porsche 918 Spyder", "rank": "S"}, "vehicle039": {"type": "item", "name": "Draco", "rank": "S"}, - "Vehicle044": {"type": "item", "name": "Tide", "rank": "S"}, - "weeklycloneboss_selectbox_02": {"type": "item", "name": "Anomaly Pilgrimage Material Selection Box", "rank": "S"} + "Fashion_1051_rpg_level3_Dye1": {"type": "cosmetic", "name": "Glory Days - Braveheart", "rank": "S"}, + "Fashion_1051_rpg_level3_Dye2": {"type": "cosmetic", "name": "Glory Days - Starfall", "rank": "S"}, + "Fashion_1051_rpg_level3_Base": {"type": "cosmetic", "name": "Glory Days", "rank": "S"}, + "SpecialGift_ticket": {"type": "item", "name": "Floe Cinema Ticket", "rank": "A"}, + "Fons": {"type": "item", "name": "Fons", "rank": "A"}, + "CharacterUpMaterial_lv3": {"type": "item", "name": "Elite Hunter Guide", "rank": "A"}, + "Gold": {"type": "item", "name": "Beetle Coin", "rank": "B"}, + "CityAbility_UpMaterial": {"type": "item", "name": "Dreamless Seed", "rank": "B"}, + "SpecialGift_letter": {"type": "item", "name": "A Handwritten Letter", "rank": "S"}, + "Frame_MangHe_shinku": {"type": "cosmetic", "name": "Ember Trail", "rank": "S"}, + "Fashion_character_1051_07": {"type": "cosmetic", "name": "Perfect Partners", "rank": "S"}, + "vehicle026": {"type": "item", "name": "Porsche 918 Spyder", "rank": "S"}, + "Frame_MangHe_bsj": {"type": "cosmetic", "name": "Heartbeat Moment", "rank": "S"}, + "BussinessCard_MangHe_bsj": {"type": "cosmetic", "name": "Trackside Shadow", "rank": "S"} } diff --git a/pyproject.toml b/pyproject.toml index 81e3d68..c5c4a4d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,7 +1,7 @@ [project] name = "nte-history-exporter" -version = "0.4.0" -description = "Cross-platform Neverness to Everness pull-history and achievement exporter." +version = "0.2.1" +description = "Cross-platform Neverness to Everness pull-history exporter." requires-python = ">=3.10" dependencies = [] diff --git a/src/nte_history_exporter/__init__.py b/src/nte_history_exporter/__init__.py index 6a9beea..3ced358 100644 --- a/src/nte_history_exporter/__init__.py +++ b/src/nte_history_exporter/__init__.py @@ -1 +1 @@ -__version__ = "0.4.0" +__version__ = "0.2.1" diff --git a/src/nte_history_exporter/console.py b/src/nte_history_exporter/console.py index 15a3dc4..3a7908c 100644 --- a/src/nte_history_exporter/console.py +++ b/src/nte_history_exporter/console.py @@ -94,13 +94,9 @@ def print_live_instructions(local_ip: str, backend: str = "windows_raw", detail: print(style(" You can capture more than one board in the same session.", DIM)) print() print(style(" Achievements", BOLD)) - print(" Your tracked achievements are captured automatically when") + print(" Your completed achievements are captured automatically when") print(" you log in after following the step above.") print() - print(style(" Gear", BOLD)) - print(" Your cartridges, modules and equipped loadouts are captured") - print(" automatically at login, in the same step as achievements.") - print() print(style(" Ready - use the game, then press any key here when finished.", BOLD, GREEN)) print(rule()) @@ -110,26 +106,12 @@ def print_page_captured(label: str, page: int | None, *, recaptured: bool = Fals print(style(" + ", GREEN, BOLD) + label + style(f" {action} {page}", DIM)) -def print_achievements_captured( - in_game_completed: int, - in_game_in_progress: int, - playstation_completed: int, - playstation_in_progress: int, -) -> None: - print(style(" + ", GREEN, BOLD) + "Achievements captured") +def print_achievements_captured(total: int, playstation: int) -> None: + visible = total - playstation print( - style( - f" In-game {in_game_completed} completed, " - f"{in_game_in_progress} in progress", - DIM, - ) - ) - print( - style( - f" PlayStation {playstation_completed} completed, " - f"{playstation_in_progress} in progress", - DIM, - ) + style(" + ", GREEN, BOLD) + + "Achievements" + + style(f" captured {visible} visible + {playstation} PlayStation", DIM) ) @@ -202,7 +184,7 @@ def print_problem(text: str) -> None: def print_update_available(current_version: str, latest_version: str, release_url: str) -> None: print() print(style(f" Update available: v{current_version} -> {latest_version}", YELLOW, BOLD)) - print(style(" Some reward or achievement mappings may be incomplete in this version.", YELLOW)) + print(style(" Some banner/reward mappings may be incomplete in this version.", YELLOW)) print(style(" Updating is recommended for the most accurate export labels.", YELLOW)) print(style(f" Download when ready: {release_url}", YELLOW)) diff --git a/src/nte_history_exporter/constants.py b/src/nte_history_exporter/constants.py index 6fe8b4e..883f920 100644 --- a/src/nte_history_exporter/constants.py +++ b/src/nte_history_exporter/constants.py @@ -13,7 +13,7 @@ ARC_BANNER_NAME = "Arc Miracle Box" MYSTERY_BOX_BANNER_ID = "Gashapon_MysteryBox" MYSTERY_BOX_BANNER_NAME = "Mystery Box" EXPORTER_NAME = "nte-history-exporter" -EXPORTER_VERSION = "0.4.0" +EXPORTER_VERSION = "0.2.1" HISTORY_REQUEST_BANNER = 4220 HISTORY_REQUEST_LENGTH = 45 diff --git a/src/nte_history_exporter/decoder/achievement.py b/src/nte_history_exporter/decoder/achievement.py index 0c3b33e..44cd5e9 100644 --- a/src/nte_history_exporter/decoder/achievement.py +++ b/src/nte_history_exporter/decoder/achievement.py @@ -27,10 +27,7 @@ class AchievementRecord: def status(self) -> str: if self.completed: return "completed" - # Presence in the server response means the achievement has tracked - # state. Compound achievements can have hidden checklist progress while - # exposing a numeric progress value of zero to the client. - return "in_progress" + return "in_progress" if self.progress else "not_started" @property def completed_at(self) -> str | None: diff --git a/src/nte_history_exporter/decoder/arc.py b/src/nte_history_exporter/decoder/arc.py index 48d498e..bc5ed10 100644 --- a/src/nte_history_exporter/decoder/arc.py +++ b/src/nte_history_exporter/decoder/arc.py @@ -21,7 +21,7 @@ from nte_history_exporter.constants import ( ) from nte_history_exporter.decoder.boundary import select_continuous_run_from_page_1 from nte_history_exporter.decoder.run import fmt_packet_time -from nte_history_exporter.mappings import ARC_META_BY_CASEFOLD +from nte_history_exporter.mappings import ARC_META from nte_history_exporter.decoder.structured_protocol import ( StructuredProtocolAssembler, StructuredRecord, @@ -95,7 +95,7 @@ def _parse_legacy_arc_response(response: bytes) -> list[dict[str, Any]]: timestamp_raw = response[pos : pos + 8] pos += 8 arc_id = decode_arc_key(name_raw) or name_raw.hex() - canonical_id, meta = _resolve_arc_metadata(arc_id) + meta = ARC_META.get(arc_id, {}) try: ticks, unix_seconds, timestamp_decoded = decode_arc_timestamp(timestamp_raw) except ValueError: @@ -107,7 +107,7 @@ def _parse_legacy_arc_response(response: bytes) -> list[dict[str, Any]]: "record_len": pos - start, "reward_key_hex": name_raw.hex(), "reward_type": "arc", - "reward_id": canonical_id, + "reward_id": arc_id, "reward_name": meta.get("name", "UNKNOWN"), "reward_rank": meta.get("rank", ""), "type_key_hex": type_raw.hex(), @@ -122,15 +122,18 @@ def _parse_legacy_arc_response(response: bytes) -> list[dict[str, Any]]: return records -def _resolve_arc_metadata(arc_id: str) -> tuple[str, dict[str, Any]]: - canonical_id, meta = ARC_META_BY_CASEFOLD.get(arc_id.casefold(), (arc_id, {})) - return canonical_id, meta +def _arc_metadata(arc_id: str) -> dict[str, Any]: + direct = ARC_META.get(arc_id) + if direct is not None: + return direct + folded = arc_id.casefold() + return next((meta for item_id, meta in ARC_META.items() if item_id.casefold() == folded), {}) def structured_arc_rows(structured_rows: list[StructuredRecord]) -> list[dict[str, Any]]: records = [] for structured in structured_rows: - canonical_id, meta = _resolve_arc_metadata(structured.item_id) + meta = _arc_metadata(structured.item_id) records.append( { "record_start": structured.record_start, @@ -138,7 +141,7 @@ def structured_arc_rows(structured_rows: list[StructuredRecord]) -> list[dict[st "record_len": structured.record_end - structured.record_start, "reward_key_hex": "", "reward_type": "arc", - "reward_id": canonical_id, + "reward_id": structured.item_id, "reward_name": meta.get("name", "UNKNOWN"), "reward_rank": meta.get("rank", ""), "type_key_hex": "", diff --git a/src/nte_history_exporter/decoder/gear.py b/src/nte_history_exporter/decoder/gear.py deleted file mode 100644 index 87a1ef6..0000000 --- a/src/nte_history_exporter/decoder/gear.py +++ /dev/null @@ -1,499 +0,0 @@ -"""Decode cartridges, modules and character loadouts from the gameplay UDP flow. - -The game speaks Unreal's bit-packed net protocol. Each datagram is a bit stream: -a 15-byte packet and bunch header, the payload, then two 1 bits - the packet -handler's terminator and the bit writer's - zero-padded to the next byte -boundary. Records routinely straddle datagram boundaries, so the payload bit -ranges have to be spliced; concatenating whole datagrams splices the framing into -the middle of a record and corrupts it. - -Inside the spliced stream, an item record reads: - - length == len(id) + 1 (self-validating frame check) - \0 - <5 bytes> <8-byte instance id> ... - ... <12-byte owner reference at +312 bits, all zero when unequipped> ... - then N stat entries, each: - \0 <4 zero bytes> - -Consecutive entries drift one bit apart, so entries are chained forward from the -previous one rather than found by global search - that way a neighbouring -record's entries can never be stolen. - -A character block is marked by its `GA__Melee|Skill|UltraSkill|QTE` ability -strings, with the character's level and breakthrough count at fixed bit offsets -before it, and its equipped modules listed afterwards one entry per occupied -grid cell. - -Every failure path returns empty rather than raising. -""" - -from __future__ import annotations - -import re -import struct -from dataclasses import dataclass, field - -# --- framing --------------------------------------------------------------- - -HEADER_BITS = 120 -# Datagrams with no room for a payload are pure acks and carry nothing. -MIN_PAYLOAD_BITS = 32 - -# --- record layout --------------------------------------------------------- - -LEVEL_OFFSET_BYTES = 38 -MAX_ITEM_LEVEL = 20 -OWNER_OFFSET_BITS = 312 -OWNER_BYTES = 12 -# An item packs its entries tightly; anything further away belongs to something -# else, so a wider scan would wander into unrelated data. -CHAIN_SCAN_BITS = 1400 -# Observed entry spacing is (len+12) bytes plus a 1-bit separator, tightest at -# ~145 bits. Advancing by a floor below that and rescanning avoids overshooting -# entries that are shorter than their computed width - main stats run two bytes -# shorter than substats. -MIN_ENTRY_SPACING_BITS = 100 -MODULE_ENTRIES = 6 -CARTRIDGE_ENTRIES = 5 - -# --- character layout ------------------------------------------------------ - -# Bit offsets relative to the start of the character's first ability frame. -CHARACTER_MARKER_OFFSET_BITS = -466 -CHARACTER_LEVEL_OFFSET_BITS = -434 -CHARACTER_BREAKTHROUGH_OFFSET_BITS = -402 -CHARACTER_MARKER = 67806 -MAX_CHARACTER_LEVEL = 90 -MAX_BREAKTHROUGHS = 10 -# A loadout cell entry stores doubled grid coordinates after the item id. -CELL_ROW_OFFSET_BITS = 96 -CELL_COL_OFFSET_BITS = 128 - -# --- vocabulary ------------------------------------------------------------ - -# Ids are whitelisted: the length prefix validates a frame but cannot catch a -# single flipped bit inside the name - `Ingantation` and `GetEfviciency` both -# passed the length check before the whitelist went in. -ITEM_ID = re.compile( - rb"(cell[234]_style[1-6]_\d_(?:Orange|Purple|Blue)" - # Psychically must precede Psyche: both ids exist and name different sets. - rb"|(?:Attack|Chaos|Cosmos|GetEfficiency|Heal|Incantation|Lakshana|Mag|Nature" - rb"|Psychically|Psyche|Shield)_(?:orange|purple|blue))\x00" -) -ABILITY_ID = re.compile(rb"GA_([A-Za-z0-9]+)_(?:Melee|Skill|UltraSkill|QTE)\x00") - -STAT_NAMES = frozenset({ - "AtkAdd", "AtkUp", "DefAdd", "DefUp", "HPMaxAdd", "HPMaxUp", "CritBase", - "CritDamageBase", "MagBase", "UnbalIntensityBase", "HealUp", -}) - -# Substat values are fixed per stat and item size, with no roll variance, so any -# value off this table means the parse went wrong. Main stats read 0.0. -CANONICAL_VALUES = frozenset({ - 0.0, 200.0, 300.0, 400.0, 1000.0, 16.0, 24.0, 32.0, 80.0, 48.0, 64.0, 600.0, 800.0, - 12.0, 18.0, 60.0, 36.0, - 0.025, 0.0375, 0.05, 0.125, 0.035, 0.0525, 0.07, 0.075, 0.175, 0.02, 0.03, 0.04, - 0.1, 0.06, 0.08, 0.2, 0.105, 0.14, -}) - -# A main stat's displayed value is never transmitted: it is base * (1 + level/5), -# so +0 shows the base and +20 shows five times it. -CARTRIDGE_MAIN_BASE = { - "HPMaxUp": 0.075, "AtkUp": 0.075, "DefUp": 0.105, "CritBase": 0.06, - "CritDamageBase": 0.12, "MagBase": 36.0, "UnbalIntensityBase": 36.0, - "DamageUpGeneralBase": 0.075, - # Healing Bonus reads 6.90% at +0 in game - the one main that is not round. - "HealUp": 0.069, -} -# Module mains scale per cell: 56 HP and 4.2 ATK each at +0. Measured in game as -# Type III 168/12.6 -> 840/63 and Type IV 224/16.8 -> 1120/84. The client floors -# flat values, which is why ATK displays as 12 and 16 at +0. -MODULE_HP_PER_CELL = 56.0 -MODULE_ATK_PER_CELL = 4.2 - -# Confirmed against everness's cartridge boxes, which name all twelve sets. The -# three the capture has never carried are Psyche, Shield and Heal - NOT Blood, -# Night and Kingdom, which were a guess and would have decoded as nothing. -SET_NAMES = { - "Attack": "Shadow Creed", "Chaos": "Diabolos", "Cosmos": "Lost Radiance", - "GetEfficiency": "Speedy Hedgehog", "Incantation": "Crimson: Twin Butterflies", - "Lakshana": "Street Boxer", "Nature": "Fireflies and the Forest", - "Psychically": "Quiet Manor", "Mag": "Tiny Big Adventure", - "Psyche": "Devil's Blood: Curse", "Heal": "Thea's Night Tavern", - "Shield": "Kingdom's Guard", -} - -MODULE_TYPE_NAMES = {2: "II", 3: "III", 4: "IV"} - - -@dataclass(frozen=True) -class GearItem: - instance: str - kind: str - item_id: str - level: int - rarity: str - main_stats: tuple[tuple[str, float | None], ...] - substats: tuple[tuple[str, float], ...] - owner_group: str | None - shape: str | None = None - module_type: str | None = None - set_name: str | None = None - - -@dataclass(frozen=True) -class GearCharacter: - key: str - level: int | None - breakthroughs: int | None - board: tuple[tuple[int, int], ...] = () - - -@dataclass(frozen=True) -class GearSnapshot: - items: tuple[GearItem, ...] = () - characters: tuple[GearCharacter, ...] = () - warnings: tuple[str, ...] = field(default=()) - - @property - def cartridges(self) -> int: - return sum(item.kind == "cartridge" for item in self.items) - - @property - def modules(self) -> int: - return sum(item.kind == "module" for item in self.items) - - @property - def characters_with_loadouts(self) -> int: - return sum(bool(character.board) for character in self.characters) - - -def is_stat_name(name: str) -> bool: - return name in STAT_NAMES or (name.startswith("DamageUp") and name.endswith("Base")) - - -def bitshift(buf: bytes, shift: int) -> bytes: - if shift == 0: - return buf - return bytes( - ((buf[i] >> shift) | (buf[i + 1] << (8 - shift))) & 0xFF - for i in range(len(buf) - 1) - ) - - -class _BitAccumulator: - """Append arbitrary bit ranges; the result is a byte buffer, LSB-first.""" - - def __init__(self) -> None: - self.buf = bytearray() - self.nbits = 0 - - def append(self, data: bytes, start: int, nbits: int) -> None: - if nbits <= 0: - return - value = (int.from_bytes(data, "little") >> start) & ((1 << nbits) - 1) - offset = self.nbits % 8 - if offset: - value <<= offset - raw = value.to_bytes((offset + nbits + 7) // 8, "little") - self.buf[-1] |= raw[0] - self.buf.extend(raw[1:]) - else: - self.buf.extend(value.to_bytes((nbits + 7) // 8, "little")) - self.nbits += nbits - del self.buf[(self.nbits + 7) // 8 :] - - -def payload_end_bit(datagram: bytes) -> int: - """First bit past the payload. - - Unreal ends a packet with two 1 bits then zero-pads to a byte boundary, so - the payload stops one bit below the highest set bit of the last non-zero - byte. - """ - - i = len(datagram) - 1 - while i >= 0 and datagram[i] == 0: - i -= 1 - if i < 0: - return 0 - return i * 8 + datagram[i].bit_length() - 2 - - -def splice_datagrams(datagrams) -> bytes: - """Concatenate every datagram's payload bits with the framing removed.""" - - accumulator = _BitAccumulator() - for datagram in datagrams: - end = payload_end_bit(datagram) - if end - HEADER_BITS >= MIN_PAYLOAD_BITS: - accumulator.append(datagram, HEADER_BITS, end - HEADER_BITS) - return bytes(accumulator.buf) - - -class _BitView: - """Random access to a buffer at arbitrary bit offsets.""" - - def __init__(self, buf: bytes) -> None: - self.views = [bitshift(buf, shift) for shift in range(8)] - - def at(self, bitpos: int) -> tuple[bytes, int]: - return self.views[bitpos % 8], bitpos // 8 - - def u32(self, bitpos: int) -> int | None: - view, offset = self.at(bitpos) - if offset + 4 > len(view): - return None - return struct.unpack_from(" float | None: - view, offset = self.at(bitpos) - if offset + 4 > len(view): - return None - return struct.unpack_from(" bytes: - view, offset = self.at(bitpos) - return view[offset : offset + length] - - def text(self, bitpos: int, length: int) -> str | None: - view, offset = self.at(bitpos) - if offset + length > len(view): - return None - raw = view[offset : offset + length] - if not raw.endswith(b"\0"): - return None - try: - text = raw[:-1].decode("ascii") - except UnicodeDecodeError: - return None - return text if text.isprintable() else None - - -def _validated_frames(view: _BitView, pattern): - """Every self-validating frame, in bit order.""" - - found = [] - for shift in range(8): - haystack = view.views[shift] - for match in pattern.finditer(haystack): - body = match.group() - if match.start() < 4: - continue - if struct.unpack_from(" tuple[str, float] | None: - """Read one length-prefixed stat entry.""" - - declared = view.u32(bitpos) - if declared is None or not 4 <= declared <= 64: - return None - name = view.text(bitpos + 32, declared) - if name is None: - return None - # Layout: <4 zero bytes> - value = view.f32(bitpos + 32 + declared * 8 + 32) - if value is None: - return None - return name, round(value, 6) - - -def _chain_entries(view: _BitView, start_bit: int, needed: int): - """Walk `needed` stat entries forward from just past the item id.""" - - cursor, entries = start_bit, [] - while len(entries) < needed: - found = None - for delta in range(CHAIN_SCAN_BITS): - probe = read_entry(view, cursor + delta) - if probe and is_stat_name(probe[0]): - found = (probe, cursor + delta) - break - if not found: - return entries - entries.append(found[0]) - cursor = found[1] + MIN_ENTRY_SPACING_BITS - return entries - - -def _main_value(kind: str, stat: str, level: int, cells: int | None) -> float | None: - if kind == "module": - base = { - "HPMaxAdd": MODULE_HP_PER_CELL * (cells or 0), - "AtkAdd": MODULE_ATK_PER_CELL * (cells or 0), - }.get(stat) - else: - base = CARTRIDGE_MAIN_BASE.get(stat) - # Every elemental damage bonus shares the general bonus's base. - if base is None and stat.startswith("DamageUp") and stat.endswith("Base"): - base = CARTRIDGE_MAIN_BASE["DamageUpGeneralBase"] - if base is None: - return None - value = base * (1 + level / 5) - # Flat module stats display floored; percentages keep their precision. - return float(int(value)) if kind == "module" else round(value, 4) - - -def _item_from_frame(view: _BitView, frame) -> GearItem | None: - bitpos, match, shift, id_end_bit = frame - item_id = match.group()[:-1].decode() - is_module = item_id.startswith("cell") - needed = MODULE_ENTRIES if is_module else CARTRIDGE_ENTRIES - main_count = 2 if is_module else 1 - - entries = _chain_entries(view, id_end_bit, needed) - if len(entries) != needed: - return None - # Substats never vary, so an off-table value means the chain went astray. - if not all(value in CANONICAL_VALUES for _, value in entries): - return None - # In every record confirmed against in-game values the main stats come - # first. A main appearing later means the chain wrapped into a neighbour. - if not all(value == 0.0 for _, value in entries[:main_count]): - return None - if sum(value == 0.0 for _, value in entries) != main_count: - return None - - id_end_byte = match.end() - haystack = view.views[shift] - instance = haystack[id_end_byte + 5 : id_end_byte + 13] - if len(instance) != 8: - return None - - level_offset = id_end_byte + LEVEL_OFFSET_BYTES - if level_offset + 4 > len(haystack): - return None - level = struct.unpack_from(" MAX_ITEM_LEVEL: - level = 0 - - owner_bit = id_end_bit + OWNER_OFFSET_BITS - # Gate on the leading word: trailing bytes of the field carry unrelated data - # on some records, so testing the whole span over-reports. - owner = view.raw(owner_bit, OWNER_BYTES) if view.u32(owner_bit) else b"" - - cells = int(item_id[4]) if is_module else None - mains = tuple( - (name, _main_value("module" if is_module else "cartridge", name, level, cells)) - for name, value in entries if value == 0.0 - ) - substats = tuple((name, value) for name, value in entries if value != 0.0) - rarity = item_id.rsplit("_", 1)[1].lower() - - if is_module: - shape = "_".join(item_id.split("_")[:2]) - return GearItem( - instance=instance.hex(), kind="module", item_id=item_id, level=level, - rarity=rarity, main_stats=mains, substats=substats, - owner_group=owner.hex() or None, - shape=shape, module_type=MODULE_TYPE_NAMES.get(cells or 0), - ) - return GearItem( - instance=instance.hex(), kind="cartridge", item_id=item_id, level=level, - rarity=rarity, main_stats=mains, substats=substats, - owner_group=owner.hex() or None, - set_name=SET_NAMES.get(item_id.rsplit("_", 1)[0]), - ) - - -def _character_blocks(view: _BitView) -> list[tuple[int, str]]: - """The first ability frame of each character, in stream order.""" - - blocks: list[tuple[int, str]] = [] - for bitpos, match, _shift, _end in _validated_frames(view, ABILITY_ID): - name = match.group(1).decode() - if not blocks or blocks[-1][1] != name: - blocks.append((bitpos, name)) - return blocks - - -def _character_progress(view: _BitView, block_bit: int) -> tuple[int | None, int | None]: - """Level and breakthrough count, validated against the block's marker.""" - - if view.u32(block_bit + CHARACTER_MARKER_OFFSET_BITS) != CHARACTER_MARKER: - return None, None - level = view.u32(block_bit + CHARACTER_LEVEL_OFFSET_BITS) - breakthroughs = view.u32(block_bit + CHARACTER_BREAKTHROUGH_OFFSET_BITS) - if level is None or not 1 <= level <= MAX_CHARACTER_LEVEL: - level = None - if breakthroughs is None or breakthroughs > MAX_BREAKTHROUGHS: - breakthroughs = None - return level, breakthroughs - - -def extract_gear(datagrams) -> GearSnapshot: - """Decode every item and character loadout in a captured UDP flow.""" - - try: - buf = splice_datagrams(datagrams) - except (ValueError, MemoryError): - return GearSnapshot() - if len(buf) < 64: - return GearSnapshot() - - view = _BitView(buf) - frames = _validated_frames(view, ITEM_ID) - blocks = _character_blocks(view) - - items: dict[str, GearItem] = {} - # An id frame that chains no stat entries is a loadout reference, not a - # record: the same id quoted inside a character's equipped list. - loadouts: dict[str, list[tuple[int, int, int]]] = {} - for frame in frames: - item = _item_from_frame(view, frame) - if item is not None: - items.setdefault(item.instance, item) - continue - bitpos, _match, _shift, id_end_bit = frame - row = view.u32(id_end_bit + CELL_ROW_OFFSET_BITS) - column = view.u32(id_end_bit + CELL_COL_OFFSET_BITS) - if not row or not column: - continue - owner_index = -1 - for index, (block_bit, _name) in enumerate(blocks): - if block_bit > bitpos: - break - owner_index = index - if owner_index >= 0: - # Coordinates are doubled; a cell is one square, not four. - loadouts.setdefault(blocks[owner_index][1], []).append( - (bitpos, row // 2, column // 2) - ) - - # A character's block is transmitted more than once in a login - the active - # team appears early and again in the full roster - so merge by name and - # keep the first reading that validated. - merged: dict[str, GearCharacter] = {} - for block_bit, name in blocks: - level, breakthroughs = _character_progress(view, block_bit) - board = tuple(sorted({(r, c) for _bit, r, c in loadouts.get(name, [])})) - previous = merged.get(name) - if previous is None: - merged[name] = GearCharacter(name, level, breakthroughs, board) - continue - merged[name] = GearCharacter( - key=name, - level=previous.level if previous.level is not None else level, - breakthroughs=( - previous.breakthroughs - if previous.breakthroughs is not None - else breakthroughs - ), - board=previous.board or board, - ) - - return GearSnapshot( - items=tuple(items.values()), - characters=tuple(merged.values()), - ) diff --git a/src/nte_history_exporter/decoder/mystery_box.py b/src/nte_history_exporter/decoder/mystery_box.py index 09088af..2c9eaf2 100644 --- a/src/nte_history_exporter/decoder/mystery_box.py +++ b/src/nte_history_exporter/decoder/mystery_box.py @@ -24,11 +24,16 @@ from nte_history_exporter.constants import ( from nte_history_exporter.decoder.boundary import select_continuous_run_from_page_1 from nte_history_exporter.decoder.protocol import infer_reward_type from nte_history_exporter.decoder.run import fmt_packet_time -from nte_history_exporter.mappings import REWARDS_BY_CASEFOLD +from nte_history_exporter.mappings import REWARDS_BY_ID DOTNET_TICKS_PER_SECOND = 10_000_000 MAX_RECORDS_PER_BLOCK = 100 MAX_REWARD_ID_LENGTH = 256 +REWARDS_BY_CASEFOLDED_ID = { + reward_id.casefold(): reward for reward_id, reward in REWARDS_BY_ID.items() +} + + def is_mystery_box_history_request(content: bytes) -> bool: if len(content) < MYSTERY_BOX_HISTORY_REQUEST_LENGTH: return False @@ -103,15 +108,16 @@ def _parse_view(data: bytes) -> list[dict[str, Any]]: timestamp_raw = data[pos : pos + 8] pos += 8 ticks, unix_seconds, timestamp_decoded = _decode_timestamp(timestamp_raw) - reward = REWARDS_BY_CASEFOLD.get(reward_id.casefold(), {}) - canonical_reward_id = reward.get("id", reward_id) + reward = REWARDS_BY_ID.get(reward_id) or REWARDS_BY_CASEFOLDED_ID.get( + reward_id.casefold(), {} + ) rows.append( { "record_start": record_start, "record_end": pos, "record_len": pos - record_start, "reward_type": reward.get("type") or infer_reward_type(reward_id), - "reward_id": canonical_reward_id, + "reward_id": reward_id, "reward_name": reward.get("name", ""), "reward_rank": reward.get("rank"), "quantity": quantity, diff --git a/src/nte_history_exporter/decoder/protocol.py b/src/nte_history_exporter/decoder/protocol.py index 17c7267..3fed6c7 100644 --- a/src/nte_history_exporter/decoder/protocol.py +++ b/src/nte_history_exporter/decoder/protocol.py @@ -15,7 +15,7 @@ from nte_history_exporter.constants import ( TIMESTAMP_TICKS_PER_SECOND, VALID_DICE_FIELDS, ) -from nte_history_exporter.mappings import REWARDS_BY_CASEFOLD +from nte_history_exporter.mappings import REWARDS_BY_ID from nte_history_exporter.decoder.structured_protocol import ( StructuredRecord, parse_structured_records, @@ -61,12 +61,11 @@ def decode_reward_key(raw: bytes) -> str: def infer_reward_type(reward_id: str) -> str: if not reward_id: return "" - folded = reward_id.casefold() - if folded.startswith("fork_"): + if reward_id.startswith("fork_"): return "arc" if reward_id.isdigit(): return "character" - if folded.startswith("fashion_"): + if reward_id.startswith("Fashion_"): return "cosmetic" return "item" @@ -166,7 +165,7 @@ def classify_result_type( ) -> tuple[str, int | None]: if dice is None or dice_offset is None: return "unknown", None - if reward_id.casefold() == "dice_ticket_01" and WARP_PIECE_CHASE_PATTERN in chunk_without_marker: + if reward_id == "Dice_ticket_01" and WARP_PIECE_CHASE_PATTERN in chunk_without_marker: return "chase_reward", -4 if dice == 0: return "points_gift", 0 @@ -183,14 +182,13 @@ def classify_result_type( def guess_quantity(chunk_hex: str, reward_id: str, result_type: str | None = None) -> int | None: - folded = reward_id.casefold() - if folded == "dice_ticket_01": + if reward_id == "Dice_ticket_01": if result_type == "chase_reward": return 30 return 4 - if folded == "dicenormal": + if reward_id == "DiceNormal": return 1 - if folded == "dice_ticket_02": + if reward_id == "Dice_ticket_02": if "c8b0d4c0" in chunk_hex: return 50 if "c8b0ccc0" in chunk_hex: @@ -268,8 +266,7 @@ def _decode_aligned_response_records(response_content: bytes) -> list[dict[str, elif result_type == "chase_reward": dice = -4 dice_raw = -4 - reward = _reward_metadata(reward_id) - canonical_reward_id = reward.get("id", reward_id) + reward = REWARDS_BY_ID.get(reward_id, {}) timestamp_raw = response_content[marker_offset + len(marker) : marker_offset + len(marker) + 8] timestamp_ticks, timestamp_unix, timestamp_decoded = decode_history_timestamp(timestamp_raw) chunk_hex = chunk.hex() @@ -291,7 +288,7 @@ def _decode_aligned_response_records(response_content: bytes) -> list[dict[str, "dice_offset_in_record": dice_offset, "reward_key_hex": key_hex, "reward_type": reward.get("type") or infer_reward_type(reward_id), - "reward_id": canonical_reward_id, + "reward_id": reward_id, "reward_name": reward.get("name", ""), "reward_rank": reward.get("rank"), "quantity": guess_quantity(chunk_hex, reward_id, result_type), @@ -338,8 +335,8 @@ def _enrich_heuristic_rows( return heuristic_rows for heuristic, structured in zip(heuristic_rows, structured_rows): if not heuristic.get("reward_id"): + heuristic["reward_id"] = structured.item_id reward = _reward_metadata(structured.item_id) - heuristic["reward_id"] = reward.get("id", structured.item_id) heuristic["reward_type"] = reward.get("type") or infer_reward_type(structured.item_id) heuristic["reward_name"] = reward.get("name", "") heuristic["reward_rank"] = reward.get("rank") @@ -364,7 +361,11 @@ def _enrich_heuristic_rows( def _reward_metadata(reward_id: str) -> dict[str, Any]: - return REWARDS_BY_CASEFOLD.get(reward_id.casefold(), {}) + direct = REWARDS_BY_ID.get(reward_id) + if direct is not None: + return direct + folded = reward_id.casefold() + return next((meta for item_id, meta in REWARDS_BY_ID.items() if item_id.casefold() == folded), {}) def structured_monopoly_rows(structured_rows: list[StructuredRecord]) -> list[dict[str, Any]]: @@ -372,7 +373,6 @@ def structured_monopoly_rows(structured_rows: list[StructuredRecord]) -> list[di for row_index, structured in enumerate(structured_rows, start=1): dice, dice_raw, result_type, result_source = _structured_result(structured.roll_points_raw) reward = _reward_metadata(structured.item_id) - canonical_reward_id = reward.get("id", structured.item_id) rows.append( { "row": row_index, @@ -391,7 +391,7 @@ def structured_monopoly_rows(structured_rows: list[StructuredRecord]) -> list[di "dice_offset_in_record": None, "reward_key_hex": "", "reward_type": reward.get("type") or infer_reward_type(structured.item_id), - "reward_id": canonical_reward_id, + "reward_id": structured.item_id, "reward_name": reward.get("name", ""), "reward_rank": reward.get("rank"), "quantity": structured.count, diff --git a/src/nte_history_exporter/export/gear_export.py b/src/nte_history_exporter/export/gear_export.py deleted file mode 100644 index 26bd382..0000000 --- a/src/nte_history_exporter/export/gear_export.py +++ /dev/null @@ -1,99 +0,0 @@ -"""Gear export envelope. - -Kept out of ``json_export`` so the gear feature is a set of added files rather -than edits to upstream ones - this repo tracks Golumpa's exporter and has to be -able to merge it. -""" - -from __future__ import annotations - -from typing import Any - -from nte_history_exporter import __version__ -from nte_history_exporter.constants import EXPORTER_NAME, GAME_NAME -from nte_history_exporter.decoder.gear import GearCharacter, GearItem, GearSnapshot -from nte_history_exporter.decoder.server_region import account_region_for_server - - -def build_gear_export_json( - snapshot: GearSnapshot, - *, - source: str = "packet_capture", - capture_source: str | None = None, - user_uid: str | None = None, - server_id: str | None = None, -) -> dict[str, Any]: - scan = { - "cartridges": snapshot.cartridges, - "modules": snapshot.modules, - "characters": len(snapshot.characters), - "characters_with_loadouts": snapshot.characters_with_loadouts, - "warnings": list(snapshot.warnings), - } - - export: dict[str, Any] = { - "format": "nte-gear-export", - "format_version": 1, - "game": GAME_NAME, - "source": source, - } - if capture_source: - export["capture_source"] = capture_source - export["exporter"] = {"name": EXPORTER_NAME, "version": __version__} - export["scan"] = scan - - normalized_user_uid = user_uid.strip() if user_uid else "" - normalized_server_id = str(server_id).strip() if server_id else "" - if normalized_user_uid: - export["user_uid"] = normalized_user_uid - if normalized_server_id: - export["server_id"] = normalized_server_id - account_region = account_region_for_server(normalized_server_id) - if account_region: - export["account_region"] = account_region - - export["characters"] = [ - _gear_character_for_export(character) for character in snapshot.characters - ] - export["items"] = [_gear_item_for_export(item) for item in snapshot.items] - return export - - -def _gear_character_for_export(character: GearCharacter) -> dict[str, Any]: - record: dict[str, Any] = { - "key": character.key, - "level": character.level, - "breakthroughs": character.breakthroughs, - } - # Only emitted when a loadout was actually seen; an empty board and an - # unequipped character are different states and must not collapse. - if character.board: - record["board"] = [[row, column] for row, column in character.board] - # The capture never ties an owner group to a character name, so the app - # resolves this itself rather than being handed a guess. - record["owner_group"] = None - return record - - -def _gear_item_for_export(item: GearItem) -> dict[str, Any]: - record: dict[str, Any] = { - "instance": item.instance, - "kind": item.kind, - "item_id": item.item_id, - } - if item.shape: - record["shape"] = item.shape - if item.module_type: - record["module_type"] = item.module_type - if item.set_name: - record["set"] = item.set_name - record["level"] = item.level - record["rarity"] = item.rarity - record["main_stats"] = [ - {"stat": stat, "value": value} for stat, value in item.main_stats - ] - record["substats"] = [ - {"stat": stat, "value": value} for stat, value in item.substats - ] - record["owner_group"] = item.owner_group - return record diff --git a/src/nte_history_exporter/export/json_export.py b/src/nte_history_exporter/export/json_export.py index 4c500de..9915791 100644 --- a/src/nte_history_exporter/export/json_export.py +++ b/src/nte_history_exporter/export/json_export.py @@ -94,11 +94,13 @@ def build_achievement_export_json( "total_achievements": 0, "completed_achievements": 0, "in_progress_achievements": 0, + "not_started_achievements": 0, }, "playstation": { "total_achievements": 0, "completed_achievements": 0, "in_progress_achievements": 0, + "not_started_achievements": 0, }, } for achievement in achievement_records: @@ -108,8 +110,10 @@ def build_achievement_export_json( summary["total_achievements"] += 1 if achievement.completed: summary["completed_achievements"] += 1 - else: + elif achievement.progress: summary["in_progress_achievements"] += 1 + else: + summary["not_started_achievements"] += 1 category, separator, _number = achievement_id.partition("_") category_key = category.casefold() if separator else "uncategorized" record = { diff --git a/src/nte_history_exporter/live_capture/gear_collector.py b/src/nte_history_exporter/live_capture/gear_collector.py deleted file mode 100644 index bc9bae0..0000000 --- a/src/nte_history_exporter/live_capture/gear_collector.py +++ /dev/null @@ -1,191 +0,0 @@ -"""Gear capture, held in one object so ``runner`` only has to call it. - -This repo tracks Golumpa's exporter, so the gear feature is deliberately shaped -as added files plus a handful of one-line calls in ``runner.run_live_capture`` -rather than as edits scattered through the capture loop. Everything gear needs - -flow accumulation, the mid-capture decode, the fallback scan, the console line -and the export file - lives here. -""" - -from __future__ import annotations - -import json -import time -from datetime import datetime -from pathlib import Path - -from nte_history_exporter.console import BOLD, DIM, GREEN, style -from nte_history_exporter.decoder.gear import GearSnapshot, extract_gear -from nte_history_exporter.export.gear_export import build_gear_export_json - - -Flow = tuple[str, int, str, int] - -# The login gear burst is roughly 500 KB; the cap only exists so a long -# session on a busy flow cannot grow without bound. -MAX_UDP_FLOW_BYTES = 8 * 1024 * 1024 -# Decoding is linear in flow size, so only the few busiest are worth trying. -MAX_GEAR_FLOW_ATTEMPTS = 4 -# A gear burst arrives in a rush and then stops, so "this flow has gone quiet" -# is the only mid-stream marker there is. Attempts are gated on the flow being -# large enough to hold a burst, having been idle, and having grown since it was -# last tried, so a quiet flow is never decoded twice for the same bytes. -GEAR_LIVE_MIN_FLOW_BYTES = 64 * 1024 -GEAR_LIVE_IDLE_SECONDS = 1.0 -GEAR_LIVE_MAX_ATTEMPTS = 8 - - -def print_gear_captured( - cartridges: int, - modules: int, - characters: int, - characters_with_loadouts: int, -) -> None: - print(style(" + ", GREEN, BOLD) + "Gear captured") - print( - style( - f" Inventory {cartridges} cartridges, {modules} modules", - DIM, - ) - ) - print( - style( - f" Characters {characters} seen, " - f"{characters_with_loadouts} with loadouts", - DIM, - ) - ) - - -def gear_path(user_uid: str | None = None) -> Path: - export_dir = Path("exports") - export_dir.mkdir(parents=True, exist_ok=True) - uid_prefix = _safe_filename_part(user_uid) if user_uid else "unknown" - base = export_dir / f"{uid_prefix}_Gear_{datetime.now():%Y%m%d_%H%M%S}" - path = base.with_suffix(".json") - counter = 2 - while path.exists(): - path = export_dir / f"{base.name}_{counter}.json" - counter += 1 - return path - - -def _safe_filename_part(value: str | None) -> str: - # Deliberately a copy of runner's helper rather than an import: importing - # runner from here would be circular, and the rule is four lines long. - if not value: - return "unknown" - cleaned = "".join(ch for ch in value.strip() if ch.isalnum() or ch in ("-", "_")) - return cleaned or "unknown" - - -class GearCollector: - """Accumulates inbound UDP and decodes gear out of it. - - Call ``poll`` once per capture iteration (above any ``packet is None`` - guard - the pcap read timeout is what keeps it firing after the burst goes - quiet), ``observe`` for each packet, ``finalize`` after the loop, and - ``write`` to emit the export. - """ - - def __init__(self, local_ip: str | None) -> None: - self._local_ip = local_ip - self._flows: dict[Flow, list[bytes]] = {} - self._flow_bytes: dict[Flow, int] = {} - self._flow_last_seen: dict[Flow, float] = {} - self._attempted_bytes: dict[Flow, int] = {} - self._attempts = 0 - self.snapshot: GearSnapshot | None = None - - def observe(self, packet) -> None: - """Keep an inbound UDP payload whole. - - Datagram boundaries carry the framing, so payloads are spliced at - decode time and never concatenated here. - """ - if ( - packet is None - or packet.protocol != "udp" - or not packet.payload - or packet.dst_ip != self._local_ip - ): - return - flow: Flow = (packet.src_ip, packet.src_port, packet.dst_ip, packet.dst_port) - if self._flow_bytes.get(flow, 0) + len(packet.payload) > MAX_UDP_FLOW_BYTES: - return - self._flows.setdefault(flow, []).append(packet.payload) - self._flow_bytes[flow] = self._flow_bytes.get(flow, 0) + len(packet.payload) - self._flow_last_seen[flow] = time.monotonic() - - def poll(self) -> None: - """Decode a flow that has stopped growing, and report it immediately.""" - if self.snapshot is not None or self._attempts >= GEAR_LIVE_MAX_ATTEMPTS: - return - now = time.monotonic() - candidates = [ - flow - for flow, size in self._flow_bytes.items() - if size >= GEAR_LIVE_MIN_FLOW_BYTES - and self._attempted_bytes.get(flow) != size - and now - self._flow_last_seen.get(flow, now) >= GEAR_LIVE_IDLE_SECONDS - ] - candidates.sort(key=self._flow_bytes.__getitem__, reverse=True) - - for flow in candidates[:MAX_GEAR_FLOW_ATTEMPTS]: - self._attempted_bytes[flow] = self._flow_bytes[flow] - self._attempts += 1 - snapshot = extract_gear(self._flows[flow]) - if snapshot.items: - self._report(snapshot) - return - - def finalize(self) -> None: - """Fallback scan for a flow that never idled. - - Normally ``poll`` has already decoded the burst. Rescanning per packet - instead of only here would be quadratic in flow size. - """ - if self.snapshot is not None: - return - # The gameplay flow dwarfs the others, but history traffic can outweigh - # a short login burst, so size only sets the order - the first flow that - # actually decodes wins. - busiest = sorted(self._flow_bytes, key=self._flow_bytes.__getitem__, reverse=True) - for flow in busiest[:MAX_GEAR_FLOW_ATTEMPTS]: - snapshot = extract_gear(self._flows[flow]) - if snapshot.items: - self._report(snapshot) - return - - def write( - self, - *, - capture_source: str | None = None, - user_uid: str | None = None, - server_id: str | None = None, - source: str = "live_capture", - ) -> Path | None: - if self.snapshot is None: - return None - path = gear_path(user_uid) - export = build_gear_export_json( - self.snapshot, - source=source, - capture_source=capture_source, - user_uid=user_uid, - server_id=server_id, - ) - path.write_text( - json.dumps(export, ensure_ascii=False, indent=2) + "\n", - encoding="utf-8", - ) - return path - - def _report(self, snapshot: GearSnapshot) -> None: - self.snapshot = snapshot - print_gear_captured( - snapshot.cartridges, - snapshot.modules, - len(snapshot.characters), - snapshot.characters_with_loadouts, - ) diff --git a/src/nte_history_exporter/live_capture/runner.py b/src/nte_history_exporter/live_capture/runner.py index eb6051c..4a25c0a 100644 --- a/src/nte_history_exporter/live_capture/runner.py +++ b/src/nte_history_exporter/live_capture/runner.py @@ -21,7 +21,6 @@ from nte_history_exporter.export.json_export import ( ) from nte_history_exporter.live_capture.backends import open_capture_backend from nte_history_exporter.live_capture.diagnostics import new_diagnostics_path, write_capture_diagnostics -from nte_history_exporter.live_capture.gear_collector import GearCollector from nte_history_exporter.live_capture.session import LiveHistorySession, UdpPacket from nte_history_exporter.live_capture.stop_key import StopKeyMonitor from nte_history_exporter.live_capture.windows_raw import detect_local_ipv4 @@ -89,17 +88,14 @@ def run_live_capture( active_gap_notices: set[str] = set() tcp_segments: dict[tuple[str, int, str, int], list[tuple[int, bytes]]] = {} achievement_records = [] - gear = GearCollector(local_ip) try: with StopKeyMonitor() as stop_key: for packet in capture.packets(): if stop_key.pressed(): break - gear.poll() # above the guard: the pcap read timeout drives it if packet is None: continue - gear.observe(packet) if ( not achievement_records and packet.protocol == "tcp" @@ -115,27 +111,12 @@ def run_live_capture( reassemble_tcp_segments(tcp_segments[flow]) ) if achievement_records: - in_game = [ - record - for record in achievement_records - if not record.achievement_id.casefold().startswith( - "playstation_" - ) - ] - playstation = [ - record - for record in achievement_records - if record.achievement_id.casefold().startswith( - "playstation_" - ) - ] + completed = [record for record in achievement_records if record.completed] console.print_achievements_captured( - sum(record.completed for record in in_game), - sum(record.status == "in_progress" for record in in_game), - sum(record.completed for record in playstation), + len(completed), sum( - record.status == "in_progress" - for record in playstation + record.achievement_id.lower().startswith("playstation_") + for record in completed ), ) pair_count_before = len(session.pairs) @@ -180,8 +161,6 @@ def run_live_capture( stats = capture.stats() capture.close() - gear.finalize() - exports = [] achievement_path = None diagnostics_path = None @@ -189,9 +168,7 @@ def run_live_capture( diagnostics_path = new_diagnostics_path() write_capture_diagnostics(diagnostics_path, session.diagnostic_report()) resolved_user_uid = user_uid or session.user_uid - if ( - session.kinds_seen() or achievement_records or gear.snapshot - ) and not resolved_user_uid: + if (session.kinds_seen() or achievement_records) and not resolved_user_uid: resolved_user_uid = console.prompt_user_uid() if achievement_records: achievement_path = _achievement_path(resolved_user_uid) @@ -206,11 +183,6 @@ def run_live_capture( json.dumps(achievement_export, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) - gear_path = gear.write( - capture_source=CAPTURE_SOURCE_LABELS.get(capture.name, capture.name), - user_uid=resolved_user_uid, - server_id=session.server_id, - ) resolved_server_id = session.server_id if session.kinds_seen() and not resolved_server_id: resolved_server_id = console.prompt_server_id() @@ -264,23 +236,16 @@ def run_live_capture( console.print_note("No pull history was captured (this is fine if you only wanted achievements).") print() console.print_note(f"Export written: {achievement_path}") - elif gear_path is not None: - console.print_success("Gear export complete.") - console.print_note("No pull history was captured (this is fine if you only wanted gear).") - print() else: console.print_problem("No history pages were captured.") console.print_note("Reopen a supported history screen and scroll from page 1.") console.print_note("If no page messages appear, return to the main menu and re-enter the game.") - if gear_path is not None: - console.print_note(f"Export written: {gear_path}") if diagnostics_path is not None: console.print_note(f"Diagnostics written: {diagnostics_path}") return { "exports": [], "diagnostics_path": diagnostics_path, "achievement_path": achievement_path, - "gear_path": gear_path, } for item in exports: @@ -302,8 +267,6 @@ def run_live_capture( if achievement_path is not None: print() console.print_note(f"Export written: {achievement_path}") - if gear_path is not None: - console.print_note(f"Export written: {gear_path}") if diagnostics_path is not None: console.print_note(f"Diagnostics written: {diagnostics_path}") @@ -319,7 +282,6 @@ def run_live_capture( "exports": exports, "diagnostics_path": diagnostics_path, "achievement_path": achievement_path, - "gear_path": gear_path, } diff --git a/src/nte_history_exporter/mapping_update.py b/src/nte_history_exporter/mapping_update.py index 2fa99ee..b37b9be 100644 --- a/src/nte_history_exporter/mapping_update.py +++ b/src/nte_history_exporter/mapping_update.py @@ -324,16 +324,9 @@ def _build_item_mapping( candidate_id.isdigit() or folded.startswith("fork_") or folded.startswith("characterawaken_") + or folded.startswith("fashion_vehicle_") ): continue - if folded.startswith("fashion_vehicle_"): - illustration = tables["illustrations"].get(candidate_id) - if illustration is not None: - vehicle_meta = _normalise_vehicle_livery(candidate_id, illustration, translations) - if vehicle_meta is None: - continue - result[candidate_id] = vehicle_meta - continue sources = (inventory, capital, appearances, vehicle_inventory) match = next((source.get(folded) for source in sources if folded in source), None) if match is None: @@ -350,31 +343,6 @@ def _build_item_mapping( return dict(sorted(result.items(), key=lambda pair: (pair[0].casefold(), pair[0]))) -def _normalise_vehicle_livery( - item_id: str, - row: Any, - translations: dict[str, list[tuple[str, str]]], -) -> dict[str, str] | None: - """Resolve a livery from its illustration, ignoring mismatched placeholder rows.""" - if not isinstance(row, dict): - raise MappingUpdateError(f"NTE_Assets row {item_id} must be an object") - head_icon = row.get("HeadIcon") - asset_path = head_icon.get("AssetPathName") if isinstance(head_icon, dict) else None - if not isinstance(asset_path, str) or not asset_path: - raise MappingUpdateError(f"NTE_Assets vehicle livery row {item_id} has no HeadIcon") - icon_id = asset_path.rsplit("/", 1)[-1].split(".", 1)[0] - if icon_id.casefold() != item_id.casefold(): - return None - item_name = row.get("ItemName_Override") - if not isinstance(item_name, dict): - raise MappingUpdateError(f"NTE_Assets vehicle livery row {item_id} has no ItemName_Override") - return { - "type": "cosmetic", - "name": _translate_name(item_id, item_name, translations).strip(), - "rank": "S", - } - - def _build_achievement_mapping( rows: dict[str, Any], translations: dict[str, list[tuple[str, str]]], diff --git a/src/nte_history_exporter/mappings.py b/src/nte_history_exporter/mappings.py index 5ce00ab..7165a9e 100644 --- a/src/nte_history_exporter/mappings.py +++ b/src/nte_history_exporter/mappings.py @@ -14,9 +14,6 @@ def load_mapping_file(filename: str) -> dict[str, Any]: ARC_META: dict[str, dict[str, Any]] = load_mapping_file("arcs.json") -ARC_META_BY_CASEFOLD = { - arc_id.casefold(): (arc_id, meta) for arc_id, meta in ARC_META.items() -} CHARACTERS: dict[str, dict[str, Any]] = load_mapping_file("characters.json") ITEMS: dict[str, dict[str, Any]] = load_mapping_file("items.json") ACHIEVEMENTS: dict[str, dict[str, Any]] = load_mapping_file("achievements.json") @@ -39,6 +36,3 @@ for _character_id, _meta in CHARACTERS.items(): } for _item_id, _meta in ITEMS.items(): REWARDS_BY_ID[_item_id] = {"type": _meta["type"], "id": _item_id, "name": _meta["name"], "rank": _meta.get("rank")} -REWARDS_BY_CASEFOLD = { - reward_id.casefold(): reward for reward_id, reward in REWARDS_BY_ID.items() -} diff --git a/tests/fixtures/nte_assets/DataTable/Gacha/GachaIllustrate.json b/tests/fixtures/nte_assets/DataTable/Gacha/GachaIllustrate.json index 9e28621..bd2fdf3 100644 --- a/tests/fixtures/nte_assets/DataTable/Gacha/GachaIllustrate.json +++ b/tests/fixtures/nte_assets/DataTable/Gacha/GachaIllustrate.json @@ -1 +1 @@ -[{"Rows":{"1003":{},"1099":{},"fork_alpha":{},"fork_new":{},"DIceNormal":{},"Dice_ticket_03":{},"Fashion_glide_2000":{},"Characterawaken_1003":{},"Fashion_vehicle_1010_V008":{"HeadIcon":{"AssetPathName":"/Game/UI/UI_Icon/Item/Fashion_vehicle_1010_V008.Fashion_vehicle_1010_V008"},"ItemName_Override":{"TableId":"/Game/Text/ST_VehicleData.ST_VehicleData","Key":"vehicle_1010_livery"}},"Fashion_vehicle_1072_V006":{"HeadIcon":{"AssetPathName":"/Game/UI/UI_Icon/Item/Fashion_vehicle_1036_V043.Fashion_vehicle_1036_V043"},"ItemName_Override":{"TableId":"/Game/Text/ST_VehicleData.ST_VehicleData","Key":"vehicle_placeholder_livery"}}}}] +[{"Rows":{"1003":{},"1099":{},"fork_alpha":{},"fork_new":{},"DIceNormal":{},"Dice_ticket_03":{},"Fashion_glide_2000":{},"Characterawaken_1003":{}}}] diff --git a/tests/fixtures/nte_assets/Localization/en/game.json b/tests/fixtures/nte_assets/Localization/en/game.json index 97e2667..0bbd71e 100644 --- a/tests/fixtures/nte_assets/Localization/en/game.json +++ b/tests/fixtures/nte_assets/Localization/en/game.json @@ -18,9 +18,5 @@ }, "ST_Appearance": { "glider_2000": "Inventory Glider" - }, - "ST_VehicleData": { - "vehicle_1010_livery": "Tiger Incoming! - Livery", - "vehicle_placeholder_livery": "Sandy Mirage - Livery" } } diff --git a/tests/test_achievement.py b/tests/test_achievement.py index 9145c1d..531895f 100644 --- a/tests/test_achievement.py +++ b/tests/test_achievement.py @@ -28,7 +28,7 @@ class AchievementPacketTests(unittest.TestCase): self.assertEqual(records[0].achievement_id, "Battle_30") self.assertTrue(records[0].completed) self.assertEqual(records[1].achievement_id, "Playstation_035") - self.assertEqual(records[1].status, "in_progress") + self.assertEqual(records[1].status, "not_started") def test_reassembles_tcp_data(self): self.assertEqual( diff --git a/tests/test_arc_decoding.py b/tests/test_arc_decoding.py index 84251e5..2d05040 100644 --- a/tests/test_arc_decoding.py +++ b/tests/test_arc_decoding.py @@ -1,14 +1,8 @@ from tests.support import * # noqa: F401,F403 -from nte_history_exporter.decoder.arc import annotate_arc_groups, _resolve_arc_metadata +from nte_history_exporter.decoder.arc import annotate_arc_groups class ArcDecodingTests(unittest.TestCase): - def test_arc_mapping_lookup_is_case_insensitive_and_canonical(self): - reward_id, metadata = _resolve_arc_metadata("fork_Wushoutieyu") - - self.assertEqual(reward_id, "fork_wushoutieyu") - self.assertEqual(metadata, {"name": "Raging Flames", "rank": "S"}) - def test_page_split_timestamp_variants_form_one_ten_pull(self): rows = [ { diff --git a/tests/test_export_contract.py b/tests/test_export_contract.py index 5400737..522e2b4 100644 --- a/tests/test_export_contract.py +++ b/tests/test_export_contract.py @@ -1,16 +1,11 @@ from tests.support import * # noqa: F401,F403 from nte_history_exporter import __version__ -from nte_history_exporter.constants import EXPORTER_VERSION from nte_history_exporter.decoder.achievement import AchievementRecord from nte_history_exporter.export.json_export import build_achievement_export_json from nte_history_exporter.live_capture.runner import _achievement_path class ExportContractTests(unittest.TestCase): - def test_public_version_references_match(self): - self.assertEqual(__version__, "0.4.0") - self.assertEqual(EXPORTER_VERSION, __version__) - def test_sanitized_export_omits_raw_packet_fields(self): annotated = annotate_groups(fixture_session().build_rows("permanent")) export = build_export_json(annotated, []) @@ -158,11 +153,13 @@ class ExportContractTests(unittest.TestCase): "total_achievements": 2, "completed_achievements": 1, "in_progress_achievements": 1, + "not_started_achievements": 0, }, "playstation": { "total_achievements": 1, "completed_achievements": 0, - "in_progress_achievements": 1, + "in_progress_achievements": 0, + "not_started_achievements": 1, }, }, ) diff --git a/tests/test_gear_decoding.py b/tests/test_gear_decoding.py deleted file mode 100644 index 235c65a..0000000 --- a/tests/test_gear_decoding.py +++ /dev/null @@ -1,747 +0,0 @@ -"""Gear decoding, driven entirely by synthetic packets. - -The wire format is fully understood, so every packet here is constructed from -the spec rather than captured: a 120-bit header, payload bits, two 1 bits and -zero padding. No real capture is used - see tests/fixtures/README.md. -""" - -import contextlib -import io -import json -import os -import struct -import tempfile -import unittest - -from nte_history_exporter import console - -from nte_history_exporter import __version__ -from nte_history_exporter.decoder.gear import ( - HEADER_BITS, - extract_gear, - payload_end_bit, - splice_datagrams, -) -from nte_history_exporter.export.gear_export import build_gear_export_json -from nte_history_exporter.live_capture import gear_collector, runner -from nte_history_exporter.live_capture.windows_raw import ParsedIpPacket - -from tests import support - - -# --- synthetic wire construction ------------------------------------------- - - -def string_frame(text: bytes) -> bytes: - """A self-validating frame.""" - return struct.pack(" bytes: - """<4 zero bytes>.""" - return string_frame(name) + b"\0\0\0\0" + struct.pack(" None: - if len(buf) < byte_offset + len(raw): - buf.extend(b"\0" * (byte_offset + len(raw) - len(buf))) - buf[byte_offset : byte_offset + len(raw)] = raw - - -def poke_u32_at_bit(buf: bytearray, bitpos: int, value: int) -> None: - """Write a little-endian u32 at an arbitrary bit offset.""" - needed = (bitpos + 32 + 7) // 8 + 1 - if len(buf) < needed: - buf.extend(b"\0" * (needed - len(buf))) - whole = int.from_bytes(buf, "little") - whole &= ~(0xFFFFFFFF << bitpos) - whole |= value << bitpos - packed = whole.to_bytes(len(buf) + 8, "little") - buf[:] = packed[: len(buf)] - - -def item_record( - item_id: bytes, - *, - level: int = 20, - entries=(), - instance: bytes = b"\x11\x22\x33\x44\x55\x66\x77\x88", - owner: bytes | None = None, -) -> bytes: - """One item record laid out at the documented offsets past the id frame.""" - body = bytearray(b"\0" * 64) - poke(body, 5, instance) - # Level is a u32 at +38 bytes; the owner reference starts at +39 bits-wise - # (312 bits) and its first three bytes are always zero, so the two fields - # coexist exactly as they do on the wire. - poke(body, 38, struct.pack(" bytes: - """An ability frame with the progress fields at their negative offsets.""" - lead = 128 # bytes of room for the fields that sit before the frame - buf = bytearray(b"\0" * lead) - frame = string_frame(b"GA_" + name + b"_Melee") - block_bit = lead * 8 - poke_u32_at_bit(buf, block_bit - 466, 67806) - poke_u32_at_bit(buf, block_bit - 434, level) - poke_u32_at_bit(buf, block_bit - 402, breakthroughs) - return bytes(buf) + frame - - -def loadout_cell(item_id: bytes, row: int, column: int) -> bytes: - """An id frame quoted with board coordinates and no stat chain.""" - body = bytearray(b"\0" * 32) - # Coordinates sit 96 and 128 bits past the id and are stored doubled. - poke(body, 12, struct.pack("> start) & ((1 << nbits) - 1) - - header = bytearray(15) - header[0:3] = b"\xaa\xbb\xcc" - # 14-bit packet sequence at bit 24. - header[3] = seq & 0xFF - header[4] = (seq >> 8) & 0x3F - header[12:15] = b"\x64\x06\xee" - - # payload bits, then the two 1 bits Unreal terminates with. - value = int.from_bytes(header, "little") - value |= chunk << HEADER_BITS - value |= 0b11 << (HEADER_BITS + nbits) - width = (HEADER_BITS + nbits + 2 + 7) // 8 - datagrams.append(value.to_bytes(width, "little")) - - start += nbits - seq = (seq + 1) & 0x3FFF - return datagrams - - -MODULE_ENTRIES = ( - (b"HPMaxAdd", 0.0), - (b"AtkAdd", 0.0), - (b"CritDamageBase", 0.06), - (b"CritBase", 0.03), - (b"AtkUp", 0.04), - (b"HPMaxUp", 0.04), -) - -CARTRIDGE_ENTRIES = ( - (b"AtkUp", 0.0), - (b"CritDamageBase", 0.06), - (b"CritBase", 0.03), - (b"DefUp", 0.07), - (b"HPMaxUp", 0.04), -) - -OWNER = b"\0\0\0\x0d\x0d\x20\x3d\x6d\x00\x00\x00\x00" - - -class FramingTests(unittest.TestCase): - def test_payload_end_bit_finds_the_double_terminator(self): - [datagram] = pack_datagrams(b"\x01\x02\x03\x04") - self.assertEqual(payload_end_bit(datagram), HEADER_BITS + 32) - - def test_splice_round_trips_a_payload_across_datagrams(self): - payload = bytes(range(256)) * 3 - # A chunk width that is not a multiple of 8 forces the accumulator to - # rejoin the stream at a bit offset, as the real capture does. - datagrams = pack_datagrams(payload, chunk_bits=333) - self.assertGreater(len(datagrams), 1) - self.assertEqual(splice_datagrams(datagrams)[: len(payload)], payload) - - def test_pure_acks_are_skipped(self): - payload = b"\x01\x02\x03\x04\x05\x06\x07\x08" - ack = bytes(12) - datagrams = pack_datagrams(payload) - self.assertEqual(splice_datagrams([ack] + datagrams + [ack])[:8], payload) - - -class ItemRecordTests(unittest.TestCase): - def decode(self, payload: bytes, **kwargs): - return extract_gear(pack_datagrams(b"\0" * 8 + payload, **kwargs)) - - def test_module_record_decodes(self): - snapshot = self.decode( - item_record(b"cell3_style6_1_Orange", level=20, entries=MODULE_ENTRIES) - ) - [item] = snapshot.items - self.assertEqual(item.kind, "module") - self.assertEqual(item.item_id, "cell3_style6_1_Orange") - self.assertEqual(item.shape, "cell3_style6") - self.assertEqual(item.module_type, "III") - self.assertEqual(item.level, 20) - self.assertEqual(item.rarity, "orange") - self.assertEqual(item.instance, "1122334455667788") - self.assertIsNone(item.owner_group) - # Both mains are computed from the cell count, never transmitted. - self.assertEqual(dict(item.main_stats), {"HPMaxAdd": 840.0, "AtkAdd": 63.0}) - self.assertEqual(len(item.substats), 4) - - def test_cartridge_record_decodes_with_its_set(self): - snapshot = self.decode( - item_record(b"Incantation_orange", level=20, entries=CARTRIDGE_ENTRIES) - ) - [item] = snapshot.items - self.assertEqual(item.kind, "cartridge") - self.assertIsNotNone(item.set_name) - self.assertEqual(len(item.main_stats), 1) - self.assertEqual(len(item.substats), 4) - - def test_owner_reference_is_read_when_equipped(self): - snapshot = self.decode( - item_record( - b"cell3_style6_1_Orange", - level=20, - entries=MODULE_ENTRIES, - owner=OWNER, - ) - ) - [item] = snapshot.items - self.assertEqual(item.owner_group, OWNER.hex()) - # The owner field overlaps the level word without corrupting it. - self.assertEqual(item.level, 20) - - def test_record_spanning_a_datagram_boundary_decodes(self): - record = item_record( - b"cell3_style6_1_Orange", level=20, entries=MODULE_ENTRIES, owner=OWNER - ) - # Chunks far smaller than the record, at a width that is not a whole - # number of bytes: every field lands across at least one boundary. - snapshot = self.decode(record, chunk_bits=101) - [item] = snapshot.items - self.assertEqual(item.item_id, "cell3_style6_1_Orange") - self.assertEqual(item.level, 20) - self.assertEqual(item.owner_group, OWNER.hex()) - self.assertEqual(dict(item.main_stats), {"HPMaxAdd": 840.0, "AtkAdd": 63.0}) - - def test_unframed_id_is_ignored(self): - # Same bytes, wrong length prefix: the frame check must reject it. - record = bytearray( - item_record(b"cell3_style6_1_Orange", entries=MODULE_ENTRIES) - ) - record[0:4] = struct.pack("