import hashlib import json from datetime import datetime, timezone from typing import Any, Dict, List, Mapping, Optional, Sequence from .models import Claim, ClaimGraph, EvidenceToken, NarrativeBlock, NarrativeDiff, Provenance, Signal, Source def _model_dump(model: Any) -> Dict[str, Any]: if hasattr(model, "model_dump"): return model.model_dump() return model.dict() def _jsonable(value: Any) -> Any: if hasattr(value, "model_dump"): return value.model_dump() if hasattr(value, "dict"): return value.dict() if isinstance(value, dict): return {key: _jsonable(item) for key, item in value.items()} if isinstance(value, list): return [_jsonable(item) for item in value] if isinstance(value, tuple): return [_jsonable(item) for item in value] if isinstance(value, datetime): return value.isoformat() return value def _utc_naive(value: Optional[Any]) -> datetime: if value is None: return datetime.fromtimestamp(0) if isinstance(value, datetime): ts = value else: text = str(value).replace("Z", "+00:00") ts = datetime.fromisoformat(text) if ts.tzinfo is not None: ts = ts.astimezone(timezone.utc).replace(tzinfo=None) return ts def canonical_json(data: Any) -> str: return json.dumps(data, sort_keys=True, separators=(",", ":"), default=str) def _normalize_events(events: Sequence[Mapping[str, Any]]) -> List[Dict[str, Any]]: normalized: List[Dict[str, Any]] = [] for item in events: if "payload" in item and "type" in item: payload = dict(item["payload"]) event_type = str(item["type"]) ts = _utc_naive(item.get("ts") or payload.get("ts")) else: payload = dict(item) event_type = str(payload.get("type", "unknown")) ts = _utc_naive(payload.get("ts")) normalized.append({"type": event_type, "payload": payload, "ts": ts}) normalized.sort(key=lambda entry: (entry["ts"], entry["type"], canonical_json(entry["payload"]))) return normalized def build_claim_graph(block: NarrativeBlock) -> ClaimGraph: claims: List[Claim] = [] sources = [_model_dump(source) for source in block.sources] provenance = _model_dump(block.provenance) if block.provenance else None if block.signals: for signal in block.signals: signal_data = _model_dump(signal) evidence: List[EvidenceToken] = [] for source in block.sources: source_data = _model_dump(source) evidence.append( EvidenceToken( kind=source_data["type"], uri=source_data["uri"], confidence=source_data.get("confidence"), metadata={}, ) ) if signal_data.get("provenance"): evidence.append( EvidenceToken( kind="signal_provenance", uri=f"signal://{signal_data['name']}", confidence=1.0, metadata=signal_data["provenance"], ) ) claim_id = hashlib.sha256(canonical_json({"signal": signal_data, "sources": sources}).encode("utf-8")).hexdigest() statement = f"{block.topic}: {signal_data['name']}={signal_data['value']}" claims.append( Claim( id=claim_id, statement=statement, evidence=evidence, confidence=signal_data.get("value"), ) ) else: claim_id = hashlib.sha256(canonical_json({"topic": block.topic, "sources": sources}).encode("utf-8")).hexdigest() claims.append( Claim( id=claim_id, statement=f"{block.topic}: no structured signals captured", evidence=[ EvidenceToken(kind=source.type, uri=source.uri, confidence=source.confidence, metadata={}) for source in block.sources ], confidence=block.sentiment, ) ) return ClaimGraph(block_id=block.id, claims=claims, provenance=block.provenance) def render_narrative_block(block: NarrativeBlock, redacted: bool = False) -> str: graph = build_claim_graph(block) lines: List[str] = [] lines.append(f"# {block.topic}") lines.append(f"Block ID: `{block.id}`") lines.append(f"Timestamp: `{block.timestamp.isoformat()}`") if block.sentiment is not None: lines.append(f"Sentiment: `{block.sentiment}`") if block.risk_factors: lines.append("Risk factors:") for risk in sorted(block.risk_factors): lines.append(f"- {risk}") if block.scenario_flags: lines.append("Scenario flags:") for flag in sorted(block.scenario_flags): lines.append(f"- {flag}") lines.append("\n## Claims") for claim in sorted(graph.claims, key=lambda item: item.id): lines.append(f"- `{claim.id}` {claim.statement}") for evidence in sorted(claim.evidence, key=lambda item: (item.kind, item.uri)): uri = evidence.uri if not redacted else "[redacted]" lines.append(f" - {evidence.kind}: {uri}") if graph.provenance: lines.append("\n## Provenance") lines.append(f"- trace_id: `{graph.provenance.trace_id}`") if graph.provenance.signer: lines.append(f"- signer: `{graph.provenance.signer}`") if graph.provenance.ledger_anchor: lines.append(f"- ledger_anchor: `{graph.provenance.ledger_anchor}`") return "\n".join(lines).strip() + "\n" def diff_narrative_blocks(left: NarrativeBlock, right: NarrativeBlock) -> NarrativeDiff: left_claims = {claim.id: claim for claim in build_claim_graph(left).claims} right_claims = {claim.id: claim for claim in build_claim_graph(right).claims} changed_fields: List[Dict[str, Any]] = [] for field_name in ["topic", "timestamp", "sentiment", "risk_factors", "scenario_flags", "sources", "signals", "provenance"]: left_value = getattr(left, field_name) right_value = getattr(right, field_name) if left_value != right_value: changed_fields.append( { "field": field_name, "before": _jsonable(left_value), "after": _jsonable(right_value), } ) added_claims = [right_claims[claim_id] for claim_id in sorted(set(right_claims) - set(left_claims))] removed_claims = [left_claims[claim_id] for claim_id in sorted(set(left_claims) - set(right_claims))] modified_claims: List[Dict[str, Any]] = [] for claim_id in sorted(set(left_claims) & set(right_claims)): if _model_dump(left_claims[claim_id]) != _model_dump(right_claims[claim_id]): modified_claims.append( { "claim_id": claim_id, "before": _model_dump(left_claims[claim_id]), "after": _model_dump(right_claims[claim_id]), } ) return NarrativeDiff( left_block_id=left.id, right_block_id=right.id, changed_fields=changed_fields, added_claims=added_claims, removed_claims=removed_claims, modified_claims=modified_claims, ) def build_block_from_events(events: Sequence[Mapping[str, Any]]) -> NarrativeBlock: normalized = _normalize_events(events) concat = canonical_json(normalized) block_id = hashlib.sha256(concat.encode("utf-8")).hexdigest() topic = " | ".join(sorted({entry["type"] for entry in normalized if entry["type"]})) timestamp = max((entry["ts"] for entry in normalized), default=datetime.fromtimestamp(0)) sources: List[Source] = [] signals: List[Signal] = [] sentiment_vals = [] for entry in normalized: payload = entry["payload"] if entry["type"] == "news": sources.append(Source(type="news", uri=str(payload.get("uri", "")), confidence=float(payload.get("confidence", 0.0)))) sentiment_vals.append(len(payload.get("headline", "")) % 5 - 2) elif entry["type"] == "transcript": sources.append(Source(type="transcript", uri=f"transcript://{payload.get('speaker')}", confidence=1.0)) signals.append( Signal( name="word_count", value=float(len(payload.get("text", "").split())), provenance={"speaker": payload.get("speaker"), "ts": payload.get("ts")}, ) ) sentiment = sum(sentiment_vals) / len(sentiment_vals) if sentiment_vals else None provenance = Provenance(trace_id=block_id) return NarrativeBlock( id=block_id, topic=topic, timestamp=timestamp, sources=sources, signals=signals, sentiment=sentiment, provenance=provenance, )