diff --git a/README.md b/README.md index 48d9e6b..ba9f98c 100644 --- a/README.md +++ b/README.md @@ -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 ` | Override the auto-detected NTE user UID in the JSON export. | | `--copy-clipboard` | Copy a single live export JSON to clipboard after saving. | diff --git a/src/nte_history_exporter/cli.py b/src/nte_history_exporter/cli.py index a02ed86..5fa1fef 100644 --- a/src/nte_history_exporter/cli.py +++ b/src/nte_history_exporter/cli.py @@ -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( diff --git a/src/nte_history_exporter/console.py b/src/nte_history_exporter/console.py index 10bb086..cb50cc4 100644 --- a/src/nte_history_exporter/console.py +++ b/src/nte_history_exporter/console.py @@ -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)) diff --git a/src/nte_history_exporter/decoder/achievement.py b/src/nte_history_exporter/decoder/achievement.py new file mode 100644 index 0000000..57e6643 --- /dev/null +++ b/src/nte_history_exporter/decoder/achievement.py @@ -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(" 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(" len(stream): + return + frame = stream[position + 4 : frame_end] + if len(frame) < 4: + return + header_size = struct.unpack_from(" 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(" 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 diff --git a/src/nte_history_exporter/live_capture/runner.py b/src/nte_history_exporter/live_capture/runner.py index 06e0914..7bf94e3 100644 --- a/src/nte_history_exporter/live_capture/runner.py +++ b/src/nte_history_exporter/live_capture/runner.py @@ -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]: diff --git a/src/nte_history_exporter/live_capture/windows_raw.py b/src/nte_history_exporter/live_capture/windows_raw.py index 83b366b..d2035d0 100644 --- a/src/nte_history_exporter/live_capture/windows_raw.py +++ b/src/nte_history_exporter/live_capture/windows_raw.py @@ -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 diff --git a/tests/test_achievement.py b/tests/test_achievement.py new file mode 100644 index 0000000..49e52cc --- /dev/null +++ b/tests/test_achievement.py @@ -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("