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

Export partially-captured groups as stable

Browse source 
Stop treating partially-captured oldest timestamp groups (including Arc 10-pulls) as a special "incomplete" case. All decoded rows are now exported and marked with stable UIDs; the previous INCOMPLETE_* warnings and related console hint logic were removed. annotate_groups and annotate_arc_groups were simplified to always set uid_status="stable" and export_record=True, and arc_stability_warnings was deleted. CLI, live capture runner, console, tests, docs and mappings were updated to reflect the new behavior and to remove emission/handling of those informational warnings.
This commit is contained in:
Golumpa 2026-06-11 14:57:08 +01:00
parent 20b2086299
commit 5d16dac4d5
10 changed files with 53 additions and 170 deletions
Download patch file Download diff file Expand all files Collapse all files

8
src/nte_history_exporter/cli.py
View file

@ -5,7 +5,6 @@ import json
from nte_history_exporter import console
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
from nte_history_exporter.export.csv_export import write_csv
from nte_history_exporter.export.json_export import build_export_json
@ -39,13 +38,13 @@ def main(argv: list[str] | None = None) -> int:
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)
warnings = decoded["arc_warnings"]
kind = "arc_miracle_box"
best_run = decoded["best_arc_run"]
pair_count = len(decoded["arc_pairs"])
else:
rows, group_warnings = annotate_groups(decoded["rows"])
warnings = decoded["run_warnings"] + group_warnings
rows = annotate_groups(decoded["rows"])
warnings = decoded["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"])
@ -72,7 +71,6 @@ def main(argv: list[str] | None = None) -> int:
)
for warning in warnings:
console.print_warning(warning["code"], warning["reason"], warning.get("records"))
console.maybe_print_incomplete_hint(warnings)
print()
if args.debug:
console.print_note(f"CSV written: {out_path}")

34
src/nte_history_exporter/console.py
View file

@ -119,37 +119,3 @@ def print_success(text: str) -> None:
def print_problem(text: str) -> None:
print(style(f" {text}", YELLOW, BOLD))
def print_boxed(lines: list[str], *codes: str) -> None:
inner = WIDTH - 4
border = " +" + "-" * (WIDTH - 2) + "+"
print(style(border, *codes))
for line in lines:
print(style(" | " + line.ljust(inner) + " |", *codes))
print(style(border, *codes))
# Warning codes for an oldest timestamp group that was exported despite possibly
# being an unfinished 10-pull. The exported rows are stable, so the warning is
# informational and the user can ignore it once the rest is captured.
INCOMPLETE_EXPORT_CODES = {
"INCOMPLETE_TIMESTAMP_GROUP_EXPORTED",
"INCOMPLETE_ARC_10_PULL_EXPORTED",
}
def maybe_print_incomplete_hint(warnings: list[dict]) -> None:
if not any(w.get("code") in INCOMPLETE_EXPORT_CODES for w in warnings):
return
print()
print_boxed(
[
"The INCOMPLETE_..._EXPORTED warning above is safe",
"to ignore if the partial 10-pull it captured is",
"already in your tracker. Otherwise restart",
"the capture and go to a further page",
],
GREEN,
BOLD,
)

48
src/nte_history_exporter/decoder/arc.py
View file

@ -141,59 +141,23 @@ def make_arc_uid(timestamp_raw: str, ordinal: int, arc_key_hex: str) -> str:
def annotate_arc_groups(rows: list[dict[str, Any]]) -> None:
# Pages are anchored at page 1, so ordinal 0 of every group is captured and
# all UIDs are stable -- even a partially captured oldest 10-pull, whose
# unseen rows can only append after the captured ones. All rows are exported.
groups: dict[str, list[int]] = defaultdict(list)
for index, row in enumerate(rows):
groups[row["timestamp_raw_hex"]].append(index)
group_items = list(groups.items())
for group_index, (timestamp_raw, indexes) in enumerate(group_items):
at_oldest_boundary = group_index == len(group_items) - 1
# Arc pulls are always 10-pulls, so the only group that can be short is
# the oldest one, where the scan stopped mid-10-pull. Its captured prefix
# is ordinal-stable (ordinal 0 is captured, unseen rows append after it),
# so export it and flag it rather than dropping it.
incomplete = at_oldest_boundary and len(indexes) % 10 != 0
skip_reason = (
"arc timestamp group is not a complete 10-pull in this capture; "
"exported rows are stable, scroll further to capture the rest"
if incomplete
else ""
)
for group_index, (timestamp_raw, indexes) in enumerate(groups.items()):
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"] = "incomplete_stable" if incomplete else "stable"
row["uid_status"] = "stable"
row["export_record"] = True
row["skip_reason"] = skip_reason
def arc_stability_warnings(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
warnings = []
seen = set()
for row in rows:
if row.get("uid_status") != "incomplete_stable":
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_EXPORTED",
"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; "
"exported rows are stable, scroll further to capture the rest"
),
}
)
return warnings
row["skip_reason"] = ""
def select_continuous_arc_run(pairs: list[tuple]) -> tuple[list[tuple], list[dict[str, Any]]]:

51
src/nte_history_exporter/decoder/boundary.py
View file

@ -100,23 +100,17 @@ def is_dice_record(row: dict[str, Any]) -> bool:
return False
def annotate_groups(rows: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
def annotate_groups(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Assign timestamp-group ordinals and stable UIDs to every decoded row.
Pages are anchored at page 1 (see select_continuous_run_from_page_1), so the
newest group's ordinal 0 is always captured and every UID is stable. The only
nuance is the oldest group: if the capture did not reach the true end of
history (final page full) and the dice-only count is not a complete multiple
of 10, it may be an unfinished 10-pull continuing onto an uncaptured page. Its
captured prefix is still ordinal-stable, so it is exported with an
informational warning rather than withheld.
newest group's ordinal 0 is always captured. Within a group, ordinal 0 is the
newest record and unseen continuation rows can only append after the captured
ones, so every exported UID is stable -- even a partially captured oldest
group. All decoded rows are therefore exported.
"""
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
return rows
groups: list[list[dict[str, Any]]] = []
current: list[dict[str, Any]] = []
@ -132,39 +126,18 @@ def annotate_groups(rows: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], l
if current:
groups.append(current)
warnings: list[dict[str, Any]] = []
for group_index, group in enumerate(groups):
at_oldest_boundary = group_index == len(groups) - 1
dice_record_count = sum(1 for row in group if is_dice_record(row))
dice_complete = dice_record_count > 0 and dice_record_count % 10 == 0
incomplete = at_oldest_boundary and not final_page_is_partial and not dice_complete
skip_reason = (
"oldest timestamp group may continue onto the next uncaptured page; "
"exported rows are stable, scroll further to capture the rest"
if incomplete
else ""
)
if incomplete:
warnings.append(
{
"code": "INCOMPLETE_TIMESTAMP_GROUP_EXPORTED",
"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"] = "oldest" if at_oldest_boundary else ("newest" if group_index == 0 else "")
row["uid_status"] = "incomplete_stable" if incomplete else "stable"
row["timestamp_group_boundary"] = (
"oldest" if group_index == len(groups) - 1 else ("newest" if group_index == 0 else "")
)
row["uid_status"] = "stable"
row["uid"] = make_uid(row, ordinal)
row["export_record"] = True
row["skip_reason"] = skip_reason
return rows, warnings
row["skip_reason"] = ""
return rows

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

@ -9,7 +9,6 @@ from pathlib import Path
from nte_history_exporter import console
from nte_history_exporter.constants import POOL_META
from nte_history_exporter.decoder.arc import arc_stability_warnings
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
@ -88,11 +87,9 @@ def run_live_capture(
pairs = session.pairs_for_kind(kind)
best_run, run_warnings = select_continuous_run_from_page_1(pairs)
rows = session.build_rows(kind)
if kind == "arc_miracle_box":
warnings = run_warnings + arc_stability_warnings(rows)
else:
rows, group_warnings = annotate_groups(rows)
warnings = run_warnings + group_warnings
if kind != "arc_miracle_box":
rows = annotate_groups(rows)
warnings = run_warnings
pages_seen = [p[0] for p in best_run]
csv_path, json_path = export_paths(kind)
if write_debug_csv:
@ -123,7 +120,6 @@ def run_live_capture(
console.print_note("again, then reopen the history screen.")
return {"exports": []}
all_warnings = []
for item in exports:
scan = item["export"]["scan"]
console.print_export_summary(
@ -134,8 +130,6 @@ def run_live_capture(
)
for warning in scan["warnings"]:
console.print_warning(warning["code"], warning["reason"], warning.get("records"))
all_warnings.extend(scan["warnings"])
console.maybe_print_incomplete_hint(all_warnings)
print()
for item in exports: