"""JSON检查点恢复合成研究任务；按合同验证数据，有限重试，不运行外部任务。"""
import copy
import hashlib
import json
from datetime import datetime
from decimal import Decimal as D, InvalidOperation

MAX_ATTEMPTS = 2


def digest(value):
    return hashlib.sha256(json.dumps(value, sort_keys=True).encode()).hexdigest()


def finite(value):
    try:
        number = D(str(value))
    except (InvalidOperation, ValueError, TypeError) as exc:
        raise ValueError("invalid_numeric_input") from exc
    if not number.is_finite():
        raise ValueError("nonfinite_step_value")
    return number


def valid(row, kind, contract):
    if not row:
        return False
    try:
        if row.get("symbol") != contract["symbol"] or row.get("currency") != contract["currency"] or not row.get("source_id"):
            return False
        decision, available, expires = map(datetime.fromisoformat, [contract["decision"], row["available"], row["expires"]])
        if any(t.utcoffset() is None for t in [decision, available, expires]) or not available <= decision <= expires:
            return False
        if kind == "price" and row.get("date") != contract["price_date"]:
            return False
        if kind == "eps" and (row.get("year") != contract["year"] or row.get("basis") != contract["basis"]):
            return False
        return finite(row["value"]) > 0
    except (ValueError, KeyError):
        return False


def resume(checkpoint, contract, provider):
    run_id = digest(contract)
    record = checkpoint["runs"].setdefault(run_id, {"contract": copy.deepcopy(contract), "attempts": {}, "steps": {}, "status": "pending"})
    if record["status"] == "complete":
        expected = digest({"steps": record["steps"], "formula": "pe-v1"})
        if record.get("dependency_hash") != expected or not all(valid(record["steps"].get(kind), kind, contract) for kind in ["price", "eps"]):
            return {"status": "checkpoint_invalid", "pe": None, "trace": []}
        expected_pe = finite(record["steps"]["price"]["value"])/finite(record["steps"]["eps"]["value"])
        if finite(record["pe"]) != expected_pe:
            return {"status": "checkpoint_invalid", "pe": None, "trace": []}
        return {"status": "complete_reused", "pe": record["pe"], "trace": []}
    trace = []
    for kind in ["price", "eps"]:
        row = record["steps"].get(kind) or checkpoint["cache"].get(kind)
        if valid(row, kind, contract):
            record["steps"][kind] = copy.deepcopy(row)
            trace.append("reuse:"+kind)
            continue
        record["steps"].pop(kind, None)
        record.pop("pe", None)
        if record["attempts"].get(kind, 0) >= MAX_ATTEMPTS:
            record["status"] = "blocked"
            return {"status": "retry_budget_exhausted", "pe": None, "trace": trace}
        record["attempts"][kind] = record["attempts"].get(kind, 0)+1
        trace.append("fetch:"+kind)
        try:
            row = provider(kind, contract)
            if not valid(row, kind, contract):
                raise ValueError("invalid_step_payload")
        except (TimeoutError, ValueError) as exc:
            record["status"] = "partial"
            return {"status": "partial", "pe": None, "reason": str(exc), "trace": trace}
        record["steps"][kind] = copy.deepcopy(row)
        checkpoint["cache"][kind] = copy.deepcopy(row)
    trace.append("calculate:pe-v1")
    dependency_hash = digest({"steps": record["steps"], "formula": "pe-v1"})
    pe = finite(record["steps"]["price"]["value"])/finite(record["steps"]["eps"]["value"])
    record.update(status="complete", pe=str(pe), dependency_hash=dependency_hash)
    if run_id not in checkpoint["completed"]:
        checkpoint["completed"].append(run_id)
    return {"status": "complete", "pe": str(pe), "trace": trace}


def payload(kind, contract, value):
    day = contract["price_date"]
    return {"symbol": contract["symbol"], "currency": "CNY", "basis": "diluted", "year": 2027,
            "date": day, "available": day+"T15:00:00+08:00", "expires": day+"T23:59:59+08:00",
            "source_id": kind+":"+day, "value": value}


t1 = {"symbol": "AAA", "year": 2027, "basis": "diluted", "currency": "CNY",
      "price_date": "2026-10-26", "decision": "2026-10-26T16:00:00+08:00"}
price = payload("price", t1, "24")
checkpoint = {"cache": {"price": price}, "completed": [], "runs": {digest(t1): {
    "contract": t1, "attempts": {"price": 1, "eps": 1}, "steps": {"price": price}, "status": "partial"}}}
checkpoint = json.loads(json.dumps(checkpoint))
calls = []


def provider(kind, contract):
    calls.append(kind)
    return payload(kind, contract, "1.2")


main = resume(checkpoint, t1, provider)
assert main["pe"] == "2E+1" and calls == ["eps"]
first_saved = copy.deepcopy(checkpoint["runs"][digest(t1)])
duplicate = resume(checkpoint, t1, provider)
assert duplicate["status"] == "complete_reused" and calls == ["eps"] and len(checkpoint["completed"]) == 1
t2 = {**t1, "price_date": "2026-10-27", "decision": "2026-10-27T16:00:00+08:00"}


def fail_eps(kind, contract):
    calls.append(kind)
    if kind == "eps":
        raise TimeoutError("new_eps_unavailable")
    return payload(kind, contract, "30")


interrupted = resume(checkpoint, t2, fail_eps)
assert interrupted["pe"] is None and interrupted["trace"] == ["fetch:price", "fetch:eps"]
assert checkpoint["cache"]["eps"]["expires"].startswith("2026-10-26")
failure_copy = copy.deepcopy(checkpoint)
resume(failure_copy, t2, fail_eps)
exhausted = resume(failure_copy, t2, fail_eps)
assert exhausted["status"] == "retry_budget_exhausted" and exhausted["pe"] is None
repaired = resume(checkpoint, t2, lambda kind, contract: payload(kind, contract, "1.5"))
assert repaired["pe"] == "2E+1" and repaired["trace"] == ["reuse:price", "fetch:eps", "calculate:pe-v1"]
assert checkpoint["runs"][digest(t1)] == first_saved and len(checkpoint["completed"]) == 2
tampered = copy.deepcopy(checkpoint)
tampered["runs"][digest(t1)]["steps"]["price"]["value"] = "99"
assert resume(tampered, t1, provider)["status"] == "checkpoint_invalid"
for token in ["NaN", "Infinity", "-Infinity"]:
    assert not valid(payload("eps", t2, token), "eps", t2)
print(json.dumps({"resumed": main, "duplicate": duplicate, "new_date_interrupted": interrupted,
                  "new_date_repaired": repaired, "exhausted": exhausted, "completion_count": len(checkpoint["completed"])}, sort_keys=True))