64 lines
2.3 KiB
Python
64 lines
2.3 KiB
Python
from typing import Dict, Any, List
|
|
from .models import NarrativeBlock, Source, Signal, Provenance
|
|
from datetime import datetime
|
|
import uuid
|
|
|
|
|
|
def news_adapter_event(headline: str, uri: str, confidence: float = 0.9) -> Dict[str, Any]:
|
|
"""Create a normalized event payload from a news headline. This is an adapter stub.
|
|
|
|
In real usage this would parse entities, extract sentiment, etc. Here we keep a simple,
|
|
deterministic mapping for deterministic replay testing.
|
|
"""
|
|
return {
|
|
"type": "news",
|
|
"headline": headline,
|
|
"uri": uri,
|
|
"confidence": float(confidence),
|
|
}
|
|
|
|
|
|
def transcript_adapter_event(speaker: str, text: str, ts: str) -> Dict[str, Any]:
|
|
return {"type": "transcript", "speaker": speaker, "text": text, "ts": ts}
|
|
|
|
|
|
def build_block_from_events(events: List[Dict[str, Any]]) -> NarrativeBlock:
|
|
"""Deterministically build a NarrativeBlock from a list of events.
|
|
|
|
This function demonstrates normalization logic: aggregating sources and signals.
|
|
"""
|
|
# derive an id deterministically from the concatenated event payloads
|
|
concat = "".join([str(e) for e in events])
|
|
block_id = uuid.uuid5(uuid.NAMESPACE_URL, concat).hex
|
|
topic = " | ".join(set([e.get("type") for e in events if e.get("type")]))
|
|
ts = datetime.utcnow()
|
|
sources = []
|
|
signals = []
|
|
sentiment_vals = []
|
|
for e in events:
|
|
if e.get("type") == "news":
|
|
sources.append(Source(type="news", uri=e.get("uri"), confidence=e.get("confidence")))
|
|
# naive deterministic sentiment from headline length
|
|
sentiment_vals.append(len(e.get("headline", "")) % 5 - 2)
|
|
if e.get("type") == "transcript":
|
|
sources.append(Source(type="transcript", uri=f"transcript://{e.get('speaker')}", confidence=1.0))
|
|
# a toy signal: word count
|
|
signals.append(Signal(name="word_count", value=float(len(e.get("text", "").split())), provenance={"speaker": e.get("speaker")}))
|
|
|
|
sentiment = None
|
|
if sentiment_vals:
|
|
sentiment = sum(sentiment_vals) / len(sentiment_vals)
|
|
|
|
prov = Provenance(trace_id=block_id)
|
|
|
|
block = NarrativeBlock(
|
|
id=block_id,
|
|
topic=topic,
|
|
timestamp=ts,
|
|
sources=sources,
|
|
signals=signals,
|
|
sentiment=sentiment,
|
|
provenance=prov,
|
|
)
|
|
return block
|