Spaces:
Running
Running
"""Validation and durable history helpers for the daily refresh pipeline.
Pure helpers only: no network calls and no Streamlit imports.
"""
from __future__ import annotations
import csv
import hashlib
import io
import json
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path
from typing import Mapping
from sinetti_services import certificate_succession, renewal_match
from sinetti_services.event_classification import expired_before
from sinetti_services.returning_holder import classify_returning, format_off_scheme_evidence
_SEVERITY_RANK = {"ok": 0, "warning": 1, "critical": 2}
def _utc_iso(value=None) -> str:
if value is None:
value = datetime.now(timezone.utc)
if isinstance(value, str):
return value
if value.tzinfo is None:
value = value.replace(tzinfo=timezone.utc)
return value.astimezone(timezone.utc).isoformat()
def _norm_header(value: str) -> str:
return "".join(ch for ch in str(value or "").strip().lower() if ch.isalnum())
def _check(severity: str, code: str, message: str, **context) -> dict:
out = {"severity": severity, "code": code, "message": message}
if context:
out["context"] = context
return out
def _status(checks: list[dict]) -> str:
worst = "ok"
for check in checks:
sev = check.get("severity", "ok")
if _SEVERITY_RANK.get(sev, 0) > _SEVERITY_RANK[worst]:
worst = sev
return worst
def _ratio(ok: int, total: int) -> float:
return round((ok / total), 4) if total else 0.0
def _parse_date(value: str) -> bool:
raw = str(value or "").strip()
if not raw:
return False
for fmt in ("%Y-%m-%d", "%d.%m.%Y", "%d/%m/%Y"):
try:
datetime.strptime(raw, fmt)
return True
except ValueError:
pass
return False
def _collapse_check(checks: list[dict], row_count: int, previous_count: int) -> None:
if previous_count <= 0:
return
if row_count < max(1, int(previous_count * 0.5)):
checks.append(_check(
"critical", "row_count_collapse",
"Row count collapsed versus the previous accepted snapshot.",
previous_count=previous_count, row_count=row_count,
))
elif row_count > int(previous_count * 2.0):
checks.append(_check(
"warning", "row_count_spike",
"Row count is much larger than the previous accepted snapshot.",
previous_count=previous_count, row_count=row_count,
))
_ISCC_HEADER_GROUPS = {
"certificate_id": {"certificate"},
"holder_name": {"certificateholder"},
"country": {"country"},
"scope": {"scope"},
"feedstocks": {"rawmaterial"},
"valid_from": {"validfrom"},
"valid_until": {"validuntil"},
"issuing_cb": {"issuingcb", "certificationbody"},
}
def validate_iscc_csv_text(csv_text: str, *, previous_count: int = 0) -> dict:
text = str(csv_text or "").lstrip("\ufeff")
reader = csv.DictReader(io.StringIO(text))
headers = list(reader.fieldnames or [])
norm_headers = {_norm_header(h) for h in headers}
rows = list(reader)
checks: list[dict] = []
missing = [
name for name, aliases in _ISCC_HEADER_GROUPS.items()
if not (aliases & norm_headers)
]
if missing:
checks.append(_check(
"critical", "missing_required_columns",
"ISCC CSV is missing required columns.",
missing_columns=missing,
))
if not rows:
checks.append(_check("critical", "empty_export", "ISCC CSV parsed zero data rows."))
if text.lstrip().startswith("<"):
checks.append(_check("critical", "html_response", "ISCC CSV looks like an HTML response."))
_collapse_check(checks, len(rows), int(previous_count or 0))
def value_for(row: dict, canonical: str) -> str:
aliases = _ISCC_HEADER_GROUPS[canonical]
for key, value in row.items():
if _norm_header(key) in aliases:
return str(value or "").strip()
return ""
if rows and not missing:
id_rate = _ratio(sum(1 for r in rows if value_for(r, "certificate_id")), len(rows))
holder_rate = _ratio(sum(1 for r in rows if value_for(r, "holder_name")), len(rows))
date_rate = _ratio(sum(1 for r in rows if _parse_date(value_for(r, "valid_until"))), len(rows))
if id_rate < 0.98:
checks.append(_check("critical", "low_certificate_id_coverage",
"Too many ISCC rows have no certificate id.", coverage=id_rate))
if holder_rate < 0.95:
checks.append(_check("critical", "low_holder_coverage",
"Too many ISCC rows have no holder name.", coverage=holder_rate))
if date_rate < 0.90:
checks.append(_check("warning", "low_valid_until_parse_rate",
"Many ISCC valid-until dates did not parse.", coverage=date_rate))
else:
date_rate = 0.0
matched_headers = set().union(*_ISCC_HEADER_GROUPS.values())
return {
"status": _status(checks),
"row_count": len(rows),
"missing_columns": missing,
"extra_columns": [h for h in headers if _norm_header(h) not in matched_headers],
"date_parse_rate": date_rate,
"checks": checks or [_check("ok", "contract_ok", "ISCC CSV contract passed.")],
}
_IN_FORCE_STATUSES = frozenset({"active", "valid", "expiring"})
def validate_normalized_certs(scheme: str, certs: list[dict], *, previous_count: int = 0) -> dict:
rows = list(certs or [])
checks: list[dict] = []
if not rows:
checks.append(_check("critical", "empty_normalized_snapshot",
f"{scheme} normalized snapshot has zero rows."))
_collapse_check(checks, len(rows), int(previous_count or 0))
required = ["certificate_id", "holder_name", "holder_norm"]
if str(scheme).lower() != "ins":
required += ["country", "valid_until"]
bad_fields = []
# RSB publishes no holder country for lapsed certificates (its holder-country comes from the
# certificate PDF, which the register only keeps for certificates still in force). Country
# coverage for RSB is therefore judged over the in-force rows; every other rule, and every
# other scheme, is judged over all rows. Brief 2026-09-29-rsb-register-brief.md.
in_force_rows = [
row for row in rows
if str(row.get("status") or "").strip().casefold() in _IN_FORCE_STATUSES
]
for field in required:
judged = (
in_force_rows
if str(scheme).lower() == "rsb" and field == "country" and in_force_rows
else rows
)
coverage = _ratio(sum(1 for row in judged if str(row.get(field, "")).strip()), len(judged))
if coverage < 0.90:
bad_fields.append(field)
checks.append(_check(
"critical", "low_field_coverage",
f"{scheme} field coverage is below threshold.",
field=field, coverage=coverage,
))
statuses = sorted({str(row.get("status") or row.get("raw_status") or "").strip() for row in rows})
return {
"status": _status(checks),
"row_count": len(rows),
"bad_fields": bad_fields,
"status_values": [s for s in statuses if s][:25],
"checks": checks or [_check("ok", "contract_ok", f"{scheme} normalized contract passed.")],
}
def validate_history_retention(
scheme: str,
old_rows: list[dict],
new_rows: list[dict],
*,
today: str | None = None,
) -> dict:
"""Guard for registers that keep their expired certificates (REDcert publishes history back
to 2010). Our snapshot write is replace-all, so if the register ever trims old rows the trim
would silently propagate into our corpus on the next refresh. Expired rows leaving the export
are anomalous for such a register: any disappearance is a warning, and a disappearance beyond
max(25, 1% of the previously expired rows) is critical — the refresh keeps the previous
snapshot. Not for valid-only feeds (ISCC), where expired rows leaving is normal lifecycle."""
checks: list[dict] = []
today = today or datetime.now(timezone.utc).date().isoformat()
new_ids = {
str(row.get("certificate_id") or "").strip().casefold()
for row in (new_rows or [])
if str(row.get("certificate_id") or "").strip()
}
expired_old = [
row for row in (old_rows or [])
if len(str(row.get("valid_until") or "").strip()) == 10
and str(row.get("valid_until")).strip() < today
]
missing = [
row for row in expired_old
if str(row.get("certificate_id") or "").strip().casefold() not in new_ids
]
threshold = max(25, int(0.01 * len(expired_old)))
if missing:
severity = "critical" if len(missing) > threshold else "warning"
checks.append(_check(
severity, "history_trim",
f"{scheme} export dropped previously expired certificates — possible register "
"history trim.",
missing_expired=len(missing),
expired_previous=len(expired_old),
threshold=threshold,
examples=sorted(
str(row.get("certificate_id") or "").strip() for row in missing
)[:5],
))
return {
"status": _status(checks),
"row_count": len(new_rows or []),
"missing_expired": len(missing),
"checks": checks or [_check(
"ok", "history_retained",
f"{scheme} export still carries all previously expired certificates.",
)],
}
def validate_tabular_headers(source: str, rows: list[dict], header_groups: dict[str, set[str]]) -> dict:
rows = list(rows or [])
checks: list[dict] = []
if not rows:
checks.append(_check("critical", "empty_table", f"{source} parsed zero rows."))
headers = set()
else:
headers = {_norm_header(k) for row in rows[:10] for k in row}
normalized_groups = {
name: {_norm_header(alias) for alias in aliases}
for name, aliases in (header_groups or {}).items()
}
missing = [
name for name, aliases in normalized_groups.items()
if not (aliases & headers)
]
if missing:
checks.append(_check(
"critical", "missing_required_columns",
f"{source} is missing required columns.",
missing_columns=missing,
))
return {
"status": _status(checks),
"row_count": len(rows),
"missing_columns": missing,
"checks": checks or [_check("ok", "contract_ok", f"{source} header contract passed.")],
}
def validate_compliance_records(records: list[dict], *, previous_count: int = 0) -> dict:
rows = list(records or [])
checks: list[dict] = []
_collapse_check(checks, len(rows), int(previous_count or 0))
if not rows:
checks.append(_check("warning", "empty_compliance_snapshot",
"Compliance snapshot has zero rows."))
status_values = {str(r.get("status") or r.get("raw_status") or "").lower() for r in rows}
unexpected = sorted(v for v in status_values if v and v not in {"suspended", "withdrawn"})
if unexpected:
checks.append(_check("warning", "unexpected_compliance_status",
"Compliance snapshot contains unexpected status values.",
values=unexpected[:10]))
id_rate = _ratio(sum(1 for r in rows if str(r.get("certificate_id", "")).strip()), len(rows))
if rows and id_rate < 0.90:
checks.append(_check("critical", "low_certificate_id_coverage",
"Too many compliance rows have no certificate id.", coverage=id_rate))
return {
"status": _status(checks),
"row_count": len(rows),
"status_values": sorted(v for v in status_values if v),
"checks": checks or [_check("ok", "contract_ok", "Compliance contract passed.")],
}
def validate_blocklist_records(records: list[dict], *, previous_count: int = 0) -> dict:
rows = list(records or [])
checks: list[dict] = []
_collapse_check(checks, len(rows), int(previous_count or 0))
if not rows:
checks.append(_check("warning", "empty_blocklist_snapshot",
"Blocklist snapshot has zero rows."))
category_rate = _ratio(sum(1 for r in rows if str(r.get("category", "")).strip()), len(rows))
name_rate = _ratio(sum(
1 for r in rows
if str(r.get("name") or r.get("holder_name") or r.get("company") or "").strip()
),
len(rows))
if rows and category_rate < 0.95:
checks.append(_check("critical", "low_category_coverage",
"Too many blocklist rows have no category.", coverage=category_rate))
if rows and name_rate < 0.90:
checks.append(_check("critical", "low_name_coverage",
"Too many blocklist rows have no name.", coverage=name_rate))
return {
"status": _status(checks),
"row_count": len(rows),
"checks": checks or [_check("ok", "contract_ok", "Blocklist contract passed.")],
}
def validate_ins_blobs(blobs: dict) -> dict:
checks: list[dict] = []
counts = {k: len(blobs.get(k) or []) for k in ("operatori", "tipologie", "prodotti")}
if counts["operatori"] <= 0:
checks.append(_check("critical", "missing_operatori_blob",
"INS operator blob is missing or empty."))
for key in ("tipologie", "prodotti"):
if counts[key] <= 0:
checks.append(_check("warning", f"missing_{key}_blob",
f"INS {key} blob is missing or empty."))
return {
"status": _status(checks),
"row_count": counts["operatori"],
"blob_counts": counts,
"checks": checks or [_check("ok", "contract_ok", "INS embedded blobs passed.")],
}
def combine_validations(*reports: dict) -> dict:
checks: list[dict] = []
missing_columns: list[str] = []
bad_fields: list[str] = []
row_count = 0
for report in reports:
if not isinstance(report, dict):
continue
checks.extend(report.get("checks", []) or [])
missing_columns.extend(report.get("missing_columns", []) or [])
bad_fields.extend(report.get("bad_fields", []) or [])
row_count = max(row_count, int(report.get("row_count") or 0))
return {
"status": _status(checks),
"row_count": row_count,
"missing_columns": sorted(set(missing_columns)),
"bad_fields": sorted(set(bad_fields)),
"checks": checks or [_check("ok", "contract_ok", "Combined contract passed.")],
}
def validation_is_critical(report: dict) -> bool:
return (report or {}).get("status") == "critical"
def append_jsonl(path: str | Path, record: dict) -> None:
p = Path(path)
p.parent.mkdir(parents=True, exist_ok=True)
with p.open("a", encoding="utf-8") as fh:
fh.write(json.dumps(record, sort_keys=True, ensure_ascii=False) + "\n")
def read_jsonl(path: str | Path, *, limit: int | None = None) -> dict:
p = Path(path)
records: list[dict] = []
skipped = 0
if not p.exists():
return {"records": [], "skipped": 0}
for line in p.read_text(encoding="utf-8").splitlines():
raw = line.strip()
if not raw:
continue
try:
value = json.loads(raw)
except json.JSONDecodeError:
skipped += 1
continue
if isinstance(value, dict):
records.append(value)
else:
skipped += 1
if limit is not None:
records = records[-max(0, int(limit)):]
return {"records": records, "skipped": skipped}
def sha256_file(path: str | Path) -> str:
p = Path(path)
h = hashlib.sha256()
with p.open("rb") as fh:
for chunk in iter(lambda: fh.read(1024 * 1024), b""):
h.update(chunk)
return h.hexdigest()
def content_sha256(records: list[dict]) -> str:
"""Order-independent content hash over a list of records. Unlike sha256_file(), this hashes
the DATA, not a container — a gzip file's header embeds the write timestamp, so hashing the
file makes byte-identical content produce a different hash on every run (see
fingerprint_record's `records` param)."""
canon = sorted(
json.dumps(r, sort_keys=True, ensure_ascii=False, separators=(",", ":"))
for r in records
)
return hashlib.sha256("".join(canon).encode("utf-8")).hexdigest()
def fingerprint_record(*, run_id: str, scheme: str, kind: str, path: str | Path,
count: int, source: str = "daily_refresh",
accepted_at=None, records: list[dict] | None = None) -> dict:
return {
"run_id": run_id,
"scheme": scheme,
"kind": kind,
"count": int(count or 0),
"sha256": content_sha256(records) if records is not None else sha256_file(path),
"source": source,
"accepted_at": _utc_iso(accepted_at),
}
# A certificate that leaves a risky status for one of these did not come back — it ended.
_TERMINAL_STATUSES = {"expired", "terminated", "withdrawn"}
_RISKY_STATUSES = {
"suspended": ("certificate_suspended", "high"),
# Authority cancellation across every register (REDcert "entzogen", INS "ANNULLATO",
# LCFS/BC/CFR "withdrawn") now lands on the single harmonised status, so it emits the
# single event type. "certificate_revoked_or_terminated" survives for terminated only:
# the event log is append-only, so its stored name keeps its historical spelling.
"withdrawn": ("certificate_withdrawn", "high"),
"terminated": ("certificate_revoked_or_terminated", "low"),
}
def _cert_key(cert: dict) -> str:
return str(cert.get("certificate_id") or "").strip()
def _status_value(cert: dict) -> str:
return str(cert.get("status") or cert.get("raw_status") or "").strip().lower()
def _risk_category(event_type: str, *, status: str = "") -> str:
if event_type == "certificate_revoked_or_terminated" and status == "terminated":
return "status_change"
if event_type in {"certificate_suspended", "certificate_withdrawn", "certificate_revoked_or_terminated"}:
return "compliance_risk"
if event_type in {"certificate_removed_from_valid_list", "certificate_reinstated"}:
return "status_change"
if event_type in {"holder_new", "holder_returning"}:
return "market_intelligence"
return "operational"
def _as_str_list(value) -> list[str]:
if value is None:
return []
if isinstance(value, (list, tuple, set)):
return [str(item).strip() for item in value if str(item).strip()]
text = str(value).strip()
return [text] if text else []
def _cert_context(cert: dict) -> dict:
return {
"country": str(cert.get("country") or "").strip(),
"scope": _as_str_list(cert.get("scope")),
"feedstocks": _as_str_list(cert.get("feedstocks")),
"valid_from": str(cert.get("valid_from") or "").strip(),
"valid_until": str(cert.get("valid_until") or "").strip(),
"issuing_cb": str(cert.get("issuing_cb") or "").strip(),
"status": _status_value(cert),
}
def _event(run_id: str, scheme: str, event_type: str, severity: str, confidence: str,
cert: dict, *, detected_at, evidence: str, old=None, new=None) -> dict:
new = new or {}
status = str((new.get("status") if isinstance(new, dict) else "") or "").strip().lower()
if not status:
status = _status_value(cert)
event = {
"run_id": run_id,
"detected_at": _utc_iso(detected_at),
"scheme": scheme,
"event_type": event_type,
"severity": severity,
"risk_category": _risk_category(event_type, status=status),
"confidence": confidence,
"certificate_id": _cert_key(cert),
"holder_norm": str(cert.get("holder_norm", "")),
"holder_name": str(cert.get("holder_name", "")),
"old": old or {},
"new": new,
"evidence": evidence,
}
event.update(_cert_context(cert))
# ISCC publishes the suspension window on the compliance register (`cert_suspended_from` /
# `cert_suspended_until`). Keep it on the event so an episode has real dates instead of only
# the observation interval between two nightly runs. Absent keys stay absent: every other
# event type would otherwise gain two permanently empty fields.
for key in ("suspended_date", "suspended_until"):
value = str(cert.get(key) or "").strip()
if value:
event[key] = value
return event
# Event types that settle a withdrawn certificate's story: once one is on the append-only log
# for a certificate, the held-withdrawal branch below must never fire for it again.
_SETTLING_EVENT_TYPES = frozenset({
"certificate_withdrawn", "certificate_replaced", "certificate_revoked_or_terminated",
})
# A held withdrawal older than this is left alone: it predates event logging (or the log lost
# it), and the row already reads withdrawn. Keeps a first run with an incomplete log from
# emitting a night of thousands of stale high-severity events.
_HELD_WITHDRAWAL_MAX_AGE_DAYS = 60
def settled_certificate_ids(events: list[dict], scheme: str) -> set[str]:
"""Certificate ids of `scheme` that already carry a withdrawn / replaced / terminated event."""
out: set[str] = set()
for event in events or []:
if not isinstance(event, dict):
continue
if str(event.get("scheme") or "") != scheme:
continue
if str(event.get("event_type") or "") not in _SETTLING_EVENT_TYPES:
continue
cid = str(event.get("certificate_id") or "").strip()
if cid:
out.add(cid)
return out
def _days_since(day_text: object, detected_iso: str) -> int | None:
day = certificate_succession._parse_date(day_text)
detected = certificate_succession._parse_date(detected_iso)
if day is None or detected is None:
return None
return (detected - day).days
def _replaced_event(run_id: str, scheme: str, pair, *, detected_at, old_status: str) -> dict:
old_id = _cert_key(pair.old)
new_id = _cert_key(pair.new)
changes = list(pair.changes)
change_text = "; ".join(changes) if changes else "no visible scope or feedstock change"
event = _event(
run_id, scheme, "certificate_replaced", "low", "inferred", pair.old,
detected_at=detected_at,
evidence=(
f"{scheme} re-issued this certificate under a new number: {new_id} starts "
f"{pair.new.get('valid_from') or '?'} ({change_text}). The register marks the old "
f"number withdrawn when it re-issues; this is not an enforcement action."
),
old={"status": old_status, "certificate_id": old_id},
new={
"status": _status_value(pair.old),
"replaced_by": new_id,
"valid_from": str(pair.new.get("valid_from") or ""),
"valid_until": str(pair.new.get("valid_until") or ""),
"changes": changes,
},
)
event["replaced_by"] = new_id
event["replacement_changes"] = changes
event["replacement_gap_days"] = pair.gap_days
return event
def build_change_events(run_id: str, scheme: str, old_certs: list[dict], new_certs: list[dict],
*, detected_at=None, status_authority: list[dict] | None = None,
authority_source: str | None = None,
prior_coverage: Mapping[tuple[str, str], dict] | None = None,
settled_certificate_ids: set[str] | None = None) -> list[dict]:
"""Diff two snapshots of one scheme into change events.
`settled_certificate_ids` -- certificate ids that already carry a withdrawn / replaced /
terminated event on the log (see `settled_certificate_ids()`). When given, a REDcert
withdrawal with no successor yet is HELD for `certificate_succession.REPLACEMENT_WINDOW_DAYS`
and fires high only once that window passes with no replacement; when None, the flip fires
immediately as before (callers that do not keep a log)."""
old_by_id = {_cert_key(c): c for c in (old_certs or []) if _cert_key(c)}
new_by_id = {_cert_key(c): c for c in (new_certs or []) if _cert_key(c)}
authority_by_id = {_cert_key(c): c for c in (status_authority or []) if _cert_key(c)}
source_label = (authority_source or "").strip() or f"{scheme} status authority"
old_holders = {
str(c.get("holder_norm", "")).strip()
for c in (old_certs or [])
if str(c.get("holder_norm", "")).strip()
}
events: list[dict] = []
detected_iso = _utc_iso(detected_at)
# ISCC and REDcert renumber on renewal, so a routine renewal reaches us as one certificate
# disappearing and a differently-numbered one appearing. Pair those up BEFORE classifying
# either side: unpaired, the successor reads as a brand-new certificate and the superseded
# certificate reads as an exit — a false high-severity "removed while still valid" alarm
# whenever the renewal was issued before the old certificate expired.
renewals = renewal_match.match_renewals(
[old for cid, old in old_by_id.items() if cid not in new_by_id],
[new for cid, new in new_by_id.items() if cid not in old_by_id],
)
renewed_from = {_cert_key(pair.new): pair.old for pair in renewals}
superseded = {_cert_key(pair.old) for pair in renewals}
# REDcert marks the OLD number withdrawn when it re-issues a certificate (scope change,
# correction, early renewal) -- 2,758 of its 3,744 withdrawn certificates have a same-holder
# successor within 3 days. Read literally, that is a high-severity enforcement signal on a
# holder that simply got a new number. Pair withdrawn rows with their successor in the NEW
# snapshot first; a paired withdrawal becomes one low-severity `certificate_replaced` and its
# successor's appearance is folded into it (brief 2026-09-17).
replacement_by_old: dict[str, certificate_succession.Replacement] = {}
if scheme in certificate_succession.REPLACEMENT_SCHEMES:
for pair in certificate_succession.match_replacements(list(new_by_id.values()), scheme=scheme):
replacement_by_old[_cert_key(pair.old)] = pair
replaced_successors = {_cert_key(pair.new) for pair in replacement_by_old.values()}
hold_withdrawals = (
settled_certificate_ids is not None
and scheme in certificate_succession.REPLACEMENT_SCHEMES
)
window_days = certificate_succession.REPLACEMENT_WINDOW_DAYS
for cid, old in sorted(old_by_id.items()):
if cid not in new_by_id:
if cid in superseded:
continue # not an exit — the holder renewed under a new number
if expired_before(old.get("valid_until"), detected_iso):
events.append(_event(
run_id, scheme, "certificate_expired", "low", "explicit", old,
detected_at=detected_at,
evidence=f"{scheme} certificate reached its expiry date",
old={"status": _status_value(old)}, new={},
))
else:
auth = authority_by_id.get(cid)
auth_status = _status_value(auth) if auth else ""
if auth is not None and auth_status in _RISKY_STATUSES:
event_type, severity = _RISKY_STATUSES[auth_status]
# Prefer old row for holder identity; overlay status fields from authority.
cert_for_event = dict(old)
for key in ("status", "raw_status", "suspended_date", "suspended_until",
"valid_until", "issuing_cb"):
if auth.get(key):
cert_for_event[key] = auth.get(key)
# Valid-only feeds (ISCC CSV) often carry raw codes like "1", not a named
# lifecycle label — omit those from old.status so digests say "suspended"
# rather than "1 → suspended".
prior_status = _status_value(old)
if prior_status and prior_status not in _RISKY_STATUSES and prior_status not in {
"active", "valid", "expiring", "expired",
}:
prior_status = ""
events.append(_event(
run_id, scheme, event_type, severity, "explicit", cert_for_event,
detected_at=detected_at,
evidence=source_label,
old={"status": prior_status},
new={"status": auth_status},
))
else:
events.append(_event(
run_id, scheme, "certificate_removed_from_valid_list", "high", "inferred", old,
detected_at=detected_at,
evidence=(
f"{scheme} certificate left accepted snapshot while still valid "
f"(no terminated / suspended / withdrawn match)"
),
old={"status": _status_value(old)}, new={},
))
for cid, new in sorted(new_by_id.items()):
if cid not in old_by_id:
if cid in replaced_successors:
continue # its appearance IS the replacement, reported on the old number
predecessor = renewed_from.get(cid)
if predecessor is not None:
events.append(_event(
run_id, scheme, "certificate_renewed", "low", "inferred", new,
detected_at=detected_at,
evidence=(f"{scheme} certificate renewed under a new number "
f"(supersedes {_cert_key(predecessor)})"),
old={"certificate_id": _cert_key(predecessor),
"valid_until": predecessor.get("valid_until"),
"status": _status_value(predecessor)},
new={"valid_until": new.get("valid_until"), "status": _status_value(new)},
))
continue
holder_norm = str(new.get("holder_norm", "")).strip()
is_new_holder = bool(holder_norm and holder_norm not in old_holders)
if not is_new_holder:
events.append(_event(
run_id, scheme, "certificate_new", "low", "explicit", new,
detected_at=detected_at, evidence=f"{scheme} accepted snapshot",
old={}, new={"status": _status_value(new)},
))
continue
prior = (prior_coverage or {}).get((scheme, holder_norm)) or {}
decision = classify_returning(
scheme=scheme,
holder_norm=holder_norm,
valid_from=str(new.get("valid_from") or ""),
detected_at=detected_iso,
last_coverage_until=prior.get("last_coverage_until"),
prior_evidence=list(prior.get("prior_evidence") or []),
)
if decision is not None:
evidence = format_off_scheme_evidence(
scheme, decision["gap_days"], decision["last_coverage_until"],
)
event = _event(
run_id, scheme, "holder_returning", decision["severity"], "inferred", new,
detected_at=detected_at,
evidence=evidence,
old={"last_coverage_until": decision["last_coverage_until"]},
new={"status": _status_value(new), "valid_from": str(new.get("valid_from") or "")},
)
event["gap_days"] = decision["gap_days"]
event["prior_evidence"] = decision["prior_evidence"]
events.append(event)
else:
events.append(_event(
run_id, scheme, "holder_new", "medium", "explicit", new,
detected_at=detected_at, evidence=f"{scheme} accepted snapshot",
old={}, new={"status": _status_value(new)},
))
continue
old = old_by_id[cid]
old_status = _status_value(old)
new_status = _status_value(new)
pair = replacement_by_old.get(cid)
if pair is not None and (old_status != new_status or _cert_key(pair.new) not in old_by_id):
# Paired on the flip night, or paired late when the successor arrived a night or
# two after the withdrawal. A pair whose successor was already in the previous
# snapshot was reported when it appeared -- never again.
events.append(_replaced_event(
run_id, scheme, pair, detected_at=detected_at, old_status=old_status,
))
continue
if (
hold_withdrawals
and new_status == "withdrawn"
and old_status == new_status
and cid not in (settled_certificate_ids or set())
):
# Held on the flip night; fire once the window has passed with no successor.
age = _days_since(new.get("valid_until"), detected_iso)
if age is not None and window_days <= age <= _HELD_WITHDRAWAL_MAX_AGE_DAYS:
event_type, severity = _RISKY_STATUSES[new_status]
events.append(_event(
run_id, scheme, event_type, severity, "explicit", new,
detected_at=detected_at,
evidence=(
f"{scheme} explicit status field; no replacement certificate appeared "
f"within {window_days} days of the cut-off {new.get('valid_until')}"
),
old={"status": old_status}, new={"status": new_status},
))
continue
if new_status != old_status and new_status in _RISKY_STATUSES:
if hold_withdrawals and new_status == "withdrawn":
age = _days_since(new.get("valid_until"), detected_iso)
if age is not None and age < window_days:
continue # held: the successor may land in the next nightly
event_type, severity = _RISKY_STATUSES[new_status]
events.append(_event(
run_id, scheme, event_type, severity, "explicit", new,
detected_at=detected_at, evidence=f"{scheme} explicit status field",
old={"status": old_status}, new={"status": new_status},
))
elif old_status in _RISKY_STATUSES and new_status not in _RISKY_STATUSES:
# Leaving a risky status is only a reinstatement when the certificate returns to a
# LIVE status. suspended -> expired is the certificate reaching the end of its term
# while suspended; labelling that "reinstated" told the reader the opposite of what
# happened (31 of 31 stored reinstatements were REDcert suspended -> expired).
reinstated = new_status not in _TERMINAL_STATUSES
events.append(_event(
run_id, scheme,
"certificate_reinstated" if reinstated else "certificate_expired",
"medium" if reinstated else "low",
"explicit", new,
detected_at=detected_at, evidence=f"{scheme} explicit status field",
old={"status": old_status}, new={"status": new_status},
))
elif old.get("valid_until") and new.get("valid_until") and old.get("valid_until") != new.get("valid_until"):
events.append(_event(
run_id, scheme, "certificate_renewed", "low", "explicit", new,
detected_at=detected_at, evidence=f"{scheme} validity date changed",
old={"valid_until": old.get("valid_until")}, new={"valid_until": new.get("valid_until")},
))
return events
def build_compliance_reversal_events(
run_id: str,
scheme: str,
old_authority: list[dict] | None,
new_authority: list[dict] | None,
valid_certs: list[dict] | None,
*,
detected_at=None,
authority_source: str | None = None,
) -> list[dict]:
"""Reinstatements, read from the compliance register rather than from the valid-only feed.
ISCC's valid CSV carries no lifecycle status (every row reads `1`), so the reinstatement
branch in build_change_events — which needs a status field to change — can never fire for it.
A lifted ISCC suspension therefore left no trace at all: the certificate simply reappeared in
the valid list and the high-severity suspension stood forever. Diffing the compliance register
against its previous accepted copy gives the missing signal directly.
Two deliberate refusals:
* An empty/missing `new_authority` yields nothing. It is indistinguishable from a failed
compliance fetch, and treating it as "the register cleared everyone" would reinstate every
suspended certificate in a single run.
* A certificate cleared from the register but ABSENT from the valid list yields nothing. That
shape is ambiguous — a withdrawal that has also aged out looks the same — and the exit is
already covered by the disappearance branch.
"""
if not new_authority:
return []
old_adverse = {
_cert_key(row): row
for row in (old_authority or [])
if _cert_key(row) and _status_value(row) in _RISKY_STATUSES
}
if not old_adverse:
return []
still_adverse = {
_cert_key(row)
for row in new_authority
if _cert_key(row) and _status_value(row) in _RISKY_STATUSES
}
valid_by_id = {_cert_key(row): row for row in (valid_certs or []) if _cert_key(row)}
source_label = (authority_source or "").strip() or f"{scheme} status authority"
events: list[dict] = []
for cid, prior in sorted(old_adverse.items()):
if cid in still_adverse or cid not in valid_by_id:
continue
prior_status = _status_value(prior)
# Identity from the live valid row; the episode window only exists on the register row.
cert_for_event = dict(valid_by_id[cid])
for key in ("suspended_date", "suspended_until"):
if prior.get(key):
cert_for_event[key] = prior.get(key)
events.append(_event(
run_id, scheme, "certificate_reinstated", "medium", "explicit", cert_for_event,
detected_at=detected_at,
evidence=(f"{source_label} no longer lists this certificate as {prior_status}, "
f"and it is back in the accepted snapshot"),
old={"status": prior_status}, new={"status": ""},
))
return events
_ADVERSE_EVENT_STATUS = {
event_type: status for status, (event_type, _severity) in _RISKY_STATUSES.items()
}
# Events that say where a certificate's lifecycle stands. Anything else on the log (renewals,
# scope or address changes, soft alerts) neither opens nor closes an adverse episode.
_LIFECYCLE_EVENT_TYPES = frozenset(_ADVERSE_EVENT_STATUS) | {
"certificate_reinstated", "certificate_expired", "certificate_replaced",
"certificate_removed_from_valid_list",
}
# Same horizon as a held withdrawal: past it, the log entry may predate reinstatement logging,
# and good news that late is noise rather than a correction.
_ORPHANED_EPISODE_MAX_AGE_DAYS = 60
def build_orphaned_suspension_reinstatements(
run_id: str,
scheme: str,
history_events: list[dict] | None,
new_authority: list[dict] | None,
valid_certs: list[dict] | None,
*,
detected_at=None,
authority_source: str | None = None,
skip_certificate_ids: set[str] | None = None,
) -> list[dict]:
"""Reinstatements the register diff cannot see, read from the append-only change history.
build_compliance_reversal_events diffs the register against its previous ACCEPTED copy. When
the nightly fails to publish, that copy freezes while the change log keeps growing — so an
episode that opens and closes inside the outage is reported as a suspension (every night) and
never as lifted: the frozen baseline predates it. Shandong Sanju, EU-ISCC-Cert-DE129-35412331,
suspended 2026-08-31 and lifted 2026-09-14 during the 2026-08-26 outage, left 14 suspension
events and no reinstatement.
The history does not freeze. A certificate whose LATEST lifecycle event is adverse, which the
fresh register no longer lists as adverse, and which sits in the accepted valid list, has been
reinstated. Self-limiting: the emitted event becomes the latest lifecycle event, so the next
run is silent. Same two refusals as the register diff — an empty register is a failed fetch,
and a certificate absent from the valid list is ambiguous.
"""
if not new_authority:
return []
still_adverse = {
_cert_key(row)
for row in new_authority
if _cert_key(row) and _status_value(row) in _RISKY_STATUSES
}
valid_by_id = {_cert_key(row): row for row in (valid_certs or []) if _cert_key(row)}
skip = set(skip_certificate_ids or ())
latest: dict[str, dict] = {}
for event in history_events or []:
if not isinstance(event, dict) or str(event.get("scheme") or "") != scheme:
continue
if str(event.get("event_type") or "") not in _LIFECYCLE_EVENT_TYPES:
continue
cid = _cert_key(event)
if not cid:
continue
current = latest.get(cid)
# >= so that, within one run, the later line of the append-only log wins.
if current is None or str(event.get("detected_at") or "") >= str(current.get("detected_at") or ""):
latest[cid] = event
detected_iso = _utc_iso(detected_at)
source_label = (authority_source or "").strip() or f"{scheme} status authority"
events: list[dict] = []
for cid, last in sorted(latest.items()):
prior_status = _ADVERSE_EVENT_STATUS.get(str(last.get("event_type") or ""))
if prior_status is None or cid in still_adverse or cid in skip or cid not in valid_by_id:
continue
last_seen = str(last.get("detected_at") or "")[:10]
age = _days_since(last_seen, detected_iso)
if age is None or age > _ORPHANED_EPISODE_MAX_AGE_DAYS:
continue
cert_for_event = dict(valid_by_id[cid])
# Only the start survives from the log: its `suspended_until` was the SCHEDULED end while
# the episode ran. The actual end comes from the certificate's own register record.
cert_for_event.pop("suspended_until", None)
if prior_status == "suspended" and last.get("suspended_date"):
cert_for_event["suspended_date"] = last.get("suspended_date")
events.append(_event(
run_id, scheme, "certificate_reinstated", "medium", "explicit", cert_for_event,
detected_at=detected_at,
evidence=(f"{source_label} no longer lists this certificate as {prior_status} "
f"(last reported {prior_status} on {last_seen}), and it is in the accepted "
f"snapshot"),
old={"status": prior_status}, new={"status": ""},
))
return events
def build_register_diff_events(run_id: str, scheme: str, old_certs: list[dict], new_certs: list[dict],
diff: Mapping[str, object], *, detected_at=None) -> list[dict]:
"""Change events for registers with a plain added/removed-style diff (LCFS, EPA) --
no certificate renumbering-on-renewal and no external status authority, unlike
build_change_events's ISCC/REDcert reconciliation. `diff` is the diff_*_snapshot() output:
outcome name -> set/list of certificate_id. Recognized outcomes reuse EXISTING event types
so RED_FLAG_EVENT_TYPES and the digest labels in daily_digest.py need no changes:
added -> certificate_new (info only), retired/withdrawn -> certificate_withdrawn (explicit
status flip, red flag; LCFS says retired, Oregon/BC/CFR say withdrawn),
removed/deregistered -> certificate_removed_from_valid_list (the register
stopped listing this certificate at all, red flag).
RenovaBio adds suspended -> certificate_suspended (red flag), terminated -> certificate_revoked_or_terminated (not a red flag), expired -> certificate_expired, renewed pairs -> certificate_renewed on the successor."""
old_by_id = {_cert_key(c): c for c in (old_certs or []) if _cert_key(c)}
new_by_id = {_cert_key(c): c for c in (new_certs or []) if _cert_key(c)}
events: list[dict] = []
for cid in sorted(diff.get("added") or ()):
cert = new_by_id.get(cid)
if cert is None:
continue
events.append(_event(
run_id, scheme, "certificate_new", "low", "explicit", cert,
detected_at=detected_at, evidence=f"{scheme} accepted snapshot",
old={}, new={"status": _status_value(cert)},
))
withdrawn_ids = set(diff.get("retired") or ()) | set(diff.get("withdrawn") or ())
for cid in sorted(withdrawn_ids):
cert = new_by_id.get(cid) or old_by_id.get(cid)
if cert is None:
continue
events.append(_event(
run_id, scheme, "certificate_withdrawn", "high", "explicit", cert,
detected_at=detected_at, evidence=f"{scheme} status field flipped to withdrawn",
old={"status": _status_value(old_by_id.get(cid, {}))}, new={"status": "withdrawn"},
))
removed_ids = set(diff.get("removed") or ()) | set(diff.get("deregistered") or ())
for cid in sorted(removed_ids):
cert = old_by_id.get(cid)
if cert is None:
continue
events.append(_event(
run_id, scheme, "certificate_removed_from_valid_list", "high", "inferred", cert,
detected_at=detected_at,
evidence=f"{scheme} certificate disappeared from accepted snapshot",
old={"status": _status_value(cert)}, new={},
))
# Registers that publish an explicit lifecycle column per certificate (RenovaBio, spec
# 2026-09-29). The other registers' diffs never carry these keys.
for outcome, (event_type, severity) in (
("suspended", ("certificate_suspended", "high")),
("terminated", ("certificate_revoked_or_terminated", "low")),
("expired", ("certificate_expired", "low")),
):
for cid in sorted(diff.get(outcome) or ()):
cert = new_by_id.get(cid)
if cert is None:
continue
events.append(_event(
run_id, scheme, event_type, severity, "explicit", cert,
detected_at=detected_at, evidence=f"{scheme} lists the certificate as {outcome}",
old={"status": _status_value(old_by_id.get(cid, {}))}, new={"status": outcome},
))
for old_id, new_id in diff.get("renewed") or ():
new = new_by_id.get(new_id)
if new is None:
continue
predecessor = old_by_id.get(old_id) or new_by_id.get(old_id) or {}
events.append(_event(
run_id, scheme, "certificate_renewed", "low", "explicit", new,
detected_at=detected_at,
evidence=f"{scheme} names {new_id} as the successor of {old_id}",
old={"certificate_id": old_id, "valid_until": predecessor.get("valid_until"),
"status": _status_value(predecessor)},
new={"valid_until": new.get("valid_until"), "status": _status_value(new)},
))
return events
def _overall_status(report: dict) -> str:
if report.get("status"):
return str(report["status"])
refresh = report.get("refresh", {})
if isinstance(refresh, dict) and any(
isinstance(v, dict) and v.get("status") == "error"
for v in refresh.values()
):
return "error"
return "ok"
def _history_compact(report: dict) -> dict:
refresh = report.get("refresh", {}) if isinstance(report.get("refresh"), dict) else {}
decisions = {
k: v.get("decision", "")
for k, v in refresh.items()
if isinstance(v, dict) and v.get("decision")
}
return {
"run_id": report.get("started_at", ""),
"started_at": report.get("started_at", ""),
"finished_at": report.get("finished_at", ""),
"duration_s": report.get("duration_s"),
"status": _overall_status(report),
"decisions": decisions,
"geocode_status": (report.get("geocode") or {}).get("status", ""),
"app_db_status": (report.get("app_db") or {}).get("status", ""),
"upload_status": (report.get("upload") or {}).get("status", ""),
# Unfetched/unparsed PDF delta-queue backlog per scheme, carried verbatim from the
# caller-computed report["queue_depth"] (daily_refresh.py merges pdf_delta + audit_delta
# remaining_by_scheme). Absent for callers with no PDF delta queue (INS/US-local
# refreshes) -- {} there, never a KeyError downstream.
"queue_depth": dict(report.get("queue_depth") or {}),
}
def append_outcome_history(history_dir: str | Path, report: dict) -> None:
"""Append the finished run's compact history line, without repeating fingerprints.
A dataset publish records a non-success line before the upload so a crash mid-publish
is not a success. This second line is the outcome the monitor is allowed to trust.
"""
append_jsonl(Path(history_dir) / "refresh_history.jsonl", _history_compact(report))
def append_history_files(history_dir: str | Path, report: dict) -> None:
root = Path(history_dir)
refresh = report.get("refresh", {}) if isinstance(report.get("refresh"), dict) else {}
append_jsonl(root / "refresh_history.jsonl", _history_compact(report))
events: list[dict] = []
for payload in refresh.values():
if not isinstance(payload, dict):
continue
for record in payload.get("fingerprints", []) or []:
append_jsonl(root / "snapshot_history.jsonl", record)
for record in payload.get("compliance_history", []) or []:
append_jsonl(root / "compliance_history.jsonl", record)
events.extend(payload.get("change_events", []) or [])
for event in events:
append_jsonl(root / "change_events.jsonl", event)
by_type = Counter(str(e.get("event_type", "")) for e in events if e.get("event_type"))
by_severity = Counter(str(e.get("severity", "")) for e in events if e.get("severity"))
summary = {
"run_id": report.get("started_at", ""),
"events_total": len(events),
"events_by_type": dict(sorted(by_type.items())),
"events_by_severity": dict(sorted(by_severity.items())),
}
(root / "latest_change_summary.json").write_text(
json.dumps(summary, indent=2, sort_keys=True), encoding="utf-8"
)
_mirror_events_to_shared_log(root, events)
def _mirror_events_to_shared_log(history_dir: Path, events: list) -> None:
"""Flags and the daily mail read one log. A staging dir keeps its own copy;
the same event dicts are appended to state_paths.refresh_dir()."""
import state_paths
shared = Path(state_paths.refresh_dir())
if history_dir.resolve() == shared.resolve():
return
for event in events:
if isinstance(event, dict) and event.get("event_type"):
append_jsonl(shared / "change_events.jsonl", event)
"""Validation and durable history helpers for the daily refresh pipeline.
Pure helpers only: no network calls and no Streamlit imports.
"""
from __future__ import annotations
import csv
import hashlib
import io
import json
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path
from typing import Mapping
from sinetti_services import certificate_succession, renewal_match
from sinetti_services.event_classification import expired_before
from sinetti_services.returning_holder import classify_returning, format_off_scheme_evidence
_SEVERITY_RANK = {"ok": 0, "warning": 1, "critical": 2}
def _utc_iso(value=None) -> str:
if value is None:
value = datetime.now(timezone.utc)
if isinstance(value, str):
return value
if value.tzinfo is None:
value = value.replace(tzinfo=timezone.utc)
return value.astimezone(timezone.utc).isoformat()
def _norm_header(value: str) -> str:
return "".join(ch for ch in str(value or "").strip().lower() if ch.isalnum())
def _check(severity: str, code: str, message: str, **context) -> dict:
out = {"severity": severity, "code": code, "message": message}
if context:
out["context"] = context
return out
def _status(checks: list[dict]) -> str:
worst = "ok"
for check in checks:
sev = check.get("severity", "ok")
if _SEVERITY_RANK.get(sev, 0) > _SEVERITY_RANK[worst]:
worst = sev
return worst
def _ratio(ok: int, total: int) -> float:
return round((ok / total), 4) if total else 0.0
def _parse_date(value: str) -> bool:
raw = str(value or "").strip()
if not raw:
return False
for fmt in ("%Y-%m-%d", "%d.%m.%Y", "%d/%m/%Y"):
try:
datetime.strptime(raw, fmt)
return True
except ValueError:
pass
return False
def _collapse_check(checks: list[dict], row_count: int, previous_count: int) -> None:
if previous_count <= 0:
return
if row_count < max(1, int(previous_count * 0.5)):
checks.append(_check(
"critical", "row_count_collapse",
"Row count collapsed versus the previous accepted snapshot.",
previous_count=previous_count, row_count=row_count,
))
elif row_count > int(previous_count * 2.0):
checks.append(_check(
"warning", "row_count_spike",
"Row count is much larger than the previous accepted snapshot.",
previous_count=previous_count, row_count=row_count,
))
_ISCC_HEADER_GROUPS = {
"certificate_id": {"certificate"},
"holder_name": {"certificateholder"},
"country": {"country"},
"scope": {"scope"},
"feedstocks": {"rawmaterial"},
"valid_from": {"validfrom"},
"valid_until": {"validuntil"},
"issuing_cb": {"issuingcb", "certificationbody"},
}
def validate_iscc_csv_text(csv_text: str, *, previous_count: int = 0) -> dict: