232 lines
8.9 KiB
Python
232 lines
8.9 KiB
Python
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,
|
|
)
|