commit bb3fe3b19a0df1477043f7adba540927e7f57f73 Author: Golumpa Date: Wed Jun 10 23:59:01 2026 +0100 Initial NTE history exporter Initial NTE history exporter Add a Windows live-capture exporter for NTE pull history with support for Permanent Monopoly, Limited Monopoly, and Arc Miracle Box histories. Included: - live packet capture via Windows raw sockets - CSV and JSON export generation - normalized reward and banner mapping files - Monopoly v7 Points Gift / Chase Reward classification - Arc Miracle Box decoding with shared pity metadata - timestamp grouping and UID generation - Windows launcher scripts with admin prompting - docs for usage, export format, limitations, and packet notes - tests covering decoder behaviour and mapping consistency Generated exports, captures, briefs, and local data are ignored for privacy. diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..46c7aed --- /dev/null +++ b/.gitignore @@ -0,0 +1,32 @@ +__pycache__/ +*.py[cod] +*$py.class + +.pytest_cache/ +.mypy_cache/ +.ruff_cache/ +.coverage +htmlcov/ + +.venv/ +venv/ +env/ + +dist/ +build/ +*.egg-info/ + +# Captures, generated exports, and reference research outputs. +exports/* +!exports/.gitkeep +briefs/ +*.flows +*.pcap +*.pcapng +*.cap + +# Local/editor noise. +.DS_Store +Thumbs.db +.idea/ +.vscode/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..a4f2191 --- /dev/null +++ b/README.md @@ -0,0 +1,68 @@ +# NTE History Exporter + +Prototype CLI exporter for Neverness to Everness pull history. + +This version supports: + +- Game: Neverness to Everness +- System: Monopoly +- Banner ID: `Lottery_Permanent` +- Banner name: Standard Board / Permanent Board +- Banner ID: `Lottery_LimitedCharacter` +- Banner name: Limited Character Board +- System: Gashapon +- Banner ID: `Arc_MiracleBox` +- Banner name: Arc Miracle Box + +## What It Does + +The exporter decodes Permanent Board, Limited Character Board, and Arc Miracle Box history pages from captured UDP data, applies conservative timestamp-boundary handling, and writes sanitized JSON suitable for tracker import. + +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 only. + +## Privacy + +Do not commit packet captures, generated exports, research briefs, or personal account data. The repository keeps `exports/` as an empty output folder, but ignores everything generated inside it. + +## Usage + +Live capture, Windows: + +```powershell +.\run-exporter.ps1 --live +``` + +Launch it before logging in, then open any supported history board in game. The tool keeps listening until you press any key. Exports are written under `exports\` as `Permanent_.csv/json`, `Limited_.csv/json`, or `Arc_.csv/json`. + +If only one banner is captured, the export JSON is copied to the clipboard. If multiple banners are captured in the same run, clipboard copy is skipped so one banner does not overwrite another. + +File replay: + +```powershell +.\run-exporter.ps1 capture.flows +``` + +For reliable deduplication, start from page 1 and scroll downward. If you only want pages 1-5, scroll through page 6 as well so the exporter can confirm the final timestamp group boundary. + +Limited character history is exported as one shared pity pool: `Lottery_LimitedCharacter`. +Arc Miracle Box history is exported as one shared pity pool: `Arc_MiracleBox`. + +Live capture needs a local admin-capable packet socket on Windows. The prototype uses the built-in raw socket path first, without requiring mitmproxy capture files. + +## Boundary Policy + +NTE history records do not appear to contain a unique server-side roll ID. UIDs are generated from decoded record fields and the record's order within all rows sharing the same raw timestamp. + +Because 10-pulls can span page boundaries, partial timestamp groups can produce unstable UIDs. Normal mode exports only complete/stable timestamp groups. Boundary groups are skipped with warnings when the exporter cannot prove they are complete. + +For Monopoly, Points Gift and Chase Reward rows stay in the timestamp group for UID ordinal generation, but only `result_type = dice` rows count toward pull-set sizing. Arc groups are expected to be complete 10-pull timestamp groups. + +## Current Adapters + +- live Windows raw-socket capture +- `mitmproxy .flows` research decoder + +Planned later: + +- live Npcap/libpcap capture +- UI wrapper around the CLI diff --git a/docs/export-format.md b/docs/export-format.md new file mode 100644 index 0000000..d31f8f6 --- /dev/null +++ b/docs/export-format.md @@ -0,0 +1,53 @@ +# Export Format + +The sanitized JSON export uses: + +```json +{ + "format": "nte-history-export", + "format_version": 1, + "game": "Neverness to Everness", + "source": "packet_capture", + "exporter": { + "name": "nte-history-exporter", + "version": "0.1.0" + }, + "banner": { + "id": "Lottery_Permanent", + "name": "Standard Board", + "system": "Monopoly", + "shared_pity": false + }, + "scan": { + "mode": "stable_only", + "boundary_policy": "drop_incomplete_timestamp_groups", + "decoded_records": 0, + "exported_records": 0, + "skipped_records": 0, + "warnings": [] + }, + "records": [] +} +``` + +Record UIDs are deterministic: + +Monopoly: + +```text +nte|monopoly|Lottery_Permanent|timestamp_raw|timestamp_group_ordinal|roll_result|reward_key_hex|quantity +``` + +Limited character records use `Lottery_LimitedCharacter` in the same UID source position. + +Arc: + +```text +nte|gashapon|Arc_MiracleBox|timestamp_raw|timestamp_group_ordinal|reward_key_hex +``` + +The final UID is `sha256(source).hexdigest()[0:32]`. + +Each record includes `pool_group_id`. Arc records use the shared `reward_*` fields and include `source_type: "miracle_box"`. + +Normal JSON exports do not include raw packets or capture-only metadata. diff --git a/docs/limitations.md b/docs/limitations.md new file mode 100644 index 0000000..ad49d45 --- /dev/null +++ b/docs/limitations.md @@ -0,0 +1,17 @@ +# Limitations + +This is a prototype and supports `Lottery_Permanent`, `Lottery_LimitedCharacter`, and `Arc_MiracleBox`. + +Known limitations: + +- Other NTE banners are not implemented yet. +- The game appears not to provide a unique server-side roll ID in the decoded record body. +- UIDs are generated deterministically from decoded fields and timestamp-group order. +- Boundary timestamp groups may be skipped to avoid exporting unstable data. +- Page gaps are ignored outside the longest continuous run and reported as warnings. +- Live capture currently uses Windows raw sockets and requires administrator permission. +- The file adapter reads mitmproxy `.flows` captures for research and testing. +- The more stable and reliable Npcap/libpcap capture is not implemented yet. +- Reward mappings are research mappings and should be expanded as more samples are decoded. + +Privacy guardrail: raw packet data must not be included in sanitized exports. diff --git a/docs/packet-format.md b/docs/packet-format.md new file mode 100644 index 0000000..eb82f95 --- /dev/null +++ b/docs/packet-format.md @@ -0,0 +1,40 @@ +# Packet Format Notes + +This prototype supports separate Monopoly and Arc/Gashapon history decoders. + +## Monopoly + +- History is fetched over the UDP game connection. +- Client history-page requests are 45 bytes. +- History request constant: `4220` / `0x107c`. +- Request selector `4`: `Lottery_Permanent`. +- Request selector `8`: `Lottery_LimitedCharacter`. +- Request page cursor: `page_number * 4`. +- Normal server responses contain 5 history records. +- The final page may contain fewer than 5 records. + +Decoded fields: + +- `roll_result = first u32 / 4` +- `roll_result = 0` means Points Gift +- Some page-first records have a short prefix before the record body. For these, a hidden signed source flag immediately after the visible dice field overrides the visible dice: + - `source_flag = 0` means Points Gift + - `source_flag = -4` means Chase Reward +- Timestamp is an 8-byte little-endian value +- `unix_seconds = little_endian_u64(timestamp_raw) / 40000000 - 62135596800` +- Rewards are identified by stable binary keys + +Page and row numbers are research metadata only. They must not be used for permanent dedupe because they shift when new history appears. + +Timestamp groups keep all records with the same raw timestamp together for UID ordinal generation. For boundary/group-size detection, only `result_type = dice` rows count as pull-set members; Points Gift and Chase Reward rows stay in the group but do not increase the dice-only group count. + +## Arc / Gashapon + +- Arc history uses a separate 34-byte request. +- Request constant: `2060` / `0x080c`. +- Cursor step: `2`. +- Pool: `Arc_MiracleBox`. +- Each response page normally contains 5 records. +- Arc timestamps use `unix_seconds = little_endian_u64(timestamp_raw) / 20000000 - 62135596800`. +- Arc pulls are treated as 10-pull timestamp groups; incomplete groups are skipped by default. +- Arc rows use the same `reward_type`, `reward_id`, `reward_name`, `reward_rank`, and `reward_key_hex` fields as Monopoly rows. diff --git a/exports/.gitkeep b/exports/.gitkeep new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/exports/.gitkeep @@ -0,0 +1 @@ + diff --git a/mappings/arc_miracle_box.json b/mappings/arc_miracle_box.json new file mode 100644 index 0000000..902f31a --- /dev/null +++ b/mappings/arc_miracle_box.json @@ -0,0 +1,32 @@ +{ + "pool_key": "arc_miracle_box", + "game": "Neverness to Everness", + "system": { + "id": "gashapon", + "name": "Gashapon" + }, + "banner": { + "id": "Arc_MiracleBox", + "name": "Arc Miracle Box", + "shared_pity": true + }, + "request": { + "family": "arc", + "length": 34, + "constant": 2060, + "constant_hex": "0x080c", + "selector": null, + "cursor_offset": 29, + "cursor_step": 2, + "page_formula": "cursor / 2" + }, + "response": { + "records_per_full_page": 5, + "timestamp_group_size": 10, + "first_record_offset_hex": "0x4c" + }, + "notes": [ + "Arc history is separate from Monopoly history.", + "Arc timestamp groups are expected to be complete 10-pull groups." + ] +} diff --git a/mappings/arcs.json b/mappings/arcs.json new file mode 100644 index 0000000..2f1c54d --- /dev/null +++ b/mappings/arcs.json @@ -0,0 +1,14 @@ +{ + "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_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"} +} diff --git a/mappings/characters.json b/mappings/characters.json new file mode 100644 index 0000000..0a7e85a --- /dev/null +++ b/mappings/characters.json @@ -0,0 +1,21 @@ +{ + "1003": {"name": "Sakiri", "rank": "S"}, + "1004": {"name": "Lacrimosa", "rank": "S"}, + "1008": {"name": "Skia", "rank": "A"}, + "1010": {"name": "Nanally", "rank": "S"}, + "1019": {"name": "Mint", "rank": "A"}, + "1020": {"name": "Haniel", "rank": "A"}, + "1021": {"name": "Edgar", "rank": "A"}, + "1023": {"name": "Baicang", "rank": "S"}, + "1025": {"name": "Hathor", "rank": "S"}, + "1033": {"name": "Adler", "rank": "A"}, + "1039": {"name": "Fadia", "rank": "S"}, + "1046": {"name": "Zero", "rank": "S"}, + "1051": {"name": "Zero", "rank": "S"}, + "1052": {"name": "Hotori", "rank": "S"}, + "1054": {"name": "Daffodill", "rank": "S"}, + "1055": {"name": "Jiuyuan", "rank": "S"}, + "1056": {"name": "Lacrimosa", "rank": "S"}, + "1070": {"name": "Aurelia", "rank": "A"}, + "1073": {"name": "Chiz", "rank": "S"} +} diff --git a/mappings/limited_character_board.json b/mappings/limited_character_board.json new file mode 100644 index 0000000..f2c360a --- /dev/null +++ b/mappings/limited_character_board.json @@ -0,0 +1,31 @@ +{ + "pool_key": "limited_character", + "game": "Neverness to Everness", + "system": { + "id": "monopoly", + "name": "Monopoly" + }, + "banner": { + "id": "Lottery_LimitedCharacter", + "name": "Limited Character Board", + "shared_pity": true + }, + "request": { + "family": "monopoly", + "length": 45, + "constant": 4220, + "constant_hex": "0x107c", + "selector": 8, + "cursor_offset": 31, + "cursor_step": 4, + "page_formula": "cursor / 4" + }, + "response": { + "records_per_full_page": 5, + "marker_hex": "4c0000000c85c99141bdbdb17d0da185c9858dd195c901" + }, + "notes": [ + "Limited character boards are exported under one shared pool group.", + "Do not split by individual limited banners such as Nanali, Xun, or AnHunQu." + ] +} diff --git a/mappings/permanent_board.json b/mappings/permanent_board.json new file mode 100644 index 0000000..91d32a1 --- /dev/null +++ b/mappings/permanent_board.json @@ -0,0 +1,57 @@ +{ + "pool_key": "permanent", + "game": "Neverness to Everness", + "system": { + "id": "monopoly", + "name": "Monopoly" + }, + "banner": { + "id": "Lottery_Permanent", + "name": "Standard Board", + "shared_pity": false + }, + "request": { + "family": "monopoly", + "length": 45, + "constant": 4220, + "constant_hex": "0x107c", + "selector": 4, + "cursor_offset": 31, + "cursor_step": 4, + "page_formula": "cursor / 4" + }, + "response": { + "records_per_full_page": 5, + "marker_hex": "440000000c85c99141bdbdb17d3995dd49bdb19501" + }, + "character_ids": [ + "1003", + "1023", + "1025", + "1039", + "1054", + "1055", + "1008", + "1019", + "1020", + "1021", + "1033", + "1070" + ], + "ssr_character_ids": [ + "1003", + "1023", + "1025", + "1039", + "1054", + "1055" + ], + "guarantee": { + "pool_guarantee_count": 90, + "pool_upgrade_count": 70, + "main_cost_item": { + "id": "DiceNormal", + "quantity": 1 + } + } +} diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..05144bf --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,13 @@ +[project] +name = "nte-history-exporter" +version = "0.1.0" +description = "Prototype NTE Monopoly Permanent Board pull-history exporter." +requires-python = ">=3.10" +dependencies = [] + +[project.scripts] +nte-history-exporter = "nte_history_exporter.cli:main" + +[tool.pytest.ini_options] +pythonpath = ["src"] +testpaths = ["tests"] diff --git a/run-exporter.cmd b/run-exporter.cmd new file mode 100644 index 0000000..fa511ed --- /dev/null +++ b/run-exporter.cmd @@ -0,0 +1,3 @@ +@echo off +powershell -NoProfile -ExecutionPolicy Bypass -File "%~dp0run-exporter.ps1" %* +echo. diff --git a/run-exporter.ps1 b/run-exporter.ps1 new file mode 100644 index 0000000..5fe63c3 --- /dev/null +++ b/run-exporter.ps1 @@ -0,0 +1,83 @@ +$ErrorActionPreference = 'Stop' + +function Test-IsAdministrator { + $identity = [Security.Principal.WindowsIdentity]::GetCurrent() + $principal = [Security.Principal.WindowsPrincipal]::new($identity) + return $principal.IsInRole([Security.Principal.WindowsBuiltInRole]::Administrator) +} + +function ConvertTo-QuotedArgument { + param([string]$Value) + return '"' + ($Value -replace '"', '\"') + '"' +} + +function Test-RequiresAdministrator { + param([string[]]$Arguments) + + if ($Arguments.Count -eq 0) { + return $true + } + if ($Arguments -contains '--help' -or $Arguments -contains '-h') { + return $false + } + if ($Arguments -contains '--live') { + return $true + } + + foreach ($arg in $Arguments) { + if (-not $arg.StartsWith('-')) { + return $false + } + } + return $true +} + +if ((Test-RequiresAdministrator -Arguments $args) -and -not (Test-IsAdministrator)) { + Write-Host 'Administrator permission is required for live packet capture.' + Write-Host 'Windows will show a UAC prompt so the exporter can listen to game network traffic.' + Write-Host '' + $answer = Read-Host 'Continue and request administrator access? (Y/N)' + if ($answer -notin @('Y', 'y', 'Yes', 'yes')) { + Write-Host 'Cancelled. Run again when you are ready.' + exit 1 + } + Write-Host 'Requesting administrator access...' + + $arguments = @( + '-NoProfile', + '-ExecutionPolicy', 'Bypass', + '-NoExit', + '-File', (ConvertTo-QuotedArgument $PSCommandPath) + ) + + foreach ($arg in $args) { + $arguments += ConvertTo-QuotedArgument $arg + } + + Start-Process -FilePath 'powershell.exe' -Verb RunAs -ArgumentList ($arguments -join ' ') + exit +} + +function Get-PythonCommand { + $python = Get-Command 'python.exe' -ErrorAction SilentlyContinue + if ($python) { + return @($python.Source) + } + + $py = Get-Command 'py.exe' -ErrorAction SilentlyContinue + if ($py) { + return @($py.Source, '-3') + } + + throw 'Python 3.10 or newer is required. Install Python, then run this script again.' +} + +$pythonCommand = @(Get-PythonCommand) +$env:PYTHONPATH = Join-Path $PSScriptRoot 'src' + +$pythonExe = $pythonCommand[0] +if ($pythonCommand.Count -gt 1) { + & "$pythonExe" $pythonCommand[1] -m nte_history_exporter.cli @args +} else { + & "$pythonExe" -m nte_history_exporter.cli @args +} diff --git a/src/nte_history_exporter/__init__.py b/src/nte_history_exporter/__init__.py new file mode 100644 index 0000000..3dc1f76 --- /dev/null +++ b/src/nte_history_exporter/__init__.py @@ -0,0 +1 @@ +__version__ = "0.1.0" diff --git a/src/nte_history_exporter/adapters/__init__.py b/src/nte_history_exporter/adapters/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/src/nte_history_exporter/adapters/__init__.py @@ -0,0 +1 @@ + diff --git a/src/nte_history_exporter/adapters/mitmproxy_flows.py b/src/nte_history_exporter/adapters/mitmproxy_flows.py new file mode 100644 index 0000000..6cd6521 --- /dev/null +++ b/src/nte_history_exporter/adapters/mitmproxy_flows.py @@ -0,0 +1,146 @@ +from __future__ import annotations + +import struct +from pathlib import Path +from typing import Any + +from nte_history_exporter.decoder.boundary import longest_monotonic_page_run +from nte_history_exporter.decoder.arc import ( + arc_request_page, + build_arc_rows_from_pairs, + is_arc_history_request, + parse_arc_response, + select_continuous_arc_run, +) +from nte_history_exporter.decoder.protocol import ( + history_request_kind, + is_history_request, + request_page, + response_contains_history_marker, +) +from nte_history_exporter.decoder.run import build_rows_from_pairs, fmt_packet_time + + +def parse_tnetstring(data: bytes, i: int = 0) -> tuple[Any, int]: + j = data.find(b":", i) + if j < 0: + raise EOFError("no tnetstring length separator found") + length = int(data[i:j]) + start = j + 1 + end = start + length + payload = data[start:end] + typ = chr(data[end]) + next_i = end + 1 + + if typ in ",;": + return payload, next_i + if typ == "#": + return int(payload), next_i + if typ == "^": + return float(payload), next_i + if typ == "!": + return payload == b"true", next_i + if typ == "~": + return None, next_i + if typ == "]": + arr = [] + k = 0 + while k < len(payload): + value, k = parse_tnetstring(payload, k) + arr.append(value) + return arr, next_i + if typ == "}": + obj = {} + k = 0 + while k < len(payload): + key, k = parse_tnetstring(payload, k) + value, k = parse_tnetstring(payload, k) + obj[key] = value + return obj, next_i + raise ValueError(f"unknown tnetstring type {typ!r}") + + +def read_flows(path: str | Path) -> list[Any]: + data = Path(path).read_bytes() + i = 0 + flows = [] + while i < len(data): + value, i = parse_tnetstring(data, i) + flows.append(value) + return flows + + +def find_udp_flow(flows: list[Any], preferred_index: int | None = None) -> tuple[int, Any]: + if preferred_index is not None: + return preferred_index, flows[preferred_index] + candidates = [] + for idx, flow in enumerate(flows): + if b"messages" not in flow: + continue + server = flow.get(b"server_conn", {}) + if server.get(b"transport_protocol") != b"udp": + continue + candidates.append((len(flow[b"messages"]), idx, flow)) + if not candidates: + raise RuntimeError("no UDP flow with messages found") + _, idx, flow = max(candidates) + return idx, flow + + +def pair_response(messages: list[Any], request_index: int) -> tuple[int | None, bytes, float | None]: + for j in range(request_index + 1, min(request_index + 50, len(messages))): + from_client, content, ts = messages[j] + if not from_client and len(content) >= 150 and response_contains_history_marker(content): + return j, content, ts + return None, b"", None + + +def pair_arc_response(messages: list[Any], request_index: int) -> tuple[int | None, bytes, float | None]: + for j in range(request_index + 1, min(request_index + 30, len(messages))): + from_client, content, ts = messages[j] + if not from_client and len(content) >= 100 and parse_arc_response(content): + return j, content, ts + return None, b"", None + + +def decode_mitmproxy_flows(path: str | Path, flow_index: int | None = None) -> dict[str, Any]: + flows = read_flows(path) + resolved_flow_index, flow = find_udp_flow(flows, flow_index) + messages = flow[b"messages"] + + pairs = [] + arc_pairs = [] + for i, msg in enumerate(messages): + from_client, content, ts = msg + if from_client and is_arc_history_request(content): + page = arc_request_page(content) + response_index, response_content, response_ts = pair_arc_response(messages, i) + if response_index is not None: + arc_pairs.append((page, page * 2, i, ts, response_index, response_ts, response_content)) + continue + + if not from_client or not is_history_request(content): + continue + kind = history_request_kind(content) + offset = struct.unpack_from(" argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Decode NTE Monopoly history from mitmproxy .flows captures or live capture." + ) + parser.add_argument("capture_source", nargs="?", help="mitmproxy .flows file; omit when using --live") + parser.add_argument("--live", action="store_true", help="capture live UDP traffic instead of reading a .flows file") + parser.add_argument("--flow-index", type=int, default=None) + parser.add_argument("--interface-ip", default=None, help="local IPv4 address to bind for live capture") + parser.add_argument("--no-clipboard", action="store_true", help="do not copy live exports to clipboard") + parser.add_argument( + "--allow-boundary-records", + action="store_true", + help="Debug only: export boundary groups even when they may be partial.", + ) + return parser + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + if args.live or not args.capture_source: + result = run_live_capture( + interface_ip=args.interface_ip, + copy_clipboard=not args.no_clipboard, + ) + for item in result["exports"]: + export = item["export"] + print( + "{banner}: decoded {decoded}, exported {exported}, skipped {skipped}.".format( + banner=export["banner"]["name"], + decoded=export["scan"]["decoded_records"], + exported=export["scan"]["exported_records"], + skipped=export["scan"]["skipped_records"], + ) + ) + for warning in export["scan"]["warnings"]: + suffix = f" ({warning['records']} records)" if "records" in warning else "" + print(f"WARNING {warning['code']}: {warning['reason']}{suffix}") + return 0 + + decoded = decode_mitmproxy_flows(args.capture_source, args.flow_index) + if decoded["arc_rows"] and not decoded["rows"]: + rows = decoded["arc_rows"] + warnings = decoded["arc_warnings"] + arc_stability_warnings(rows) + kind = "arc_miracle_box" + best_run = decoded["best_arc_run"] + pair_count = len(decoded["arc_pairs"]) + else: + rows, warnings = annotate_groups( + decoded["rows"], + starts_from_page_1=decoded["starts_from_page_1"], + stable_only=not args.allow_boundary_records, + ) + warnings = page_gap_warnings(decoded["pairs"], decoded["best_run"]) + warnings + kind = decoded["best_run"][0][7] if decoded["best_run"] and len(decoded["best_run"][0]) > 7 else "permanent" + best_run = decoded["best_run"] + pair_count = len(decoded["pairs"]) + + out_path, json_path = export_paths(kind) + write_csv(out_path, rows) + export = build_export_json( + rows, + warnings, + source="packet_capture", + flow_index=decoded["flow_index"], + candidate_request_response_pairs=pair_count, + pages_seen=[p[0] for p in best_run], + ) + json_path.write_text(json.dumps(export, ensure_ascii=False, indent=2), encoding="utf-8") + + print(f"CSV written: {out_path}") + print(f"JSON written: {json_path}") + print( + "Decoded {decoded}, exported {exported}, skipped {skipped}.".format( + decoded=export["scan"]["decoded_records"], + exported=export["scan"]["exported_records"], + skipped=export["scan"]["skipped_records"], + ) + ) + for warning in warnings: + suffix = f" ({warning['records']} records)" if "records" in warning else "" + print(f"WARNING {warning['code']}: {warning['reason']}{suffix}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/nte_history_exporter/constants.py b/src/nte_history_exporter/constants.py new file mode 100644 index 0000000..8649c08 --- /dev/null +++ b/src/nte_history_exporter/constants.py @@ -0,0 +1,62 @@ +GAME_NAME = "Neverness to Everness" +GAME_UID_PART = "nte" +SYSTEM = "monopoly" +SYSTEM_NAME = "Monopoly" +ARC_SYSTEM = "gashapon" +ARC_SYSTEM_NAME = "Gashapon" +BANNER_ID = "Lottery_Permanent" +BANNER_NAME = "Standard Board" +LIMITED_CHARACTER_BANNER_ID = "Lottery_LimitedCharacter" +LIMITED_CHARACTER_BANNER_NAME = "Limited Character Board" +ARC_BANNER_ID = "Arc_MiracleBox" +ARC_BANNER_NAME = "Arc Miracle Box" +EXPORTER_NAME = "nte-history-exporter" +EXPORTER_VERSION = "0.1.0" + +HISTORY_REQUEST_BANNER = 4220 +HISTORY_REQUEST_LENGTH = 45 +PERMANENT_SELECTOR = 4 +LIMITED_CHARACTER_SELECTOR = 8 +HISTORY_PAGE_CURSOR_MULTIPLIER = 4 + +POOL_META = { + "permanent": { + "id": BANNER_ID, + "name": BANNER_NAME, + "system": SYSTEM_NAME, + "shared_pity": False, + }, + "limited_character": { + "id": LIMITED_CHARACTER_BANNER_ID, + "name": LIMITED_CHARACTER_BANNER_NAME, + "system": SYSTEM_NAME, + "shared_pity": True, + }, + "arc_miracle_box": { + "id": ARC_BANNER_ID, + "name": ARC_BANNER_NAME, + "system": ARC_SYSTEM_NAME, + "shared_pity": True, + }, +} + +ARC_HISTORY_REQUEST_BANNER = 2060 +ARC_HISTORY_REQUEST_LENGTH = 34 +ARC_HISTORY_CURSOR_OFFSET = 29 +ARC_HISTORY_PAGE_CURSOR_MULTIPLIER = 2 +ARC_RESPONSE_FIRST_RECORD_OFFSET = 0x4C +ARC_TIMESTAMP_TICKS_PER_SECOND = 20_000_000 + +TIMESTAMP_TICKS_PER_SECOND = 40_000_000 +DOTNET_UNIX_EPOCH_SECONDS = 62_135_596_800 + +PERMANENT_MARKER = bytes.fromhex("440000000c85c99141bdbdb17d3995dd49bdb19501") +LIMITED_CHARACTER_MARKER = bytes.fromhex("4c0000000c85c99141bdbdb17d0da185c9858dd195c901") +MARKER = PERMANENT_MARKER +MARKERS = [PERMANENT_MARKER, LIMITED_CHARACTER_MARKER] +VALID_DICE_FIELDS = {0, 4, 8, 12, 16, 20, 24} +KEY_PREFIXES = [ + bytes.fromhex("98bdc9ad"), + bytes.fromhex("10a58d95"), + bytes.fromhex("c4c0"), +] diff --git a/src/nte_history_exporter/decoder/__init__.py b/src/nte_history_exporter/decoder/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/src/nte_history_exporter/decoder/__init__.py @@ -0,0 +1 @@ + diff --git a/src/nte_history_exporter/decoder/arc.py b/src/nte_history_exporter/decoder/arc.py new file mode 100644 index 0000000..44f4d7c --- /dev/null +++ b/src/nte_history_exporter/decoder/arc.py @@ -0,0 +1,208 @@ +from __future__ import annotations + +import hashlib +import struct +from collections import defaultdict +from datetime import datetime, timezone +from typing import Any + +from nte_history_exporter.constants import ( + ARC_BANNER_ID, + ARC_HISTORY_CURSOR_OFFSET, + ARC_HISTORY_PAGE_CURSOR_MULTIPLIER, + ARC_HISTORY_REQUEST_BANNER, + ARC_HISTORY_REQUEST_LENGTH, + ARC_RESPONSE_FIRST_RECORD_OFFSET, + ARC_SYSTEM, + ARC_TIMESTAMP_TICKS_PER_SECOND, + DOTNET_UNIX_EPOCH_SECONDS, + GAME_UID_PART, + POOL_META, +) +from nte_history_exporter.decoder.boundary import longest_monotonic_page_run +from nte_history_exporter.decoder.run import fmt_packet_time +from nte_history_exporter.mappings import ARC_META + + +def is_arc_history_request(content: bytes) -> bool: + return len(content) == ARC_HISTORY_REQUEST_LENGTH and struct.unpack_from(" int: + return struct.unpack_from(" str | None: + if raw.endswith(b"\x00"): + raw = raw[:-1] + prefix = bytes.fromhex("ccdee4d6be") + if not raw.startswith(prefix): + return None + out = "fork_" + for byte in raw[len(prefix) :]: + if 0xC2 <= byte <= 0xF4 and (byte - 0xC2) % 2 == 0: + out += chr(ord("a") + (byte - 0xC2) // 2) + elif 0x82 <= byte <= 0xB4 and (byte - 0x82) % 2 == 0: + out += chr(ord("A") + (byte - 0x82) // 2) + else: + out += f"_{byte:02x}" + return out + + +def decode_arc_timestamp(raw8: bytes) -> tuple[int, float, str]: + ticks = struct.unpack(" list[dict[str, Any]]: + pos = ARC_RESPONSE_FIRST_RECORD_OFFSET + records: list[dict[str, Any]] = [] + while pos + 4 <= len(response): + start = pos + name_len2 = struct.unpack_from(" 200 or name_len2 % 2: + break + name_len = name_len2 // 2 + if pos + name_len + 4 > len(response): + break + name_raw = response[pos : pos + name_len] + pos += name_len + + type_len2 = struct.unpack_from(" 200 or type_len2 % 2: + break + type_len = type_len2 // 2 + if pos + type_len + 8 > len(response): + break + type_raw = response[pos : pos + type_len] + pos += type_len + + timestamp_raw = response[pos : pos + 8] + pos += 8 + arc_id = decode_arc_key(name_raw) or name_raw.hex() + meta = ARC_META.get(arc_id, {}) + ticks, unix_seconds, timestamp_decoded = decode_arc_timestamp(timestamp_raw) + records.append( + { + "record_start": start, + "record_end": pos, + "record_len": pos - start, + "reward_key_hex": name_raw.hex(), + "reward_type": "arc", + "reward_id": arc_id, + "reward_name": meta.get("name", "UNKNOWN"), + "reward_rank": meta.get("rank", ""), + "type_key_hex": type_raw.hex(), + "source_type": "miracle_box", + "timestamp_raw_hex": timestamp_raw.hex(), + "timestamp_ticks": ticks, + "timestamp_unix": unix_seconds, + "timestamp_decoded": timestamp_decoded, + "record_hex": response[start:pos].hex(), + } + ) + return records + + +def build_arc_rows_from_pairs(pairs: list[tuple]) -> list[dict[str, Any]]: + pool = POOL_META["arc_miracle_box"] + rows: list[dict[str, Any]] = [] + for pair in pairs: + page, offset, req_i, req_ts, resp_i, resp_ts, response = pair[:7] + records = parse_arc_response(response) + for row_index, record in enumerate(records, start=1): + rows.append( + { + **record, + "page": page, + "offset": offset, + "row": row_index, + "pool_group_id": pool["id"], + "pool_group_name": pool["name"], + "request_msg": req_i, + "request_time_utc": fmt_packet_time(req_ts), + "response_msg": resp_i, + "response_time_utc": fmt_packet_time(resp_ts), + "response_len": len(response), + "record_count": len(records), + } + ) + annotate_arc_groups(rows) + return rows + + +def make_arc_uid(timestamp_raw: str, ordinal: int, arc_key_hex: str) -> str: + source = "|".join([GAME_UID_PART, ARC_SYSTEM, ARC_BANNER_ID, timestamp_raw, str(ordinal), arc_key_hex]) + return hashlib.sha256(source.encode("utf-8")).hexdigest()[:32] + + +def annotate_arc_groups(rows: list[dict[str, Any]]) -> None: + groups: dict[str, list[int]] = defaultdict(list) + for index, row in enumerate(rows): + groups[row["timestamp_raw_hex"]].append(index) + + for group_index, (timestamp_raw, indexes) in enumerate(groups.items()): + complete = len(indexes) % 10 == 0 + for ordinal, index in enumerate(indexes): + row = rows[index] + row["timestamp_group_index"] = group_index + row["timestamp_group_ordinal"] = ordinal + row["timestamp_group_size_seen"] = len(indexes) + row["uid"] = make_arc_uid(timestamp_raw, ordinal, row["reward_key_hex"]) + row["uid_status"] = "stable" if complete else "skipped_incomplete_timestamp_group" + row["export_record"] = complete + row["skip_reason"] = "" if complete else "arc timestamp group is not a complete 10-pull in this capture" + + +def arc_stability_warnings(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: + warnings = [] + seen = set() + for row in rows: + if row.get("export_record") is True: + continue + timestamp_raw = row["timestamp_raw_hex"] + if timestamp_raw in seen: + continue + seen.add(timestamp_raw) + group = [r for r in rows if r["timestamp_raw_hex"] == timestamp_raw] + warnings.append( + { + "code": "INCOMPLETE_ARC_10_PULL_DROPPED", + "timestamp_raw": timestamp_raw, + "timestamp_decoded": row["timestamp_decoded"], + "records": len(group), + "reason": "arc timestamp group is not a complete 10-pull in this capture", + } + ) + return warnings + + +def select_continuous_arc_run(pairs: list[tuple]) -> tuple[list[tuple], list[dict[str, Any]]]: + warnings: list[dict[str, Any]] = [] + if not pairs: + return [], warnings + pairs_by_page = {pair[0]: pair for pair in pairs} + seen_pages = sorted(pairs_by_page) + if 1 in pairs_by_page: + selected_pages = [] + page = 1 + while page in pairs_by_page: + selected_pages.append(page) + page += 1 + if len(selected_pages) < len(seen_pages): + ignored = [page for page in seen_pages if page not in selected_pages] + warnings.append( + { + "code": "PAGE_GAP_DETECTED", + "ignored_pages": ignored, + "reason": f"Using continuous pages 1-{selected_pages[-1]}; ignored later pages {ignored}.", + } + ) + return [pairs_by_page[page] for page in selected_pages], warnings + + warnings.append({"code": "DID_NOT_START_AT_PAGE_1", "reason": "Arc history scan did not start at page 1."}) + return longest_monotonic_page_run(pairs), warnings diff --git a/src/nte_history_exporter/decoder/boundary.py b/src/nte_history_exporter/decoder/boundary.py new file mode 100644 index 0000000..003f3b6 --- /dev/null +++ b/src/nte_history_exporter/decoder/boundary.py @@ -0,0 +1,159 @@ +from __future__ import annotations + +import hashlib +from typing import Any + +from nte_history_exporter.constants import BANNER_ID, GAME_UID_PART, SYSTEM + + +def make_uid(record: dict[str, Any], ordinal: int) -> str: + source = "|".join( + [ + GAME_UID_PART, + SYSTEM, + str(record.get("pool_group_id", BANNER_ID)), + str(record.get("timestamp_raw_hex", "")), + str(ordinal), + str(record.get("dice", "")), + str(record.get("reward_key_hex", "")), + str(record.get("quantity", "")), + ] + ) + return hashlib.sha256(source.encode("utf-8")).hexdigest()[:32] + + +def longest_monotonic_page_run(pairs: list[tuple]) -> list[tuple]: + runs: list[list[tuple]] = [] + current: list[tuple] = [] + prev_page = None + for pair in pairs: + page = pair[0] + if prev_page is None or page == prev_page + 1: + current.append(pair) + else: + if current: + runs.append(current) + current = [pair] + prev_page = page + if current: + runs.append(current) + return max(runs, key=len) if runs else [] + + +def page_gap_warnings(pairs: list[tuple], best_run: list[tuple]) -> list[dict[str, Any]]: + warnings: list[dict[str, Any]] = [] + if len(pairs) < 2: + return warnings + + best_run_ids = {id(pair) for pair in best_run} + ignored_pages = [pair[0] for pair in pairs if id(pair) not in best_run_ids] + previous_page = pairs[0][0] + for pair in pairs[1:]: + page = pair[0] + if page != previous_page + 1: + warning = { + "code": "PAGE_GAP_DETECTED", + "previous_page": previous_page, + "next_page": page, + "ignored_pages": ignored_pages, + "reason": ( + f"Page gap detected: saw page {previous_page} then page {page}. " + "Pages outside the longest continuous run were ignored for stable dedupe. " + "Re-scan or scroll more slowly." + ), + } + warnings.append(warning) + previous_page = page + return warnings + + +def is_dice_record(row: dict[str, Any]) -> bool: + result_type = row.get("result_type") + if result_type: + return result_type == "dice" + + dice = row.get("dice") + if dice in ("", None): + return False + try: + return int(dice) > 0 + except (TypeError, ValueError): + return False + + +def annotate_groups( + rows: list[dict[str, Any]], + *, + starts_from_page_1: bool = True, + stable_only: bool = True, +) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + if not rows: + return rows, [] + + last_page = max(int(r["page"]) for r in rows if str(r.get("page", "")).isdigit()) + last_page_records = [r for r in rows if r.get("page") == last_page] + final_page_is_partial = len(last_page_records) < 5 + + groups: list[list[dict[str, Any]]] = [] + current: list[dict[str, Any]] = [] + prev_ts = None + for row in rows: + ts = row.get("timestamp_raw_hex", "") + if prev_ts is None or ts == prev_ts: + current.append(row) + else: + groups.append(current) + current = [row] + prev_ts = ts + if current: + groups.append(current) + + warnings: list[dict[str, Any]] = [] + for group_index, group in enumerate(groups): + at_newest_boundary = group_index == 0 + at_oldest_boundary = group_index == len(groups) - 1 + dice_records_in_group = [row for row in group if is_dice_record(row)] + dice_record_count = len(dice_records_in_group) + group_status = "stable" + skip_reason = "" + + if at_newest_boundary and not starts_from_page_1: + group_status = "dropped_boundary_group" + skip_reason = "newest timestamp group may be partial because scan did not start from page 1" + elif at_oldest_boundary and not final_page_is_partial: + group_status = "dropped_boundary_group" + skip_reason = "oldest timestamp group may continue onto the next uncaptured page" + + if group_status != "stable": + warnings.append( + { + "code": "PARTIAL_TIMESTAMP_GROUP_DROPPED", + "timestamp_raw": group[0].get("timestamp_raw_hex", ""), + "timestamp_decoded": group[0].get("timestamp_decoded", ""), + "records": len(group), + "dice_records": dice_record_count, + "reason": skip_reason, + } + ) + + for ordinal, row in enumerate(group): + row["timestamp_group_index"] = group_index + row["timestamp_group_ordinal"] = ordinal + row["timestamp_group_size_seen"] = dice_record_count + row["timestamp_group_record_size_seen"] = len(group) + row["timestamp_group_boundary"] = ",".join( + name + for name, yes in [("newest", at_newest_boundary), ("oldest", at_oldest_boundary)] + if yes + ) + if group_status == "stable": + row["uid_status"] = "stable" + row["uid"] = make_uid(row, ordinal) + row["export_record"] = True + row["skip_reason"] = "" + else: + row["uid_status"] = group_status + row["uid"] = "" if stable_only else make_uid(row, ordinal) + row["export_record"] = not stable_only + row["skip_reason"] = skip_reason + return rows, warnings diff --git a/src/nte_history_exporter/decoder/protocol.py b/src/nte_history_exporter/decoder/protocol.py new file mode 100644 index 0000000..99b4905 --- /dev/null +++ b/src/nte_history_exporter/decoder/protocol.py @@ -0,0 +1,179 @@ +from __future__ import annotations + +import struct +from datetime import datetime, timezone +from typing import Any + +from nte_history_exporter.constants import ( + DOTNET_UNIX_EPOCH_SECONDS, + HISTORY_REQUEST_BANNER, + HISTORY_REQUEST_LENGTH, + HISTORY_PAGE_CURSOR_MULTIPLIER, + LIMITED_CHARACTER_SELECTOR, + MARKERS, + PERMANENT_SELECTOR, + TIMESTAMP_TICKS_PER_SECOND, + VALID_DICE_FIELDS, +) +from nte_history_exporter.mappings import KNOWN_REWARDS + + +def decode_history_timestamp(raw8: bytes) -> tuple[int, float, str]: + if len(raw8) != 8: + raise ValueError("history timestamps must be exactly 8 bytes") + ticks = struct.unpack(" str: + if len(content) != HISTORY_REQUEST_LENGTH or struct.unpack_from(" bool: + return bool(history_request_kind(content)) + + +def request_page(content: bytes) -> int: + return struct.unpack_from(" bool: + return any(marker in content for marker in MARKERS) + + +def extract_key(chunk_without_marker: bytes) -> str: + fashion_prefix = bytes.fromhex("1885cda1a5bdb97d") + fashion_pos = chunk_without_marker.rfind(fashion_prefix) + if fashion_pos != -1: + return chunk_without_marker[fashion_pos:].hex() + + char_prefix = bytes.fromhex("c4c0") + char_pos = chunk_without_marker.find(char_prefix) + if char_pos != -1 and char_pos + 5 <= len(chunk_without_marker): + return chunk_without_marker[char_pos : char_pos + 5].hex() + + best = None + for prefix in [bytes.fromhex("98bdc9ad"), bytes.fromhex("10a58d95")]: + pos = chunk_without_marker.rfind(prefix) + if pos != -1 and (best is None or pos > best): + best = pos + return "" if best is None else chunk_without_marker[best:].hex() + + +def extract_dice(chunk_without_marker: bytes) -> tuple[int | None, int | None, int | None]: + for off in range(0, min(16, max(0, len(chunk_without_marker) - 3))): + val = struct.unpack_from(" tuple[str, int | None]: + if dice is None or dice_offset is None: + return "unknown", None + if dice == 0: + return "points_gift", 0 + + source_off = dice_offset + 4 + if source_off + 4 <= len(chunk_without_marker): + source_val = struct.unpack_from(" int | None: + if key_hex == "10a58d957dd1a58dad95d17dc1c400": + if result_type == "chase_reward": + return 30 + return 4 + if key_hex == "10a58d9539bdc9b585b101": + return 1 + if key_hex == "10a58d957dd1a58dad95d17dc1c800": + if "c8b0d4c0" in chunk_hex: + return 50 + if "c8b0ccc0" in chunk_hex: + return 30 + return None + if key_hex: + return 1 + return None + + +def decode_response_records(response_content: bytes) -> list[dict[str, Any]]: + marker = b"" + marker_offsets: list[int] = [] + for candidate_marker in MARKERS: + offsets = [i for i in range(len(response_content)) if response_content.startswith(candidate_marker, i)] + if offsets: + marker = candidate_marker + marker_offsets = offsets + break + if not marker_offsets: + return [] + + rows: list[dict[str, Any]] = [] + prev = 0x50 + for row_index, marker_offset in enumerate(marker_offsets, start=1): + chunk = response_content[prev:marker_offset] + full_record = response_content[prev : marker_offset + len(marker) + 8] + dice, dice_raw, dice_offset = extract_dice(chunk) + result_type, result_source_raw = classify_result_type(chunk, dice, dice_offset) + if result_type == "points_gift": + dice = 0 + dice_raw = 0 + elif result_type == "chase_reward": + dice = -4 + dice_raw = -4 + key_hex = extract_key(chunk) + reward = KNOWN_REWARDS.get(key_hex, {}) + 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() + rows.append( + { + "row": row_index, + "record_start": prev, + "record_end": marker_offset + len(marker) + 8, + "record_len": len(full_record), + "dice": dice, + "roll_result": ( + "Points Gift" + if result_type == "points_gift" + else ("Chase Reward" if result_type == "chase_reward" else (f"Dice {dice}" if dice else "")) + ), + "result_type": result_type, + "result_source_raw": result_source_raw, + "dice_raw_u32": dice_raw, + "dice_offset_in_record": dice_offset, + "reward_key_hex": key_hex, + "reward_type": reward.get("type", ""), + "reward_id": reward.get("id", ""), + "reward_name": reward.get("name", ""), + "reward_rank": reward.get("rank"), + "quantity": guess_quantity(chunk_hex, key_hex, result_type), + "timestamp_raw_hex": timestamp_raw.hex(), + "timestamp_ticks": timestamp_ticks, + "timestamp_unix": f"{timestamp_unix:.6f}", + "timestamp_decoded": timestamp_decoded, + "record_hex": full_record.hex(), + } + ) + prev = marker_offset + len(marker) + 8 + return rows diff --git a/src/nte_history_exporter/decoder/run.py b/src/nte_history_exporter/decoder/run.py new file mode 100644 index 0000000..cb421aa --- /dev/null +++ b/src/nte_history_exporter/decoder/run.py @@ -0,0 +1,57 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any + +from nte_history_exporter.constants import POOL_META +from nte_history_exporter.decoder.protocol import decode_response_records + + +def fmt_packet_time(ts: float | None) -> str: + if ts is None: + return "" + return datetime.fromtimestamp(ts, timezone.utc).strftime("%H:%M:%S.%f")[:-3] + + +def build_rows_from_pairs(pairs: list[tuple]) -> list[dict[str, Any]]: + rows_out: list[dict[str, Any]] = [] + for pair in pairs: + page, offset, req_i, req_ts, resp_i, resp_ts, response_content = pair[:7] + kind = pair[7] if len(pair) > 7 else "permanent" + pool = POOL_META.get(kind, POOL_META["permanent"]) + records = decode_response_records(response_content) + if not records: + rows_out.append( + { + "page": page, + "offset": offset, + "row": "", + "pool_group_id": pool["id"], + "pool_group_name": pool["name"], + "request_msg": req_i, + "request_time_utc": fmt_packet_time(req_ts), + "response_msg": resp_i, + "response_time_utc": fmt_packet_time(resp_ts), + "response_len": len(response_content), + "record_count": 0, + "record_hex": response_content.hex(), + } + ) + continue + for record in records: + rows_out.append( + { + "page": page, + "offset": offset, + "pool_group_id": pool["id"], + "pool_group_name": pool["name"], + "request_msg": req_i, + "request_time_utc": fmt_packet_time(req_ts), + "response_msg": resp_i, + "response_time_utc": fmt_packet_time(resp_ts), + "response_len": len(response_content), + "record_count": len(records), + **record, + } + ) + return rows_out diff --git a/src/nte_history_exporter/export/__init__.py b/src/nte_history_exporter/export/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/src/nte_history_exporter/export/__init__.py @@ -0,0 +1 @@ + diff --git a/src/nte_history_exporter/export/csv_export.py b/src/nte_history_exporter/export/csv_export.py new file mode 100644 index 0000000..1fc951a --- /dev/null +++ b/src/nte_history_exporter/export/csv_export.py @@ -0,0 +1,57 @@ +from __future__ import annotations + +import csv +from pathlib import Path +from typing import Any + +FIELDNAMES = [ + "uid", + "uid_status", + "export_record", + "skip_reason", + "page", + "offset", + "row", + "pool_group_id", + "pool_group_name", + "timestamp_group_index", + "timestamp_group_ordinal", + "timestamp_group_size_seen", + "timestamp_group_record_size_seen", + "timestamp_group_boundary", + "roll_result", + "result_type", + "result_source_raw", + "dice", + "dice_raw_u32", + "reward_type", + "reward_id", + "reward_name", + "reward_rank", + "quantity", + "timestamp_decoded", + "timestamp_raw_hex", + "timestamp_ticks", + "timestamp_unix", + "reward_key_hex", + "source_type", + "type_key_hex", + "request_msg", + "request_time_utc", + "response_msg", + "response_time_utc", + "response_len", + "record_count", + "record_start", + "record_end", + "record_len", + "dice_offset_in_record", + "record_hex", +] + + +def write_csv(path: str | Path, rows: list[dict[str, Any]]) -> None: + with Path(path).open("w", newline="", encoding="utf-8") as f: + writer = csv.DictWriter(f, fieldnames=FIELDNAMES, extrasaction="ignore") + writer.writeheader() + writer.writerows(rows) diff --git a/src/nte_history_exporter/export/json_export.py b/src/nte_history_exporter/export/json_export.py new file mode 100644 index 0000000..b1474da --- /dev/null +++ b/src/nte_history_exporter/export/json_export.py @@ -0,0 +1,88 @@ +from __future__ import annotations + +from typing import Any + +from nte_history_exporter import __version__ +from nte_history_exporter.constants import ( + ARC_BANNER_ID, + BANNER_ID, + EXPORTER_NAME, + GAME_NAME, + POOL_META, +) + + +def build_export_json( + rows: list[dict[str, Any]], + warnings: list[dict[str, Any]], + *, + source: str = "packet_capture", + flow_index: int | None = None, + candidate_request_response_pairs: int | None = None, + pages_seen: list[int] | None = None, +) -> dict[str, Any]: + exported = [r for r in rows if r.get("export_record") is True] + representative = exported[0] if exported else (rows[0] if rows else {}) + pool_group_id = representative.get("pool_group_id", BANNER_ID) + pool = next((meta for meta in POOL_META.values() if meta["id"] == pool_group_id), POOL_META["permanent"]) + scan: dict[str, Any] = { + "mode": "stable_only", + "boundary_policy": "drop_incomplete_timestamp_groups", + "decoded_records": len(rows), + "exported_records": len(exported), + "skipped_records": len(rows) - len(exported), + "warnings": warnings, + } + if flow_index is not None: + scan["udp_flow_index"] = flow_index + if candidate_request_response_pairs is not None: + scan["candidate_request_response_pairs"] = candidate_request_response_pairs + if pages_seen is not None: + scan["pages_seen"] = pages_seen + + return { + "format": "nte-history-export", + "format_version": 1, + "game": GAME_NAME, + "source": source, + "exporter": {"name": EXPORTER_NAME, "version": __version__}, + "banner": { + "id": pool["id"], + "name": pool["name"], + "system": pool["system"], + "shared_pity": pool["shared_pity"], + }, + "scan": scan, + "records": [_record_for_export(r) for r in exported], + } + + +def _record_for_export(row: dict[str, Any]) -> dict[str, Any]: + if row.get("pool_group_id") == ARC_BANNER_ID: + return { + "uid": row.get("uid"), + "pool_group_id": row.get("pool_group_id"), + "timestamp": row.get("timestamp_decoded"), + "timestamp_raw": row.get("timestamp_raw_hex"), + "timestamp_group_ordinal": row.get("timestamp_group_ordinal"), + "reward_type": row.get("reward_type"), + "reward_id": row.get("reward_id"), + "reward_name": row.get("reward_name"), + "reward_rank": row.get("reward_rank"), + "source_type": row.get("source_type"), + } + + return { + "uid": row.get("uid"), + "pool_group_id": row.get("pool_group_id", BANNER_ID), + "timestamp": row.get("timestamp_decoded"), + "timestamp_raw": row.get("timestamp_raw_hex"), + "timestamp_group_ordinal": row.get("timestamp_group_ordinal"), + "roll_result": row.get("dice"), + "result_type": row.get("result_type") or ("points_gift" if row.get("dice") == 0 else "dice"), + "reward_type": row.get("reward_type"), + "reward_id": row.get("reward_id"), + "reward_name": row.get("reward_name"), + "reward_rank": row.get("reward_rank"), + "quantity": row.get("quantity"), + } diff --git a/src/nte_history_exporter/live_capture/__init__.py b/src/nte_history_exporter/live_capture/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/src/nte_history_exporter/live_capture/__init__.py @@ -0,0 +1 @@ + diff --git a/src/nte_history_exporter/live_capture/runner.py b/src/nte_history_exporter/live_capture/runner.py new file mode 100644 index 0000000..dfd74d9 --- /dev/null +++ b/src/nte_history_exporter/live_capture/runner.py @@ -0,0 +1,140 @@ +from __future__ import annotations + +import json +import msvcrt +import socket +import time +from datetime import datetime +from pathlib import Path + +from nte_history_exporter.constants import POOL_META +from nte_history_exporter.decoder.arc import arc_stability_warnings, select_continuous_arc_run +from nte_history_exporter.decoder.boundary import annotate_groups, page_gap_warnings +from nte_history_exporter.export.csv_export import write_csv +from nte_history_exporter.export.json_export import build_export_json +from nte_history_exporter.live_capture.session import LiveHistorySession, UdpPacket +from nte_history_exporter.live_capture.windows_raw import detect_local_ipv4, open_raw_udp_socket, read_packets + + +EXPORT_PREFIXES = { + "permanent": "Permanent", + "limited_character": "Limited", + "arc_miracle_box": "Arc", +} + + +def copy_to_clipboard(text: str) -> None: + try: + import tkinter as tk + + root = tk.Tk() + root.withdraw() + root.clipboard_clear() + root.clipboard_append(text) + root.update() + root.destroy() + return + except Exception: + pass + + import subprocess + + subprocess.run(["clip"], input=text, text=True, check=True) + + +def run_live_capture( + *, + interface_ip: str | None = None, + copy_clipboard: bool = True, +) -> dict: + local_ip = interface_ip or detect_local_ipv4() + session = LiveHistorySession(local_ip) + + sock = open_raw_udp_socket(local_ip) + print(f"Device 0 ready~! Listening on {local_ip}") + print("Open either Monopoly history board. Capture will stay open until you press any key.") + + try: + for packet in read_packets(sock): + if msvcrt.kbhit(): + msvcrt.getch() + break + if packet is None: + continue + matched = session.process_packet( + UdpPacket( + timestamp=time.time(), + src_ip=packet.src_ip, + dst_ip=packet.dst_ip, + src_port=packet.src_port, + dst_port=packet.dst_port, + payload=packet.payload, + ) + ) + if matched: + kind = session.pairs[-1][7] if session.pairs else "permanent" + label = POOL_META.get(kind, POOL_META["permanent"])["name"] + print(f"Captured {label} page {session.last_page_seen}") + finally: + try: + sock.ioctl(socket.SIO_RCVALL, socket.RCVALL_OFF) + except Exception: + pass + sock.close() + + exports = [] + for kind in session.kinds_seen(): + pairs = session.pairs_for_kind(kind) + best_run = session.best_run(kind) + rows = session.build_rows(kind) + if kind == "arc_miracle_box": + _arc_run, gap_warnings = select_continuous_arc_run(pairs) + warnings = gap_warnings + arc_stability_warnings(rows) + else: + rows, warnings = annotate_groups(rows, starts_from_page_1=bool(best_run and best_run[0][0] == 1)) + warnings = page_gap_warnings(pairs, best_run) + warnings + pages_seen = [p[0] for p in best_run] + csv_path, json_path = export_paths(kind) + write_csv(csv_path, rows) + export = build_export_json( + rows, + warnings, + source="live_capture", + pages_seen=pages_seen, + ) + payload = json.dumps(export, ensure_ascii=False, indent=2) + json_path.write_text(payload, encoding="utf-8") + exports.append({"kind": kind, "csv_path": csv_path, "json_path": json_path, "export": export, "payload": payload}) + + if not exports: + print("No Monopoly history pages were captured.") + return {"exports": []} + + if copy_clipboard and len(exports) == 1: + payload = exports[0]["payload"] + copy_to_clipboard(payload) + print("Export copied to clipboard.") + elif len(exports) > 1: + print("Multiple banners captured; clipboard copy skipped.") + + for item in exports: + print(f"CSV written: {item['csv_path']}") + print(f"Export written: {item['json_path']}") + return {"exports": exports} + + +def export_paths(kind: str) -> tuple[Path, Path]: + export_dir = Path("exports") + export_dir.mkdir(parents=True, exist_ok=True) + prefix = EXPORT_PREFIXES.get(kind, "History") + stamp = datetime.now().strftime("%Y%m%d_%H%M%S") + base = export_dir / f"{prefix}_{stamp}" + csv_path = base.with_suffix(".csv") + json_path = base.with_suffix(".json") + counter = 2 + while csv_path.exists() or json_path.exists(): + base = export_dir / f"{prefix}_{stamp}_{counter}" + csv_path = base.with_suffix(".csv") + json_path = base.with_suffix(".json") + counter += 1 + return csv_path, json_path diff --git a/src/nte_history_exporter/live_capture/session.py b/src/nte_history_exporter/live_capture/session.py new file mode 100644 index 0000000..8d64241 --- /dev/null +++ b/src/nte_history_exporter/live_capture/session.py @@ -0,0 +1,153 @@ +from __future__ import annotations + +from collections import deque +from dataclasses import dataclass +from typing import Any + +from nte_history_exporter.decoder.arc import ( + arc_request_page, + build_arc_rows_from_pairs, + is_arc_history_request, + parse_arc_response, + select_continuous_arc_run, +) +from nte_history_exporter.decoder.boundary import longest_monotonic_page_run +from nte_history_exporter.decoder.protocol import ( + history_request_kind, + is_history_request, + request_page, + response_contains_history_marker, +) +from nte_history_exporter.decoder.run import build_rows_from_pairs + + +@dataclass +class UdpPacket: + timestamp: float + src_ip: str + dst_ip: str + src_port: int + dst_port: int + payload: bytes + + +@dataclass +class PendingRequest: + page: int + offset: int + kind: str + request_msg: int + request_time: float + src_ip: str + dst_ip: str + src_port: int + dst_port: int + + +class LiveHistorySession: + def __init__(self, local_ip: str) -> None: + self.local_ip = local_ip + self.pending: deque[PendingRequest] = deque() + self.pairs: list[tuple] = [] + self.packet_count = 0 + self.last_match_time: float | None = None + self.last_page_seen: int | None = None + + def process_packet(self, packet: UdpPacket) -> bool: + self.packet_count += 1 + + if packet.src_ip == self.local_ip and is_history_request(packet.payload): + offset = int.from_bytes(packet.payload[31:35], "little") + req = PendingRequest( + page=request_page(packet.payload), + offset=offset, + kind=history_request_kind(packet.payload), + request_msg=self.packet_count, + request_time=packet.timestamp, + src_ip=packet.src_ip, + dst_ip=packet.dst_ip, + src_port=packet.src_port, + dst_port=packet.dst_port, + ) + self.pending.append(req) + self.last_page_seen = req.page + return False + + if packet.src_ip == self.local_ip and is_arc_history_request(packet.payload): + page = arc_request_page(packet.payload) + req = PendingRequest( + page=page, + offset=page * 2, + kind="arc_miracle_box", + request_msg=self.packet_count, + request_time=packet.timestamp, + src_ip=packet.src_ip, + dst_ip=packet.dst_ip, + src_port=packet.src_port, + dst_port=packet.dst_port, + ) + self.pending.append(req) + self.last_page_seen = req.page + return False + + if packet.dst_ip != self.local_ip or len(packet.payload) < 100: + return False + + is_monopoly_response = response_contains_history_marker(packet.payload) + is_arc_response = bool(parse_arc_response(packet.payload)) + if not is_monopoly_response and not is_arc_response: + return False + + for req in self.pending: + if req.kind == "arc_miracle_box" and not is_arc_response: + continue + if req.kind != "arc_miracle_box" and not is_monopoly_response: + continue + if ( + packet.src_ip == req.dst_ip + and packet.dst_ip == req.src_ip + and packet.src_port == req.dst_port + and packet.dst_port == req.src_port + ): + self.pending.remove(req) + self.pairs.append( + ( + req.page, + req.offset, + req.request_msg, + req.request_time, + self.packet_count, + packet.timestamp, + packet.payload, + req.kind, + ) + ) + self.last_match_time = packet.timestamp + self.last_page_seen = req.page + return True + + return False + + def kinds_seen(self) -> list[str]: + seen = [] + for pair in self.pairs: + kind = pair[7] if len(pair) > 7 else "permanent" + if kind not in seen: + seen.append(kind) + return seen + + def pairs_for_kind(self, kind: str) -> list[tuple]: + return [pair for pair in self.pairs if (pair[7] if len(pair) > 7 else "permanent") == kind] + + def build_rows(self, kind: str | None = None) -> list[dict[str, Any]]: + if kind == "arc_miracle_box": + best_run, _warnings = select_continuous_arc_run(self.pairs_for_kind(kind)) + return build_arc_rows_from_pairs(best_run) + return build_rows_from_pairs(self.best_run(kind)) + + def best_run(self, kind: str | None = None) -> list[tuple]: + pairs = self.pairs_for_kind(kind) if kind else self.pairs + if kind == "arc_miracle_box": + best_run, _warnings = select_continuous_arc_run(pairs) + return best_run + return longest_monotonic_page_run(pairs) diff --git a/src/nte_history_exporter/live_capture/windows_raw.py b/src/nte_history_exporter/live_capture/windows_raw.py new file mode 100644 index 0000000..318fd38 --- /dev/null +++ b/src/nte_history_exporter/live_capture/windows_raw.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +import socket +import struct +from dataclasses import dataclass + +from nte_history_exporter.live_capture.session import UdpPacket + + +@dataclass +class ParsedIpUdpPacket: + src_ip: str + dst_ip: str + src_port: int + dst_port: int + payload: bytes + + +def detect_local_ipv4() -> str: + candidates: list[str] = [] + try: + host = socket.gethostname() + for ip in socket.gethostbyname_ex(host)[2]: + if not ip.startswith("127."): + candidates.append(ip) + except OSError: + pass + + try: + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as probe: + probe.connect(("1.1.1.1", 80)) + ip = probe.getsockname()[0] + if not ip.startswith("127."): + candidates.insert(0, ip) + except OSError: + pass + + if not candidates: + raise RuntimeError("could not determine a usable local IPv4 address") + return candidates[0] + + +def parse_ipv4_udp_packet(data: bytes) -> ParsedIpUdpPacket | None: + if len(data) < 28: + return None + version_ihl = data[0] + if version_ihl >> 4 != 4: + return None + ihl = (version_ihl & 0x0F) * 4 + if len(data) < ihl + 8: + return None + protocol = data[9] + if protocol != 17: + return None + + src_ip = socket.inet_ntoa(data[12:16]) + dst_ip = socket.inet_ntoa(data[16:20]) + src_port, dst_port, udp_len, _checksum = struct.unpack_from("!HHHH", data, ihl) + payload = data[ihl + 8 : ihl + udp_len] + return ParsedIpUdpPacket(src_ip, dst_ip, src_port, dst_port, payload) + + +def open_raw_udp_socket(local_ip: str) -> socket.socket: + sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_IP) + sock.bind((local_ip, 0)) + sock.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1) + sock.ioctl(socket.SIO_RCVALL, socket.RCVALL_ON) + sock.settimeout(0.5) + return sock + + +def read_packets(sock: socket.socket): + while True: + try: + data, _addr = sock.recvfrom(65535) + except socket.timeout: + yield None + continue + packet = parse_ipv4_udp_packet(data) + if packet is None: + continue + yield packet diff --git a/src/nte_history_exporter/mappings.py b/src/nte_history_exporter/mappings.py new file mode 100644 index 0000000..bfa595f --- /dev/null +++ b/src/nte_history_exporter/mappings.py @@ -0,0 +1,92 @@ +KNOWN_REWARDS = { + "98bdc9ad7d91d5cdd189a5b901": {"type": "arc", "id": "fork_dustbin", "name": "Dangerous Game", "rank": "B"}, + "98bdc9ad7dd9a5b99501": {"type": "arc", "id": "fork_vine", "name": "Be Happy", "rank": "B"}, + "98bdc9ad7db9bdb9bdcd01": {"type": "arc", "id": "fork_nonos", "name": "First Step to Success", "rank": "B"}, + "98bdc9ad7d85c1c1b1a585b98d9501": {"type": "arc", "id": "fork_appliance", "name": "\"Real Music\"", "rank": "B"}, + "98bdc9ad7d41c9bdad85c9e5bdb901": {"type": "arc", "id": "fork_Prokaryon", "name": "Us.", "rank": "B"}, + "98bdc9ad7d4185c195c941b185b99501": {"type": "arc", "id": "fork_PaperPlane", "name": "Clear Skies", "rank": "A"}, + "98bdc9ad7ddda1d585add585b99d01": {"type": "arc", "id": "fork_wuhuakuang", "name": "The Forgotten", "rank": "A"}, + "98bdc9ad7dddd5a1d585add585b99d01": {"type": "arc", "id": "fork_wuhuakuang", "name": "The Forgotten", "rank": "A"}, + "98bdc9ad7d2da5d19501": {"type": "arc", "id": "fork_Kite", "name": "Watch Your Heads!", "rank": "A"}, + "98bdc9ad7de5d5c995b901": {"type": "arc", "id": "fork_yuren", "name": "Umbrella", "rank": "A"}, + "98bdc9ad7de585bd9185bd01": {"type": "arc", "id": "fork_yaodao", "name": "Drawn Blade", "rank": "A"}, + "10a58d9539bdc9b585b101": {"type": "item", "id": "DiceNormal", "name": "Fabricated Dice", "rank": None}, + "10a58d957dd1a58dad95d17dc1c400": {"type": "item", "id": "Dice_ticket_01", "name": "Warp Piece", "rank": None}, + "10a58d957dd1a58dad95d17dc1c800": {"type": "item", "id": "Dice_ticket_02", "name": "Lost Piece", "rank": None}, + "10a58d95b1a5b5a5d19501": {"type": "item", "id": "Dicelimite", "name": "Solid Dice", "rank": None}, + "1885cda1a5bdb97dd995a1a58db1957dc5c0c4c07c59c1c0e000": { + "type": "cosmetic", + "id": "Fashion_vehicle_1010_V008", + "name": "Fashion Vehicle Skin 1010 V008", + "rank": None, + }, + "1885cda1a5bdb97d1db1a591957dc5c0c4c000": { + "type": "cosmetic", + "id": "Fashion_glide_1010", + "name": "Fashion Glider 1010", + "rank": None, + }, + "c4c0cccc00": {"type": "character", "id": "1033", "name": "Adler", "rank": "A"}, + "c4c0c8c000": {"type": "character", "id": "1020", "name": "Haniel", "rank": "A"}, + "c4c0c8cc00": {"type": "character", "id": "1023", "name": "Baicang", "rank": "S"}, + "c4c0c8d400": {"type": "character", "id": "1025", "name": "Hathor", "rank": "S"}, + "c4c0c4e400": {"type": "character", "id": "1019", "name": "Mint", "rank": "A"}, + "c4c0c0e000": {"type": "character", "id": "1008", "name": "Skia", "rank": "A"}, + "c4c0dcc0": {"type": "character", "id": "1070", "name": "Aurelia", "rank": "A"}, + "c4c0dcc000": {"type": "character", "id": "1070", "name": "Aurelia", "rank": "A"}, + "c4c0c8c4": {"type": "character", "id": "1021", "name": "Edgar", "rank": "A"}, + "c4c0c8c400": {"type": "character", "id": "1021", "name": "Edgar", "rank": "A"}, +} + +ARC_META = { + "fork_nonos": {"name": "First Step to Success", "rank": "B"}, + "fork_Prokaryon": {"name": "Us.", "rank": "B"}, + "fork_appliance": {"name": "\"Real Music\"", "rank": "B"}, + "fork_PaperPlane": {"name": "Clear Skies", "rank": "A"}, + "fork_mofeikesi": {"name": "Good Boy's Grand Adventure", "rank": "S"}, + "fork_dustbin": {"name": "Dangerous Game", "rank": "B"}, + "fork_vine": {"name": "Be Happy", "rank": "B"}, + "fork_Kite": {"name": "Watch Your Heads!", "rank": "A"}, + "fork_wuhuakuang": {"name": "The Forgotten", "rank": "A"}, + "fork_jingmotingyuan": {"name": "Camellia Society", "rank": "S"}, + "fork_yuren": {"name": "Umbrella", "rank": "A"}, + "fork_yaodao": {"name": "Drawn Blade", "rank": "A"}, +} + +CHARACTERS = { + "1003": {"name": "Sakiri", "rank": "S"}, + "1004": {"name": "Lacrimosa", "rank": "S"}, + "1008": {"name": "Skia", "rank": "A"}, + "1010": {"name": "Nanally", "rank": "S"}, + "1019": {"name": "Mint", "rank": "A"}, + "1020": {"name": "Haniel", "rank": "A"}, + "1021": {"name": "Edgar", "rank": "A"}, + "1023": {"name": "Baicang", "rank": "S"}, + "1025": {"name": "Hathor", "rank": "S"}, + "1033": {"name": "Adler", "rank": "A"}, + "1039": {"name": "Fadia", "rank": "S"}, + "1046": {"name": "Zero", "rank": "S"}, + "1051": {"name": "Zero", "rank": "S"}, + "1052": {"name": "Hotori", "rank": "S"}, + "1054": {"name": "Daffodill", "rank": "S"}, + "1055": {"name": "Jiuyuan", "rank": "S"}, + "1056": {"name": "Lacrimosa", "rank": "S"}, + "1070": {"name": "Aurelia", "rank": "A"}, + "1073": {"name": "Chiz", "rank": "S"}, +} + + +def character_id_to_key(character_id: str) -> str: + return "".join(f"{0xC0 + int(digit) * 4:02x}" for digit in character_id) + "00" + + +for _character_id, _info in CHARACTERS.items(): + KNOWN_REWARDS.setdefault( + character_id_to_key(_character_id), + { + "type": "character", + "id": _character_id, + "name": _info["name"], + "rank": _info["rank"], + }, + ) diff --git a/src/nte_history_exporter/pool_mappings.py b/src/nte_history_exporter/pool_mappings.py new file mode 100644 index 0000000..20ff705 --- /dev/null +++ b/src/nte_history_exporter/pool_mappings.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +MAPPINGS_DIR = PROJECT_ROOT / "mappings" + +POOL_MAPPING_FILES = { + "permanent": "permanent_board.json", + "limited_character": "limited_character_board.json", + "arc_miracle_box": "arc_miracle_box.json", +} + + +def load_pool_mapping(pool_key: str) -> dict[str, Any]: + filename = POOL_MAPPING_FILES[pool_key] + return json.loads((MAPPINGS_DIR / filename).read_text(encoding="utf-8")) + + +def load_pool_mappings() -> dict[str, dict[str, Any]]: + return {pool_key: load_pool_mapping(pool_key) for pool_key in POOL_MAPPING_FILES} + + +def pool_meta_from_mapping(mapping: dict[str, Any]) -> dict[str, Any]: + return { + "id": mapping["banner"]["id"], + "name": mapping["banner"]["name"], + "system": mapping["system"]["name"], + "shared_pity": mapping["banner"]["shared_pity"], + } diff --git a/tests/test_boundary_export.py b/tests/test_boundary_export.py new file mode 100644 index 0000000..6d1ee34 --- /dev/null +++ b/tests/test_boundary_export.py @@ -0,0 +1,402 @@ +import csv +import json +import sys +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +SRC = ROOT / "src" +EXPORTS = ROOT / "exports" +V7_EXPORTS = EXPORTS / "1" +ARC_EXPORTS = EXPORTS / "arc" +if str(SRC) not in sys.path: + sys.path.insert(0, str(SRC)) + +from nte_history_exporter.decoder.boundary import annotate_groups, make_uid +from nte_history_exporter.constants import LIMITED_CHARACTER_MARKER, MARKER +from nte_history_exporter.decoder.boundary import page_gap_warnings +from nte_history_exporter.decoder.protocol import decode_response_records, history_request_kind +from nte_history_exporter.constants import POOL_META +from nte_history_exporter.mappings import ARC_META, CHARACTERS +from nte_history_exporter.decoder.arc import ( + build_arc_rows_from_pairs, + decode_arc_key, + decode_arc_timestamp, + make_arc_uid, + parse_arc_response, +) +from nte_history_exporter.live_capture.session import LiveHistorySession, UdpPacket +from nte_history_exporter.export.json_export import build_export_json +from nte_history_exporter.pool_mappings import load_pool_mappings, pool_meta_from_mapping + + +def load_reference_csv(name): + path = EXPORTS / name + if not path.exists(): + raise unittest.SkipTest(f"private reference fixture not present: {path}") + rows = [] + with path.open(newline="", encoding="utf-8") as f: + for row in csv.DictReader(f): + normalized = dict(row) + for key in ["page", "offset", "row", "dice", "dice_raw_u32", "quantity"]: + if normalized.get(key) not in ("", None): + normalized[key] = int(normalized[key]) + rows.append(normalized) + return rows + + +def decode_single_record(record_hex): + return decode_response_records(bytes(0x50) + bytes.fromhex(record_hex))[0] + + +def load_v7_row(name, one_based_index): + path = V7_EXPORTS / name + if not path.exists(): + raise unittest.SkipTest(f"private reference fixture not present: {path}") + with path.open(newline="", encoding="utf-8") as f: + rows = list(csv.DictReader(f)) + return rows[one_based_index - 1] + + +def load_arc_csv(name): + path = ARC_EXPORTS / name + if not path.exists(): + raise unittest.SkipTest(f"private reference fixture not present: {path}") + with path.open(newline="", encoding="utf-8") as f: + return list(csv.DictReader(f)) + + +class BoundaryExportTests(unittest.TestCase): + def test_pool_mapping_json_files_have_uniform_shape(self): + required_top_level = {"pool_key", "game", "system", "banner", "request", "response"} + for pool_key, mapping in load_pool_mappings().items(): + with self.subTest(pool_key=pool_key): + self.assertEqual(set(required_top_level) - set(mapping), set()) + self.assertEqual(mapping["pool_key"], pool_key) + self.assertIn("id", mapping["system"]) + self.assertIn("name", mapping["system"]) + self.assertIn("id", mapping["banner"]) + self.assertIn("name", mapping["banner"]) + self.assertIn("shared_pity", mapping["banner"]) + self.assertIn("family", mapping["request"]) + self.assertIn("length", mapping["request"]) + self.assertIn("constant", mapping["request"]) + self.assertIn("cursor_step", mapping["request"]) + + def test_pool_mapping_json_matches_runtime_pool_meta(self): + for pool_key, mapping in load_pool_mappings().items(): + with self.subTest(pool_key=pool_key): + self.assertEqual(pool_meta_from_mapping(mapping), POOL_META[pool_key]) + + def test_reward_mapping_json_matches_runtime_mappings(self): + with (ROOT / "mappings" / "arcs.json").open(encoding="utf-8") as f: + arc_json = json.load(f) + with (ROOT / "mappings" / "characters.json").open(encoding="utf-8") as f: + character_json = json.load(f) + + self.assertEqual(arc_json, ARC_META) + self.assertEqual(character_json, CHARACTERS) + + def test_uid_source_matches_v4_reference(self): + rows = load_reference_csv("monopoly_history_poc_10_all_44_pages_v4.csv") + first = rows[0] + self.assertEqual(make_uid(first, 0), "5adcf52282e15445466863b271f3b745") + + def test_uid_uses_detected_pool_group_id(self): + row = { + "pool_group_id": "Lottery_LimitedCharacter", + "timestamp_raw_hex": "40e93247c3097b23", + "dice": 5, + "reward_key_hex": "10a58d957dd1a58dad95d17dc1c800", + "quantity": 50, + } + self.assertNotEqual(make_uid(row, 0), "5adcf52282e15445466863b271f3b745") + + def test_pages_1_to_5_drops_oldest_boundary_group(self): + rows = load_reference_csv("monopoly_history_poc_13_pages_1_to_5_v4.csv") + annotated, warnings = annotate_groups(rows, starts_from_page_1=True) + + exported = [row for row in annotated if row["export_record"] is True] + skipped = [row for row in annotated if row["export_record"] is False] + + self.assertEqual(len(annotated), 25) + self.assertEqual(len(exported), 14) + self.assertEqual(len(skipped), 11) + self.assertEqual(len(warnings), 1) + self.assertEqual(warnings[0]["timestamp_raw"], "00dfdaa9c1097b23") + self.assertEqual(warnings[0]["records"], 11) + self.assertEqual(warnings[0]["dice_records"], 10) + + def test_full_reference_scan_exports_all_rows(self): + rows = load_reference_csv("monopoly_history_poc_10_all_44_pages_v4.csv") + annotated, warnings = annotate_groups(rows, starts_from_page_1=True) + exported = [row for row in annotated if row["export_record"] is True] + + json_path = EXPORTS / "monopoly_history_export_10_all_44_pages_v4.json" + if not json_path.exists(): + raise unittest.SkipTest(f"private reference fixture not present: {json_path}") + with json_path.open(encoding="utf-8") as f: + reference = json.load(f) + + self.assertEqual(len(annotated), reference["scan"]["decoded_records"]) + self.assertEqual(len(exported), reference["scan"]["exported_records"]) + self.assertEqual(warnings, []) + + def test_sanitized_export_omits_raw_packet_fields(self): + rows = load_reference_csv("monopoly_history_poc_13_pages_1_to_5_v4.csv") + annotated, warnings = annotate_groups(rows, starts_from_page_1=True) + export = build_export_json(annotated, warnings) + + self.assertEqual(export["format"], "nte-history-export") + self.assertIn("exporter", export) + self.assertNotIn("record_hex", export["records"][0]) + self.assertNotIn("request_msg", export["records"][0]) + self.assertNotIn("response_msg", export["records"][0]) + + def test_limited_selector_and_marker_decode(self): + request = bytearray(45) + request[31:35] = (4).to_bytes(4, "little") + request[35:39] = (4220).to_bytes(4, "little") + request[40:44] = (8).to_bytes(4, "little") + self.assertEqual(history_request_kind(bytes(request)), "limited_character") + + response = bytearray(220) + response[0x50:0x54] = (4).to_bytes(4, "little") + response[0x54:0x58] = (20).to_bytes(4, "little") + response[0x58:0x5d] = bytes.fromhex("c4c0c4c000") + marker_offset = 0x5d + response[marker_offset:marker_offset + len(LIMITED_CHARACTER_MARKER)] = LIMITED_CHARACTER_MARKER + timestamp_raw = (2556647947780680000).to_bytes(8, "little") + response[marker_offset + len(LIMITED_CHARACTER_MARKER):marker_offset + len(LIMITED_CHARACTER_MARKER) + 8] = timestamp_raw + + rows = decode_response_records(bytes(response)) + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0]["reward_id"], "1010") + self.assertEqual(rows[0]["reward_name"], "Nanally") + + def test_v7_prefixed_points_gift_rows_override_visible_dice(self): + cases = [ + ("limited_all_04_v7.csv", 41, "1020"), + ("monopoly_history_poc_10_all_44_pages_v7.csv", 31, "1033"), + ("monopoly_history_poc_10_all_44_pages_v7.csv", 86, "fork_PaperPlane"), + ("monopoly_history_poc_10_all_44_pages_v7.csv", 141, "fork_Kite"), + ("monopoly_history_poc_10_all_44_pages_v7.csv", 196, "1021"), + ] + for filename, row_index, reward_id in cases: + with self.subTest(filename=filename, row_index=row_index): + reference = load_v7_row(filename, row_index) + decoded = decode_single_record(reference["record_hex"]) + self.assertEqual(decoded["result_type"], "points_gift") + self.assertEqual(decoded["result_source_raw"], 0) + self.assertEqual(decoded["dice"], 0) + self.assertEqual(decoded["dice_raw_u32"], 0) + self.assertEqual(decoded["reward_id"], reward_id) + + def test_v7_prefixed_chase_reward_overrides_visible_dice_and_quantity(self): + reference = load_v7_row("limited_all_04_v7.csv", 61) + decoded = decode_single_record(reference["record_hex"]) + decoded.update( + { + "pool_group_id": "Lottery_LimitedCharacter", + "timestamp_group_ordinal": int(reference["timestamp_group_ordinal"]), + } + ) + + self.assertEqual(decoded["result_type"], "chase_reward") + self.assertEqual(decoded["result_source_raw"], -4) + self.assertEqual(decoded["dice"], -4) + self.assertEqual(decoded["dice_raw_u32"], -4) + self.assertEqual(decoded["reward_id"], "Dice_ticket_01") + self.assertEqual(decoded["quantity"], 30) + self.assertEqual(make_uid(decoded, int(reference["timestamp_group_ordinal"])), "7d035ec098f856f81b403ea538810145") + + def test_page_gap_warning_reports_ignored_pages(self): + pairs = [(1,), (2,), (3,), (5,)] + best_run = [(1,), (2,), (3,)] + warnings = page_gap_warnings(pairs, best_run) + self.assertEqual(warnings[0]["code"], "PAGE_GAP_DETECTED") + self.assertEqual(warnings[0]["previous_page"], 3) + self.assertEqual(warnings[0]["next_page"], 5) + self.assertEqual(warnings[0]["ignored_pages"], [5]) + + def test_arc_key_timestamp_and_uid_match_reference(self): + row = load_arc_csv("arc_pull_10_all_pages_v2.csv")[0] + self.assertEqual(decode_arc_key(bytes.fromhex(row["arc_key_hex"])), "fork_nonos") + _ticks, _unix, decoded = decode_arc_timestamp(bytes.fromhex(row["timestamp_raw_hex"])) + self.assertEqual(decoded, "2026-06-10 23:46:29") + self.assertEqual( + make_arc_uid(row["timestamp_raw_hex"], int(row["timestamp_group_ordinal"]), row["arc_key_hex"]), + row["uid"], + ) + + def test_arc_response_parser_matches_reference_first_page(self): + reference_rows = load_arc_csv("arc_pull_10_all_pages_v2.csv")[:5] + response = bytes(0x4C) + b"".join(bytes.fromhex(row["record_hex"]) for row in reference_rows) + decoded = parse_arc_response(response) + self.assertEqual(len(decoded), 5) + self.assertEqual([row["reward_id"] for row in decoded[:3]], ["fork_nonos", "fork_nonos", "fork_Prokaryon"]) + self.assertEqual(decoded[0]["reward_type"], "arc") + self.assertEqual(decoded[0]["reward_key_hex"], reference_rows[0]["arc_key_hex"]) + + def test_arc_partial_timestamp_group_is_skipped(self): + rows = load_arc_csv("arc_pages_1_to_5_v2.csv") + pairs = [] + for page in range(1, 6): + page_rows = [row for row in rows if int(row["page"]) == page] + response = bytes(0x4C) + b"".join(bytes.fromhex(row["record_hex"]) for row in page_rows) + pairs.append((page, page * 2, page, 1.0, page + 100, 1.1, response)) + decoded = build_arc_rows_from_pairs(pairs) + exported = [row for row in decoded if row["export_record"] is True] + skipped = [row for row in decoded if row["export_record"] is False] + self.assertEqual(len(decoded), 25) + self.assertEqual(len(exported), 20) + self.assertEqual(len(skipped), 5) + + def test_arc_row_builder_accepts_live_pairs_with_kind(self): + reference_rows = load_arc_csv("arc_pull_10_all_pages_v2.csv")[:5] + response = bytes(0x4C) + b"".join(bytes.fromhex(row["record_hex"]) for row in reference_rows) + decoded = build_arc_rows_from_pairs([(1, 2, 1, 1.0, 2, 1.1, response, "arc_miracle_box")]) + self.assertEqual(len(decoded), 5) + self.assertEqual(decoded[0]["reward_id"], "fork_nonos") + + def test_arc_export_is_shared_pity(self): + reference_rows = load_arc_csv("arc_pull_10_all_pages_v2.csv")[:10] + response = bytes(0x4C) + b"".join(bytes.fromhex(row["record_hex"]) for row in reference_rows) + rows = build_arc_rows_from_pairs([(1, 2, 1, 1.0, 2, 1.1, response, "arc_miracle_box")]) + export = build_export_json(rows, []) + self.assertEqual(export["banner"]["id"], "Arc_MiracleBox") + self.assertIs(export["banner"]["shared_pity"], True) + + def test_group_detection_counts_only_dice_records_but_uid_ordinals_keep_all_rows(self): + rows = [ + { + "page": 1, + "timestamp_raw_hex": "aa", + "timestamp_decoded": "2026-01-01 00:00:00", + "result_type": "dice", + "dice": 1, + "reward_key_hex": "k1", + "quantity": 1, + }, + { + "page": 1, + "timestamp_raw_hex": "aa", + "timestamp_decoded": "2026-01-01 00:00:00", + "result_type": "points_gift", + "dice": 0, + "reward_key_hex": "k2", + "quantity": 1, + }, + { + "page": 1, + "timestamp_raw_hex": "aa", + "timestamp_decoded": "2026-01-01 00:00:00", + "result_type": "chase_reward", + "dice": -4, + "reward_key_hex": "k3", + "quantity": 30, + }, + { + "page": 1, + "timestamp_raw_hex": "aa", + "timestamp_decoded": "2026-01-01 00:00:00", + "result_type": "dice", + "dice": 2, + "reward_key_hex": "k4", + "quantity": 1, + }, + { + "page": 1, + "timestamp_raw_hex": "bb", + "timestamp_decoded": "2026-01-01 00:01:00", + "result_type": "dice", + "dice": 3, + "reward_key_hex": "k5", + "quantity": 1, + }, + ] + + annotated, warnings = annotate_groups(rows, starts_from_page_1=True) + + self.assertEqual([row["timestamp_group_ordinal"] for row in annotated[:4]], [0, 1, 2, 3]) + self.assertEqual({row["timestamp_group_size_seen"] for row in annotated[:4]}, {2}) + self.assertEqual({row["timestamp_group_record_size_seen"] for row in annotated[:4]}, {4}) + self.assertEqual(warnings[0]["dice_records"], 1) + self.assertEqual(warnings[0]["records"], 1) + + def test_live_session_pairs_request_and_response(self): + session = LiveHistorySession("192.168.0.10") + + request = bytearray(45) + request[31:35] = (4).to_bytes(4, "little") + request[35:39] = (4220).to_bytes(4, "little") + request[40:44] = (4).to_bytes(4, "little") + + response = bytearray(220) + response[0x50:0x50 + len(MARKER)] = MARKER + response[0x50 + len(MARKER):0x50 + len(MARKER) + 8] = (1).to_bytes(8, "little") + + self.assertFalse( + session.process_packet( + UdpPacket( + timestamp=1.0, + src_ip="192.168.0.10", + dst_ip="203.0.113.5", + src_port=50000, + dst_port=40000, + payload=bytes(request), + ) + ) + ) + self.assertTrue( + session.process_packet( + UdpPacket( + timestamp=1.2, + src_ip="203.0.113.5", + dst_ip="192.168.0.10", + src_port=40000, + dst_port=50000, + payload=bytes(response), + ) + ) + ) + self.assertEqual(len(session.pairs), 1) + self.assertEqual(session.last_page_seen, 1) + + def test_live_session_ignores_non_history_udp_packets(self): + session = LiveHistorySession("192.168.0.10") + + request = bytearray(45) + request[31:35] = (4).to_bytes(4, "little") + request[35:39] = (4220).to_bytes(4, "little") + request[40:44] = (4).to_bytes(4, "little") + + noise = b"not-a-history-response" * 20 + + session.process_packet( + UdpPacket( + timestamp=1.0, + src_ip="192.168.0.10", + dst_ip="203.0.113.5", + src_port=50000, + dst_port=40000, + payload=bytes(request), + ) + ) + self.assertFalse( + session.process_packet( + UdpPacket( + timestamp=1.1, + src_ip="203.0.113.5", + dst_ip="192.168.0.10", + src_port=40000, + dst_port=50000, + payload=noise, + ) + ) + ) + self.assertEqual(len(session.pairs), 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/decode_mitmproxy_flows.py b/tools/decode_mitmproxy_flows.py new file mode 100644 index 0000000..44bd9c5 --- /dev/null +++ b/tools/decode_mitmproxy_flows.py @@ -0,0 +1,13 @@ +from pathlib import Path +import sys + +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.cli import main + + +if __name__ == "__main__": + raise SystemExit(main())