"""全新临时演示缓存实际落盘/校验/续跑；不读取或删除用户缓存。"""
import hashlib
import json
import os
import tempfile
from pathlib import Path
from decimal import Decimal as D, InvalidOperation

root = Path(tempfile.mkdtemp(prefix="cotrading-r537-demo-"))
calls = {s: 0 for s in ["AAA", "BBB", "CCC", "DDD"]}

def config(symbol, interval="1D", basis="raw"):
    return {"symbol": symbol, "source": "synthetic", "start": "2026-10-19", "end": "2026-10-20",
            "interval": interval, "basis": basis, "currency": "USD"}

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

def atomic_json(path, value):
    temporary = path.with_suffix(path.suffix+".writing")
    temporary.write_text(json.dumps(value, sort_keys=True), encoding="utf-8")
    os.replace(temporary, path)

def valid_payload(rows):
    try:
        return (isinstance(rows, list) and len(rows) == 2
                and [r["date"] for r in rows] == ["2026-10-19", "2026-10-20"]
                and all(D(str(r["close"])).is_finite() and D(str(r["close"])) > 0 for r in rows))
    except (ValueError, KeyError, TypeError, InvalidOperation):
        return False

def load_cache(folder, request):
    name = key(request)
    data, marker = folder/(name+".json"), folder/(name+".ok.json")
    if not data.exists() or not marker.exists():
        return None
    try:
        metadata = json.loads(marker.read_text(encoding="utf-8"))
        content = data.read_bytes()
        if metadata["request"] != request or metadata["sha256"] != hashlib.sha256(content).hexdigest():
            return None
        rows = json.loads(content)
        return rows if valid_payload(rows) else None
    except (ValueError, KeyError, TypeError, InvalidOperation):
        return None

def commit(folder, request, rows):
    target = folder/(key(request)+".json")
    staged = folder/(key(request)+".pending.json")
    atomic_json(staged, rows)
    content = staged.read_bytes()
    if not valid_payload(json.loads(content)):
        raise ValueError("invalid_cache_payload")
    os.replace(staged, target)
    atomic_json(folder/(key(request)+".ok.json"),
                {"request": request, "sha256": hashlib.sha256(content).hexdigest()})
    assert load_cache(folder, request) == rows

def fetch(symbol):
    calls[symbol] += 1
    if symbol == "CCC":
        raise TimeoutError("synthetic_timeout")
    value = {"AAA": 10, "BBB": 20, "DDD": 40}[symbol]
    return [{"date": d, "close": value+i} for i, d in enumerate(["2026-10-19", "2026-10-20"])]

attempts = {key(config("AAA")): 1, key(config("BBB")): 1}
commit(root, config("AAA"), fetch("AAA"))
partial = fetch("BBB")
(root/(key(config("BBB"))+".json")).write_text(json.dumps(partial)[:12], encoding="utf-8")
atomic_json(root/"attempts.json", attempts)

def resume():
    counts = json.loads((root/"attempts.json").read_text(encoding="utf-8"))
    result = {}
    for batch in [["AAA", "BBB"], ["CCC", "DDD"]]:
        for symbol in batch:
            request, name = config(symbol), key(config(symbol))
            if load_cache(root, request) is not None:
                result[symbol] = "reused_verified"; continue
            while counts.get(name, 0) < 2:
                counts[name] = counts.get(name, 0)+1
                atomic_json(root/"attempts.json", counts)
                try:
                    rows = fetch(symbol)
                    commit(root, request, rows)
                    result[symbol] = "downloaded"; break
                except TimeoutError:
                    pass
            else:
                result[symbol] = "failed_budget"
    return result

first = resume()
assert first == {"AAA": "reused_verified", "BBB": "downloaded", "CCC": "failed_budget", "DDD": "downloaded"}
before_calls = dict(calls)
second = resume()
assert calls == before_calls == {"AAA": 1, "BBB": 2, "CCC": 2, "DDD": 1}
assert second == {"AAA": "reused_verified", "BBB": "reused_verified", "CCC": "failed_budget", "DDD": "reused_verified"}
assert load_cache(root, config("AAA", interval="1W")) is None
assert load_cache(root, config("AAA", basis="adjusted")) is None
probe = root/"corruption-probe"; probe.mkdir()
commit(probe, config("AAA"), load_cache(root, config("AAA")))
(probe/(key(config("AAA"))+".json")).write_text("[]", encoding="utf-8")
assert load_cache(probe, config("AAA")) is None
publication_probe = root/"publication-probe"; publication_probe.mkdir()
request = config("AAA")
good_rows = load_cache(root, request)
bad_rows = [{**good_rows[0], "close": -1}, good_rows[1]]
try:
    commit(publication_probe, request, bad_rows)
except ValueError as exc:
    assert str(exc) == "invalid_cache_payload"
else:
    raise AssertionError("无效暂存必须拒绝")
assert not (publication_probe/(key(request)+".ok.json")).exists()
commit(publication_probe, request, good_rows)
data_path = publication_probe/(key(request)+".json")
marker_path = publication_probe/(key(request)+".ok.json")
verified_bytes = (data_path.read_bytes(), marker_path.read_bytes())
try:
    commit(publication_probe, request, bad_rows)
except ValueError as exc:
    assert str(exc) == "invalid_cache_payload"
else:
    raise AssertionError("无效替换必须拒绝")
assert verified_bytes == (data_path.read_bytes(), marker_path.read_bytes())
assert load_cache(publication_probe, request) == good_rows
print(json.dumps({"first_resume": first, "second_resume": second, "fetch_calls": calls,
                  "cache_directory": str(root), "valid_symbols": [s for s in calls if load_cache(root, config(s)) is not None],
                  "frequency_and_basis_isolated": True, "corruption_detected": True}, sort_keys=True))