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

Expand achievement exports with progress and metadata

Browse source
This commit is contained in:
Golumpa 2026-08-19 12:32:16 +01:00
parent 35918f24f1
commit 40b44521aa
17 changed files with 898 additions and 52 deletions
Download patch file Download diff file Expand all files Collapse all files

67
src/nte_history_exporter/decoder/achievement.py
View file

@ -1,5 +1,7 @@
import re
import struct
from dataclasses import dataclass
from datetime import datetime, timezone
ACHIEVEMENT_RECORD_MARKER = b"AchievementRecord\0"
ACHIEVEMENT_ID = re.compile(rb"([A-Za-z][A-Za-z0-9]*_[A-Za-z0-9_]+)\0")
@ -7,17 +9,58 @@ RECORD_COUNT_OFFSET = 22
RECORDS_OFFSET = 38
COMPLETION_TICKS_OFFSET = 8
MAX_ACHIEVEMENTS = 1_000
DOTNET_EPOCH_TICKS = 621_355_968_000_000_000
DOTNET_TICKS_PER_SECOND = 10_000_000
@dataclass(frozen=True)
class AchievementRecord:
achievement_id: str
progress: int
completion_ticks: int
@property
def completed(self) -> bool:
return self.completion_ticks != 0
@property
def status(self) -> str:
if self.completed:
return "completed"
return "in_progress" if self.progress else "not_started"
@property
def completed_at(self) -> str | None:
if not self.completed:
return None
unix_seconds = (
self.completion_ticks - DOTNET_EPOCH_TICKS
) / DOTNET_TICKS_PER_SECOND
try:
return datetime.fromtimestamp(unix_seconds, timezone.utc).strftime(
"%Y-%m-%d %H:%M:%S"
)
except (OverflowError, OSError, ValueError):
return None
def extract_achievement_ids(stream: bytes) -> list[str]:
return [
record.achievement_id
for record in extract_achievement_records(stream)
if record.completed
]
def extract_achievement_records(stream: bytes) -> list[AchievementRecord]:
for block in _game_data_blocks(stream):
decoded = _decompress_lz4_block(block)
marker = decoded.find(ACHIEVEMENT_RECORD_MARKER)
if marker < 0:
continue
achievement_ids = _completed_achievement_ids(decoded, marker)
if achievement_ids:
return achievement_ids
records = _achievement_records(decoded, marker)
if records:
return records
return []
@ -34,7 +77,7 @@ def reassemble_tcp_segments(segments: list[tuple[int, bytes]]) -> bytes:
return bytes(output)
def _completed_achievement_ids(decoded: bytes, marker: int) -> list[str]:
def _achievement_records(decoded: bytes, marker: int) -> list[AchievementRecord]:
# The component header stores its record count before a fixed schema descriptor.
if marker + RECORDS_OFFSET > len(decoded):
return []
@ -44,15 +87,21 @@ def _completed_achievement_ids(decoded: bytes, marker: int) -> list[str]:
matches = list(ACHIEVEMENT_ID.finditer(decoded, marker + RECORDS_OFFSET))
if len(matches) < count:
return []
completed = []
records = []
for match in matches[:count]:
# Each name is followed by progress, then a zero timestamp until completion.
# Each name is followed by an unsigned progress value and completion ticks.
progress_offset = match.end()
ticks_offset = match.end() + COMPLETION_TICKS_OFFSET
if ticks_offset + 8 > len(decoded):
return []
if struct.unpack_from("<Q", decoded, ticks_offset)[0]:
completed.append(match.group(1).decode("ascii"))
return completed
records.append(
AchievementRecord(
achievement_id=match.group(1).decode("ascii"),
progress=struct.unpack_from("<Q", decoded, progress_offset)[0],
completion_ticks=struct.unpack_from("<Q", decoded, ticks_offset)[0],
)
)
return records
def _game_data_blocks(stream: bytes):

83
src/nte_history_exporter/export/json_export.py
View file

