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

feat: add achievement export

Browse source 
Automatically captures completed achievements during login and exports their IDs, including completed PlayStation trophies.
This commit is contained in:
Efk3 2026-08-03 21:12:20 +02:00
parent c87ad76b75
commit 9e510ab95f
7 changed files with 233 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

11
README.md
View file

@ -110,6 +110,15 @@ Exports are not copied to the clipboard by default. Add `--copy-clipboard` to co
If a page response is missed, the exporter reports the missing page number while capture is still running. Leave the exporter open, close and reopen that history board, then scroll down again. Scrolling backward within the existing view does not request the cached pages again. The replacement capture is accepted and the tool confirms when the gap has been recovered. If reopening the board still produces no page messages, return to the main menu and re-enter the game to start a fresh connection.
### Achievement export
Start the live exporter before launching the game. The completed achievement
list is loaded during login, captured automatically, and written to
`exports/Achievements_*.json` as an array of achievement IDs. Completed
PlayStation trophy entries are retained even though they are not shown in the
in-game achievement list. The console reports the visible and PlayStation
counts as soon as the list has been decoded.
#### Linux/macOS
Downloaded executable, using `sudo` when your system requires elevated capture permission:
@ -150,7 +159,7 @@ Decodes a `mitmproxy .flows` capture instead of listening live — used for rese
| Flag | Effect |
| --------- | --------------------------------------------------- |
| `--live` | Capture live UDP traffic instead of reading a file. |
| `--live` | Capture live traffic, including achievements loaded at login. |
| `--debug` | Also write the research CSV and privacy-safe capture diagnostics. |
| `--user-uid <uid>` | Override the auto-detected NTE user UID in the JSON export. |
| `--copy-clipboard` | Copy a single live export JSON to clipboard after saving. |

2
src/nte_history_exporter/cli.py
View file

