goober/nte-exporter
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Initial NTE history exporter

Browse source 
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.
This commit is contained in:
Golumpa 2026-06-10 23:59:01 +01:00
commit bb3fe3b19a
35 changed files with 2439 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

1
src/nte_history_exporter/__init__.py Normal file
View file

@ -0,0 +1 @@
__version__ = "0.1.0"

1
src/nte_history_exporter/adapters/__init__.py Normal file
View file

@ -0,0 +1 @@

146
src/nte_history_exporter/adapters/mitmproxy_flows.py Normal file
View file

@ -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("<I", content, 31)[0]
page = request_page(content)
response_index, response_content, response_ts = pair_response(messages, i)
if response_index is not None:
pairs.append((page, offset, i, ts, response_index, response_ts, response_content, kind))
best_run = longest_monotonic_page_run(pairs)
rows_out = build_rows_from_pairs(best_run)
best_arc_run, arc_warnings = select_continuous_arc_run(arc_pairs)
arc_rows = build_arc_rows_from_pairs(best_arc_run)
return {
"flow_index": resolved_flow_index,
"pairs": pairs,
"best_run": best_run,
"rows": rows_out,
"arc_pairs": arc_pairs,
"best_arc_run": best_arc_run,
"arc_rows": arc_rows,
"arc_warnings": arc_warnings,
"starts_from_page_1": bool(best_run and best_run[0][0] == 1),
}

99
src/nte_history_exporter/cli.py Normal file
View file

@ -0,0 +1,99 @@
from __future__ import annotations
import argparse
import json
from nte_history_exporter.adapters.mitmproxy_flows import decode_mitmproxy_flows
from nte_history_exporter.decoder.arc import arc_stability_warnings
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.runner import export_paths, run_live_capture
def build_parser() -> 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())

62
src/nte_history_exporter/constants.py Normal file
View file

@ -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"),
]

1
src/nte_history_exporter/decoder/__init__.py Normal file
View file

@ -0,0 +1 @@

208
src/nte_history_exporter/decoder/arc.py Normal file
View file

@ -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("<I", content, 24)[0] == ARC_HISTORY_REQUEST_BANNER
def arc_request_page(content: bytes) -> int:
return struct.unpack_from("<I", content, ARC_HISTORY_CURSOR_OFFSET)[0] // ARC_HISTORY_PAGE_CURSOR_MULTIPLIER
def decode_arc_key(raw: bytes) -> 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("<Q", raw8)[0]
unix_seconds = ticks / ARC_TIMESTAMP_TICKS_PER_SECOND - DOTNET_UNIX_EPOCH_SECONDS
decoded = datetime.fromtimestamp(unix_seconds, timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
return ticks, unix_seconds, decoded
def parse_arc_response(response: bytes) -> 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("<I", response, pos)[0]
pos += 4
if name_len2 <= 0 or name_len2 > 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("<I", response, pos)[0]
pos += 4
if type_len2 <= 0 or type_len2 > 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

159
src/nte_history_exporter/decoder/boundary.py Normal file
View file

@ -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

179
src/nte_history_exporter/decoder/protocol.py Normal file
View file

@ -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("<Q", raw8)[0]
unix_seconds = ticks / TIMESTAMP_TICKS_PER_SECOND - DOTNET_UNIX_EPOCH_SECONDS
decoded = datetime.fromtimestamp(unix_seconds, timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
return ticks, unix_seconds, decoded
def history_request_kind(content: bytes) -> str:
if len(content) != HISTORY_REQUEST_LENGTH or struct.unpack_from("<I", content, 35)[0] != HISTORY_REQUEST_BANNER:
return ""
selector = struct.unpack_from("<I", content, 40)[0]
if selector == PERMANENT_SELECTOR:
return "permanent"
if selector == LIMITED_CHARACTER_SELECTOR:
return "limited_character"
return ""
def is_history_request(content: bytes) -> bool:
return bool(history_request_kind(content))
def request_page(content: bytes) -> int:
return struct.unpack_from("<I", content, 31)[0] // HISTORY_PAGE_CURSOR_MULTIPLIER
def response_contains_history_marker(content: bytes) -> 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("<I", chunk_without_marker, off)[0]
if val in VALID_DICE_FIELDS:
return (0 if val == 0 else val // 4), val, off
return None, None, None
def classify_result_type(
chunk_without_marker: bytes,
dice: int | None,
dice_offset: int | None,
) -> 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("<i", chunk_without_marker, source_off)[0]
if source_val == 0:
return "points_gift", source_val
if source_val == -4:
return "chase_reward", source_val
return "dice", source_val
return "dice", None
def guess_quantity(chunk_hex: str, key_hex: str, result_type: str | None = None) -> 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

57
src/nte_history_exporter/decoder/run.py Normal file
View file

@ -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

1
src/nte_history_exporter/export/__init__.py Normal file
View file

@ -0,0 +1 @@

57
src/nte_history_exporter/export/csv_export.py Normal file
View file

@ -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)

88
src/nte_history_exporter/export/json_export.py Normal file
View file

@ -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"),
}

1
src/nte_history_exporter/live_capture/__init__.py Normal file
View file

@ -0,0 +1 @@

140
src/nte_history_exporter/live_capture/runner.py Normal file
View file

@ -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

153
src/nte_history_exporter/live_capture/session.py Normal file
View file

@ -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)

82
src/nte_history_exporter/live_capture/windows_raw.py Normal file
View file

@ -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

92
src/nte_history_exporter/mappings.py Normal file
View file

@ -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"],
},
)

32
src/nte_history_exporter/pool_mappings.py Normal file
View file

@ -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"],
}