"""真正按证券/年份/来源行合流；独立核算不读取报告PE，实际注入两类错误。"""
import copy
import json
from decimal import Decimal as D, InvalidOperation


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_branch_number")
    return number


raw = [{"symbol": s, "year": 2027, "source_row": "row-1", "eps_cell": e, "pe_cell": pe,
        "currency": "CNY", "basis": "diluted", "price_date": "2026-10-26"}
       for s, e, pe in [("AAA", "1.2", "20"), ("BBB", "1.5", "20"), ("CCC", "1", "18")]]
prices = {"AAA": "24", "BBB": "30", "CCC": "18"}


def extract(rows):
    return [{**{key: row[key] for key in ["symbol", "year", "source_row", "currency", "basis", "price_date"]},
             "reported_eps": str(finite(row["eps_cell"])), "reported_pe": str(finite(row["pe_cell"]))} for row in rows]


def recompute(rows):
    output = []
    for row in rows:
        eps, price = finite(row["eps_cell"]), finite(prices[row["symbol"]])
        if price <= 0 or eps <= 0:
            raise ValueError("nonpositive_pe_inputs")
        output.append({**{key: row[key] for key in ["symbol", "year", "source_row", "currency", "basis", "price_date"]},
                       "verified_eps": str(eps), "computed_pe": str(price/eps), "price": str(price)})
    return output


def origin(row):
    return row["symbol"], row["source_row"]


def key(row):
    return *origin(row), row["year"]


def index(rows):
    result = {}
    for row in rows:
        if not {"symbol", "year", "source_row", "currency", "basis", "price_date"} <= row.keys():
            raise ValueError("missing_identity_schema")
        if type(row["year"]) is not int or not row["symbol"] or not row["source_row"]:
            raise ValueError("invalid_identity_schema")
        k = origin(row)
        if k in result:
            raise ValueError("duplicate_branch_origin")
        result[k] = row
    return result


def merge(source_rows, extracted, calculated):
    if any(not {"reported_eps", "reported_pe"} <= row.keys() for row in extracted):
        raise ValueError("missing_extraction_schema")
    if any(not {"verified_eps", "computed_pe", "price"} <= row.keys() for row in calculated):
        raise ValueError("missing_calculation_schema")
    anchors, left, right = map(index, [source_rows, extracted, calculated])
    if set(left)-set(anchors) or set(right)-set(anchors):
        raise ValueError("unexpected_branch_origin")
    audit = []
    for identity, source in anchors.items():
        a, b = left.get(identity), right.get(identity)
        reasons = []
        if a is None or b is None:
            reasons.append("missing_branch")
        elif key(a) != key(source) or key(b) != key(source):
            reasons.append("year_mismatch")
        else:
            for field in ["currency", "basis", "price_date"]:
                if a[field] != source[field] or b[field] != source[field]:
                    reasons.append("scope_mismatch:"+field)
            try:
                if any(finite(row[field]) <= 0 for row, fields in [(a, ["reported_eps", "reported_pe"]), (b, ["verified_eps", "computed_pe", "price"])] for field in fields):
                    reasons.append("nonpositive_branch_number")
                if finite(a["reported_eps"]) != finite(b["verified_eps"]):
                    reasons.append("eps_mismatch")
                if abs(finite(a["reported_pe"])-finite(b["computed_pe"])) > D("0.01"):
                    reasons.append("arithmetic_mismatch")
            except ValueError as exc:
                reasons.append(str(exc))
        audit.append({"symbol": source["symbol"], "year": source["year"], "source_row": source["source_row"],
                      "status": "blocked" if reasons else "accepted", "reasons": reasons,
                      "accepted_pe": None if reasons else b["computed_pe"]})
    return audit


left, right = extract(raw), recompute(raw)
baseline = merge(raw, left, right)
assert len(baseline) == 3 and all(r["status"] == "accepted" for r in baseline)
bad = copy.deepcopy(left)
bad[1]["reported_pe"] = "21"
bad[2]["year"] = 2028
injected = merge(raw, bad, right)
assert injected[0]["status"] == "accepted"
assert injected[1]["reasons"] == ["arithmetic_mismatch"] and injected[1]["accepted_pe"] is None
assert injected[2]["reasons"] == ["year_mismatch"] and injected[2]["accepted_pe"] is None
assert merge(raw, list(reversed(bad)), list(reversed(right))) == injected
repaired = merge(raw, extract(raw), recompute(raw))
assert repaired == baseline
try:
    merge(raw, [*left, left[0]], right)
except ValueError as exc:
    assert str(exc) == "duplicate_branch_origin"
else:
    raise AssertionError("重复分支行不得最后写入获胜")
missing = merge(raw, left[:-1], right)
assert missing[-1]["reasons"] == ["missing_branch"]
for token in ["NaN", "Infinity", "-Infinity"]:
    invalid = copy.deepcopy(left)
    invalid[0]["reported_pe"] = token
    assert merge(raw, invalid, right)[0]["status"] == "blocked"
print(json.dumps({"baseline": baseline, "injected": injected, "repaired": repaired, "missing": missing}, sort_keys=True))