@ -12,6 +12,8 @@ from nte_history_exporter.constants import (
MYSTERY_BOX_BANNER_ID,
)
from nte_history_exporter.decoder.server_region import account_region_for_server
from nte_history_exporter.decoder.achievement import AchievementRecord
from nte_history_exporter.mappings import ACHIEVEMENTS_BY_CASEFOLD
def build_export_json(
@ -78,6 +80,87 @@ def build_export_json(
return export
def build_achievement_export_json(
achievement_records: list[AchievementRecord],
*,
source: str = "packet_capture",
capture_source: str | None = None,
user_uid: str | None = None,
server_id: str | None = None,
) -> dict[str, Any]:
categories: dict[str, list[dict[str, Any]]] = {}
scan = {
"in_game": {
"total_achievements": 0,
"completed_achievements": 0,
"in_progress_achievements": 0,
"not_started_achievements": 0,
},
"playstation": {
"total_achievements": 0,
"completed_achievements": 0,
"in_progress_achievements": 0,
"not_started_achievements": 0,
},
}
for achievement in achievement_records:
achievement_id = achievement.achievement_id
is_playstation = achievement_id.casefold().startswith("playstation_")
summary = scan["playstation" if is_playstation else "in_game"]
summary["total_achievements"] += 1
if achievement.completed:
summary["completed_achievements"] += 1
elif achievement.progress:
summary["in_progress_achievements"] += 1
else:
summary["not_started_achievements"] += 1
category, separator, _number = achievement_id.partition("_")
category_key = category.casefold() if separator else "uncategorized"
record = {
"id": achievement_id,
"platform": "playstation" if is_playstation else "in_game",
"status": achievement.status,
"progress": achievement.progress,
"completed": achievement.completed,
"completed_at": achievement.completed_at,
}
metadata = ACHIEVEMENTS_BY_CASEFOLD.get(achievement_id.casefold())
if metadata:
record.update(
{
"name": metadata["name"],
"description": metadata["description"],
"quality": metadata["quality"],
"target": metadata["target"],
"rewards": metadata["rewards"],
}
)
categories.setdefault(category_key, []).append(record)
export: dict[str, Any] = {
"format": "nte-achievement-export",
"format_version": 1,
"game": GAME_NAME,
"source": source,
}
if capture_source:
export["capture_source"] = capture_source
export["exporter"] = {"name": EXPORTER_NAME, "version": __version__}
export["scan"] = scan
normalized_user_uid = user_uid.strip() if user_uid else ""
normalized_server_id = str(server_id).strip() if server_id else ""
if normalized_user_uid:
export["user_uid"] = normalized_user_uid
if normalized_server_id:
export["server_id"] = normalized_server_id
account_region = account_region_for_server(normalized_server_id)
if account_region:
export["account_region"] = account_region
export["categories"] = categories
return export
def _record_for_export(row: dict[str, Any]) -> dict[str, Any]:
if row.get("pool_group_id") == ARC_BANNER_ID:
return {

52
src/nte_history_exporter/live_capture/runner.py
View file

@ -9,10 +9,16 @@ from pathlib import Path
from nte_history_exporter import console
from nte_history_exporter.constants import POOL_META
from nte_history_exporter.decoder.achievement import extract_achievement_ids, reassemble_tcp_segments
from nte_history_exporter.decoder.achievement import (
extract_achievement_records,
reassemble_tcp_segments,
)
from nte_history_exporter.decoder.boundary import annotate_groups, select_continuous_run_from_page_1
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.export.json_export import (
build_achievement_export_json,
build_export_json,
)
from nte_history_exporter.live_capture.backends import open_capture_backend
from nte_history_exporter.live_capture.diagnostics import new_diagnostics_path, write_capture_diagnostics
from nte_history_exporter.live_capture.session import LiveHistorySession, UdpPacket
@ -81,7 +87,7 @@ def run_live_capture(
reported_missing_pages: dict[str, tuple[int, ...]] = {}
active_gap_notices: set[str] = set()
tcp_segments: dict[tuple[str, int, str, int], list[tuple[int, bytes]]] = {}
achievement_ids: list[str] = []
achievement_records = []
try:
with StopKeyMonitor() as stop_key:
@ -91,7 +97,7 @@ def run_live_capture(
if packet is None:
continue
if (
not achievement_ids
not achievement_records
and packet.protocol == "tcp"
and packet.payload
and packet.dst_ip == local_ip
@ -100,16 +106,17 @@ def run_live_capture(
tcp_segments.setdefault(flow, []).append(
(packet.sequence_number, packet.payload)
)
if not achievement_ids:
achievement_ids = extract_achievement_ids(
if not achievement_records:
achievement_records = extract_achievement_records(
reassemble_tcp_segments(tcp_segments[flow])
)
if achievement_ids:
if achievement_records:
completed = [record for record in achievement_records if record.completed]
console.print_achievements_captured(
len(achievement_ids),
len(completed),
sum(
value.lower().startswith("playstation_")
for value in achievement_ids
record.achievement_id.lower().startswith("playstation_")
for record in completed
),
)
pair_count_before = len(session.pairs)
@ -156,18 +163,26 @@ def run_live_capture(
exports = []
achievement_path = None
if achievement_ids:
achievement_path = _achievement_path()
achievement_path.write_text(
json.dumps(achievement_ids, ensure_ascii=False) + "\n", encoding="utf-8"
)
diagnostics_path = None
if write_debug_csv:
diagnostics_path = new_diagnostics_path()
write_capture_diagnostics(diagnostics_path, session.diagnostic_report())
resolved_user_uid = user_uid or session.user_uid
if session.kinds_seen() and not resolved_user_uid:
if (session.kinds_seen() or achievement_records) and not resolved_user_uid:
resolved_user_uid = console.prompt_user_uid()
if achievement_records:
achievement_path = _achievement_path(resolved_user_uid)
achievement_export = build_achievement_export_json(
achievement_records,
source="live_capture",
capture_source=CAPTURE_SOURCE_LABELS.get(capture.name, capture.name),
user_uid=resolved_user_uid,
server_id=session.server_id,
)
achievement_path.write_text(
json.dumps(achievement_export, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
resolved_server_id = session.server_id
if session.kinds_seen() and not resolved_server_id:
resolved_server_id = console.prompt_server_id()
@ -270,10 +285,11 @@ def run_live_capture(
}
def _achievement_path() -> Path:
def _achievement_path(user_uid: str | None = None) -> Path:
export_dir = Path("exports")
export_dir.mkdir(parents=True, exist_ok=True)
base = export_dir / f"Achievements_{datetime.now():%Y%m%d_%H%M%S}"
uid_prefix = _safe_filename_part(user_uid) if user_uid else "unknown"
base = export_dir / f"{uid_prefix}_Achievements_{datetime.now():%Y%m%d_%H%M%S}"
path = base.with_suffix(".json")
counter = 2
while path.exists():

112
src/nte_history_exporter/mapping_update.py
View file

@ -20,10 +20,11 @@ ASSET_TABLES = {
"appearances": "DataTable/Character/Appearance/DT_AppearanceData.json",
"illustrations": "DataTable/Gacha/GachaIllustrate.json",
"mystery_box_pools": "DataTable/GashaponLottery/DT_GashaponLotteryGlobal.json",
"achievements": "DataTable/DT_AchievementConfigInfo.json",
"vehicle_inventory": "DataTable/Vehicle/DT_VehicleItemData.json",
"localization": "Localization/en/game.json",
}
REWARD_MAPPING_FILES = ("arcs.json", "characters.json", "items.json")
MAPPING_FILES = ("arcs.json", "characters.json", "items.json", "achievements.json")
RANK_BY_QUALITY = {
"EItemQuality::ITEM_QUALITY_ORANGE": "S",
"EItemQuality::ITEM_QUALITY_PURPLE": "A",
@ -103,7 +104,7 @@ def load_assets(*, assets_root: Path | None = None, source_ref: str = DEFAULT_SO
def load_current_mappings(directory: Path) -> dict[str, dict[str, dict[str, Any]]]:
mappings = {}
for filename in REWARD_MAPPING_FILES:
for filename in MAPPING_FILES:
path = directory / filename
try:
value = json.loads(path.read_text(encoding="utf-8"))
@ -126,13 +127,16 @@ def build_mapping_update(
"arcs.json": _build_primary_mapping(assets.tables["arcs"], "arc", translations),
"characters.json": _build_primary_mapping(assets.tables["characters"], "character", translations),
"items.json": _build_item_mapping(assets.tables, translations),
"achievements.json": _build_achievement_mapping(
assets.tables["achievements"], translations
),
}
validate_mappings(output)
additions: dict[str, list[str]] = {}
updates: dict[str, list[str]] = {}
deletions: dict[str, list[str]] = {}
for filename in REWARD_MAPPING_FILES:
for filename in MAPPING_FILES:
old = current[filename]
new = output[filename]
additions[filename] = sorted(new.keys() - old.keys(), key=str.casefold)
@ -169,7 +173,7 @@ def build_mapping_update(
def write_update(result: MappingUpdateResult, directory: Path) -> None:
directory.mkdir(parents=True, exist_ok=True)
for filename in REWARD_MAPPING_FILES:
for filename in MAPPING_FILES:
_atomic_write(directory / filename, _dump_mapping(result.mappings[filename]))
report_text = json.dumps(result.report, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
_atomic_write(directory / "mapping-update-report.json", report_text)
@ -180,11 +184,11 @@ def apply_update(result: MappingUpdateResult, directory: Path) -> None:
directory.mkdir(parents=True, exist_ok=True)
originals: dict[Path, bytes | None] = {}
written: list[Path] = []
for filename in REWARD_MAPPING_FILES:
for filename in MAPPING_FILES:
path = directory / filename
originals[path] = path.read_bytes() if path.exists() else None
try:
for filename in REWARD_MAPPING_FILES:
for filename in MAPPING_FILES:
path = directory / filename
_atomic_write(path, _dump_mapping(result.mappings[filename]))
written.append(path)
@ -199,11 +203,11 @@ def apply_update(result: MappingUpdateResult, directory: Path) -> None:
def validate_mappings(mappings: dict[str, dict[str, dict[str, Any]]]) -> None:
if set(mappings) != set(REWARD_MAPPING_FILES):
raise MappingUpdateError(f"mapping set must be exactly {REWARD_MAPPING_FILES}")
if set(mappings) != set(MAPPING_FILES):
raise MappingUpdateError(f"mapping set must be exactly {MAPPING_FILES}")
locations: dict[str, str] = {}
folded: dict[str, str] = {}
for filename in REWARD_MAPPING_FILES:
for filename in MAPPING_FILES:
entries = mappings[filename]
if not isinstance(entries, dict):
raise MappingUpdateError(f"{filename} must contain an object")
@ -232,6 +236,33 @@ def validate_mappings(mappings: dict[str, dict[str, dict[str, Any]]]) -> None:
raise MappingUpdateError(f"{filename}:{item_id}.type must be item or cosmetic")
if rank not in {"S", "A", "B"}:
raise MappingUpdateError(f"{filename}:{item_id}.rank must be S, A, or B")
if filename == "achievements.json":
_validate_achievement_meta(filename, item_id, meta)
def _validate_achievement_meta(filename: str, item_id: str, meta: dict[str, Any]) -> None:
for field in ("description", "category", "quality"):
if not isinstance(meta.get(field), str) or not meta[field].strip():
raise MappingUpdateError(
f"{filename}:{item_id}.{field} must be a non-empty string"
)
target = meta.get("target")
if not isinstance(target, (int, float)) or isinstance(target, bool) or target < 0:
raise MappingUpdateError(f"{filename}:{item_id}.target must be a non-negative number")
rewards = meta.get("rewards")
if not isinstance(rewards, list):
raise MappingUpdateError(f"{filename}:{item_id}.rewards must be a list")
for index, reward in enumerate(rewards):
if (
not isinstance(reward, dict)
or not isinstance(reward.get("item_id"), str)
or not reward["item_id"]
or not isinstance(reward.get("amount"), (int, float))
or isinstance(reward.get("amount"), bool)
):
raise MappingUpdateError(
f"{filename}:{item_id}.rewards[{index}] is invalid"
)
def _extract_rows(document: Any, relative_path: str) -> dict[str, Any]:
@ -289,7 +320,12 @@ def _build_item_mapping(
candidate_ids = _reward_candidate_ids(tables)
for candidate_id in candidate_ids:
folded = candidate_id.casefold()
if candidate_id.isdigit() or folded.startswith("fork_") or folded.startswith("characterawaken_"):
if (
candidate_id.isdigit()
or folded.startswith("fork_")
or folded.startswith("characterawaken_")
or folded.startswith("fashion_vehicle_")
):
continue
sources = (inventory, capital, appearances, vehicle_inventory)
match = next((source.get(folded) for source in sources if folded in source), None)
@ -307,6 +343,62 @@ def _build_item_mapping(
return dict(sorted(result.items(), key=lambda pair: (pair[0].casefold(), pair[0])))
def _build_achievement_mapping(
rows: dict[str, Any],
translations: dict[str, list[tuple[str, str]]],
) -> dict[str, dict[str, Any]]:
result = {}
for achievement_id in sorted(rows, key=lambda value: (value.casefold(), value)):
row = rows[achievement_id]
if not isinstance(row, dict):
raise MappingUpdateError(
f"NTE_Assets achievement row {achievement_id} must be an object"
)
title = row.get("TitleId")
description = row.get("ContentID")
if not isinstance(title, dict) or not isinstance(description, dict):
raise MappingUpdateError(
f"NTE_Assets achievement row {achievement_id} has no title or description"
)
target = row.get("Amout")
if not isinstance(target, (int, float)) or isinstance(target, bool):
raise MappingUpdateError(
f"NTE_Assets achievement row {achievement_id} has invalid Amout"
)
main_type = row.get("AchievementMainType")
quality = row.get("Quality")
if not isinstance(main_type, str) or "MainType" not in main_type:
raise MappingUpdateError(
f"NTE_Assets achievement row {achievement_id} has invalid main type"
)
if not isinstance(quality, str) or "::" not in quality:
raise MappingUpdateError(
f"NTE_Assets achievement row {achievement_id} has invalid quality"
)
rewards = []
for index, award in enumerate(row.get("AwardInfo", [])):
if not isinstance(award, dict):
raise MappingUpdateError(
f"NTE_Assets achievement row {achievement_id} award {index} is invalid"
)
item_id = award.get("ItemID")
amount = award.get("Amount")
if not isinstance(item_id, str) or not item_id or not isinstance(amount, (int, float)):
raise MappingUpdateError(
f"NTE_Assets achievement row {achievement_id} award {index} is invalid"
)
rewards.append({"item_id": item_id, "amount": amount})
result[achievement_id] = {
"name": _translate_name(achievement_id, title, translations),
"description": _translate_name(achievement_id, description, translations),
"category": main_type.rsplit("MainType", 1)[-1].casefold(),
"quality": quality.rsplit("::", 1)[-1].casefold(),
"target": target,
"rewards": rewards,
}
return result
def _reward_candidate_ids(tables: dict[str, dict[str, Any]]) -> list[str]:
"""Return rewards used by ordinary Gacha and every Mystery Box rotation."""
candidates: dict[str, str] = {

4
src/nte_history_exporter/mappings.py
View file

@ -16,6 +16,10 @@ def load_mapping_file(filename: str) -> dict[str, Any]:
ARC_META: dict[str, dict[str, Any]] = load_mapping_file("arcs.json")
CHARACTERS: dict[str, dict[str, Any]] = load_mapping_file("characters.json")
ITEMS: dict[str, dict[str, Any]] = load_mapping_file("items.json")
ACHIEVEMENTS: dict[str, dict[str, Any]] = load_mapping_file("achievements.json")
ACHIEVEMENTS_BY_CASEFOLD = {
achievement_id.casefold(): meta for achievement_id, meta in ACHIEVEMENTS.items()
}
# Reward keys in Monopoly history packets are the reward id string itself
# (ASCII*4 with carry, see decoder.protocol.decode_reward_key), so all reward