SniperGold_ML/ml/p3/p3_s251_external_ingest/s251_checkpoint.py

# -*- 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"