forked from chiki2bum2/SniperGold_ML
95 lines
No EOL
2.8 KiB
Python
95 lines
No EOL
2.8 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""P3-S25.1 CHECKPOINT / RESUME (atomic).
|
|
|
|
Persistent JSON checkpoint `p3_s251_checkpoint.json`. Only ever updated
|
|
atomically (write tmp -> fsync -> os.replace). A checkpoint is only recorded
|
|
once a chunk is fully parsed + hashed + its output flushed; an interrupted
|
|
chunk is reprocessed deterministically on resume (no duplicates: rows are
|
|
appended to the output files only at chunk completion).
|
|
|
|
Refusal to resume: if source size/mtime differ from the recorded run, the
|
|
pipeline refuses automatic resume and requires a new run identity.
|
|
"""
|
|
import datetime as dt
|
|
import json
|
|
import os
|
|
|
|
import s251_config as CFG
|
|
import s251_parse as P
|
|
|
|
CHECKPOINT = os.path.join(CFG.OUT, "p3_s251_checkpoint.json")
|
|
|
|
|
|
def now_utc():
|
|
return dt.datetime.now(dt.timezone.utc).isoformat()
|
|
|
|
|
|
class Checkpoint(object):
|
|
def __init__(self, data):
|
|
self.data = data
|
|
|
|
|
|
def empty_checkpoint(source_stat, run_id, chunk_rows):
|
|
return {
|
|
"run_id": run_id,
|
|
"source_path": source_stat["path"],
|
|
"source_sha256": source_stat["sha256"],
|
|
"source_size": source_stat["size_bytes"],
|
|
"source_mtime": source_stat["mtime_unix"],
|
|
"chunk_size": int(chunk_rows),
|
|
"last_completed_chunk": -1,
|
|
"current_chunk": -1,
|
|
"byte_offset": 0,
|
|
"next_byte_start": 0,
|
|
"last_timestamp": None,
|
|
"rows_processed": 0,
|
|
"malformed_counts": {k: 0 for k in P.new_malformed()},
|
|
"m15_rows_emitted": 0,
|
|
"m30_rows_emitted": 0,
|
|
"partial_carry_m15": None,
|
|
"partial_carry_m30": None,
|
|
"parser_version": CFG.PARSER_VERSION,
|
|
"schema_version": CFG.SCHEMA_VERSION,
|
|
"algorithm_version": CFG.ALGORITHM_VERSION,
|
|
"timestamp_basis": CFG.TIMESTAMP_BASIS,
|
|
"tz_offset_seconds": CFG.TZ_OFFSET_SECONDS,
|
|
"created_utc": now_utc(),
|
|
"updated_utc": now_utc(),
|
|
}
|
|
|
|
|
|
def _atomic_dump(data, path):
|
|
tmp = path + ".tmp"
|
|
with open(tmp, "w", encoding="utf-8") as f:
|
|
json.dump(data, f, indent=2, default=str)
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
os.replace(tmp, path)
|
|
|
|
|
|
def save(cp):
|
|
_atomic_dump(cp.data, CHECKPOINT)
|
|
|
|
|
|
def load():
|
|
with open(CHECKPOINT, "r", encoding="utf-8") as f:
|
|
return Checkpoint(json.load(f))
|
|
|
|
|
|
def exists():
|
|
return os.path.exists(CHECKPOINT)
|
|
|
|
|
|
def source_matches(cp):
|
|
"""Return (ok, reason). Refuses resume if source changed."""
|
|
d = cp.data
|
|
try:
|
|
m = os.path.getmtime(d["source_path"])
|
|
s = os.path.getsize(d["source_path"])
|
|
except OSError as e:
|
|
return False, "source unreadable: %s" % e
|
|
if s != d["source_size"]:
|
|
return False, "source size mismatch (%d vs %d)" % (s, d["source_size"])
|
|
if abs(m - d["source_mtime"]) > 2.0:
|
|
return False, "source mtime mismatch"
|
|
return True, "ok" |