@ -20,7 +20,7 @@ def build_parser() -> argparse.ArgumentParser:
description="Decode NTE pull 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("--live", action="store_true", help="capture live game 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(

13
src/nte_history_exporter/console.py
View file

@ -91,6 +91,10 @@ def print_live_instructions(local_ip: str, backend: str = "windows_raw", detail:
print(" last pull group can be confirmed as complete.")
print(" 5. You can open several boards in the same session.")
print()
print(style(" Achievement export", BOLD))
print(" Start the exporter before launching the game. Achievements")
print(" loaded at login are captured automatically.")
print()
print(style(" Waiting for history pages... press any key here when done.", BOLD, GREEN))
print(rule())
@ -100,6 +104,15 @@ def print_page_captured(label: str, page: int | None, *, recaptured: bool = Fals
print(style(" + ", GREEN, BOLD) + label + style(f" {action} {page}", DIM))
def print_achievements_captured(total: int, playstation: int) -> None:
visible = total - playstation
print(
style(" + ", GREEN, BOLD)
+ "Achievements"
+ style(f" captured {visible} visible + {playstation} PlayStation", DIM)
)
def print_missing_pages(label: str, pages: list[int]) -> None:
page_list = ", ".join(str(page) for page in pages)
print(style(" ! ", YELLOW, BOLD) + style(f"{label} missing page(s): {page_list}.", YELLOW, BOLD))

117
src/nte_history_exporter/decoder/achievement.py Normal file
View file

@ -0,0 +1,117 @@
import re
import struct
ACHIEVEMENT_RECORD_MARKER = b"AchievementRecord\0"
ACHIEVEMENT_ID = re.compile(rb"([A-Za-z][A-Za-z0-9]*_[A-Za-z0-9_]+)\0")
RECORD_COUNT_OFFSET = 22
RECORDS_OFFSET = 38
COMPLETION_TICKS_OFFSET = 8
MAX_ACHIEVEMENTS = 1_000
def extract_achievement_ids(stream: bytes) -> list[str]:
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
return []
def reassemble_tcp_segments(segments: list[tuple[int, bytes]]) -> bytes:
if not segments:
return b""
output = bytearray()
start = min(sequence for sequence, _payload in segments)
for sequence, payload in sorted(set(segments)):
offset = sequence - start
if offset > len(output):
return b""
output.extend(payload[max(0, len(output) - offset) :])
return bytes(output)
def _completed_achievement_ids(decoded: bytes, marker: int) -> list[str]:
# The component header stores its record count before a fixed schema descriptor.
if marker + RECORDS_OFFSET > len(decoded):
return []
count = struct.unpack_from("<I", decoded, marker + RECORD_COUNT_OFFSET)[0]
if count > MAX_ACHIEVEMENTS:
return []
matches = list(ACHIEVEMENT_ID.finditer(decoded, marker + RECORDS_OFFSET))
if len(matches) < count:
return []
completed = []
for match in matches[:count]:
# Each name is followed by progress, then a zero timestamp until completion.
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
def _game_data_blocks(stream: bytes):
position = 0
while position + 4 <= len(stream):
frame_size = struct.unpack_from("<I", stream, position)[0]
frame_end = position + 4 + frame_size
if not frame_size or frame_end > len(stream):
return
frame = stream[position + 4 : frame_end]
if len(frame) < 4:
return
header_size = struct.unpack_from("<I", frame)[0]
size_offset = 12 + header_size
data_offset = size_offset + 4
if data_offset <= len(frame):
data_size = struct.unpack_from("<I", frame, size_offset)[0]
if data_size and data_offset + data_size <= len(frame):
yield frame[data_offset : data_offset + data_size]
position = frame_end
def _decompress_lz4_block(data: bytes) -> bytes:
output = bytearray()
position = 0
while position < len(data):
if len(data) - position <= 3 and not any(data[position:]):
break
token = data[position]
position += 1
literal_length, position = _lz4_length(data, position, token >> 4)
literal_end = position + literal_length
if literal_end > len(data):
return b""
output.extend(data[position:literal_end])
position = literal_end
if position == len(data) or (
len(data) - position <= 3 and not any(data[position:])
):
break
if position + 2 > len(data):
return b""
offset = struct.unpack_from("<H", data, position)[0]
position += 2
if not offset or offset > len(output):
return b""
match_length, position = _lz4_length(data, position, token & 15)
for _ in range(match_length + 4):
output.append(output[-offset])
return bytes(output)
def _lz4_length(data: bytes, position: int, length: int) -> tuple[int, int]:
if length != 15:
return length, position
while position < len(data):
value = data[position]
position += 1
length += value
if value != 255:
return length, position
return length, position

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

@ -9,6 +9,7 @@ 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.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
@ -79,6 +80,8 @@ def run_live_capture(
console.print_capture_fallback(capture.fallback_reason)
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] = []
try:
with StopKeyMonitor() as stop_key:
@ -87,6 +90,28 @@ def run_live_capture(
break
if packet is None:
continue
if (
not achievement_ids
and packet.protocol == "tcp"
and packet.payload
and packet.dst_ip == local_ip
):
flow = (packet.src_ip, packet.src_port, packet.dst_ip, packet.dst_port)
tcp_segments.setdefault(flow, []).append(
(packet.sequence_number, packet.payload)
)
if not achievement_ids:
achievement_ids = extract_achievement_ids(
reassemble_tcp_segments(tcp_segments[flow])
)
if achievement_ids:
console.print_achievements_captured(
len(achievement_ids),
sum(
value.lower().startswith("playstation_")
for value in achievement_ids
),
)
pair_count_before = len(session.pairs)
matched = session.process_packet(
UdpPacket(
@ -137,6 +162,13 @@ 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"
)
console.print_note(f"Achievement list written: {achievement_path}")
diagnostics_path = None
if write_debug_csv:
diagnostics_path = new_diagnostics_path()
@ -183,13 +215,20 @@ def run_live_capture(
console.print_results_header()
if not exports:
console.print_problem("No history pages were captured.")
if achievement_path is not None:
console.print_note("No pull-history pages were captured.")
else:
console.print_problem("No history pages were captured.")
console.print_note("Make sure the capture backend is running, then reopen the")
console.print_note("history screen and scroll from page 1. If no page messages")
console.print_note("appear, return to the main menu and re-enter the game.")
if diagnostics_path is not None:
console.print_note(f"Diagnostics written: {diagnostics_path}")
return {"exports": [], "diagnostics_path": diagnostics_path}
return {
"exports": [],
"diagnostics_path": diagnostics_path,
"achievement_path": achievement_path,
}
for item in exports:
scan = item["export"]["scan"]
@ -218,7 +257,23 @@ def run_live_capture(
elif copy_clipboard and len(exports) > 1:
console.print_note("Multiple banners captured; clipboard copy skipped so one export")
console.print_note("does not overwrite another.")
return {"exports": exports, "diagnostics_path": diagnostics_path}
return {
"exports": exports,
"diagnostics_path": diagnostics_path,
"achievement_path": achievement_path,
}
def _achievement_path() -> 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}"
path = base.with_suffix(".json")
counter = 2
while path.exists():
path = export_dir / f"{base.name}_{counter}.json"
counter += 1
return path
def export_paths(kind: str, user_uid: str | None = None) -> tuple[Path, Path]:

7
src/nte_history_exporter/live_capture/windows_raw.py
View file

@ -17,6 +17,7 @@ class ParsedIpPacket:
dst_port: int
payload: bytes
protocol: str = "udp"
sequence_number: int = 0
ParsedIpUdpPacket = ParsedIpPacket
@ -69,12 +70,14 @@ def parse_ipv4_packet(data: bytes) -> ParsedIpPacket | None:
if protocol == 6:
if len(data) < ihl + 20:
return None
src_port, dst_port = struct.unpack_from("!HH", data, ihl)
src_port, dst_port, sequence_number = struct.unpack_from("!HHI", data, ihl)
tcp_header_len = (data[ihl + 12] >> 4) * 4
if tcp_header_len < 20 or len(data) < ihl + tcp_header_len:
return None
payload = data[ihl + tcp_header_len :]
return ParsedIpPacket(src_ip, dst_ip, src_port, dst_port, payload, "tcp")
return ParsedIpPacket(
src_ip, dst_ip, src_port, dst_port, payload, "tcp", sequence_number
)
return None

29
tests/test_achievement.py Normal file
View file

@ -0,0 +1,29 @@
import struct
import unittest
from nte_history_exporter.decoder.achievement import (
ACHIEVEMENT_RECORD_MARKER,
extract_achievement_ids,
reassemble_tcp_segments,
)
class AchievementPacketTests(unittest.TestCase):
def test_extracts_only_completed_achievements(self):
header = ACHIEVEMENT_RECORD_MARKER + b"\0" * 4 + struct.pack("<I", 2) + b"\0" * 12
completed = b"\x0a\0\0\0Battle_30\0" + b"\0" * 8 + struct.pack("<Q", 1)
incomplete = b"\x13\0\0\0Playstation_035\0" + b"\0" * 16
decoded = header + completed + incomplete
extension = len(decoded) - 15
compressed = b"\xf0" + bytes([extension]) + decoded
frame = struct.pack("<I", 0) + b"\0" * 8 + struct.pack("<I", len(compressed)) + compressed
stream = struct.pack("<I", len(frame)) + frame
self.assertEqual(extract_achievement_ids(stream), ["Battle_30"])
self.assertEqual(extract_achievement_ids(b"\x01\0\0\0\0"), [])
def test_reassembles_tcp_data(self):
self.assertEqual(
reassemble_tcp_segments([(104, b"bar"), (100, b"foobar"), (100, b"foobar")]),
b"foobarr",
)