91 lines
2.6 KiB
Python
91 lines
2.6 KiB
Python
|
"""Pure parser for the news-summarizer reduce JSON / geo / importance contract."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
import re
|
||
|
|
|
||
|
|
_EMPTY = {"summary_en": "", "ticker": [], "map_items": []}
|
||
|
|
_KEEP = frozenset({"critical", "high"})
|
||
|
|
_THINK_RE = re.compile(r"<think>.*?</think>", re.DOTALL)
|
||
|
|
_FENCE_RE = re.compile(r"```(?:json)?", re.IGNORECASE)
|
||
|
|
|
||
|
|
TICKER_HEADLINE_MAX = 140
|
||
|
|
MAP_HEADLINE_MAX = 160
|
||
|
|
TICKER_CAP = 12
|
||
|
|
|
||
|
|
|
||
|
|
def parse_reduce_json(raw: str) -> dict:
|
||
|
|
try:
|
||
|
|
text = _THINK_RE.sub("", raw or "")
|
||
|
|
text = _FENCE_RE.sub("", text)
|
||
|
|
start = text.find("{")
|
||
|
|
end = text.rfind("}")
|
||
|
|
if start == -1 or end == -1 or end < start:
|
||
|
|
return dict(_EMPTY)
|
||
|
|
data = json.loads(text[start : end + 1])
|
||
|
|
if not isinstance(data, dict):
|
||
|
|
return dict(_EMPTY)
|
||
|
|
summary = data.get("summary_en", "")
|
||
|
|
ticker = data.get("ticker", [])
|
||
|
|
map_items = data.get("map_items", [])
|
||
|
|
return {
|
||
|
|
"summary_en": summary if isinstance(summary, str) else "",
|
||
|
|
"ticker": ticker if isinstance(ticker, list) else [],
|
||
|
|
"map_items": map_items if isinstance(map_items, list) else [],
|
||
|
|
}
|
||
|
|
except Exception:
|
||
|
|
return dict(_EMPTY)
|
||
|
|
|
||
|
|
|
||
|
|
def clamp_coords(lat, lon) -> tuple[float, float] | None:
|
||
|
|
try:
|
||
|
|
lat_f = float(lat)
|
||
|
|
lon_f = float(lon)
|
||
|
|
except (TypeError, ValueError):
|
||
|
|
return None
|
||
|
|
if not (-90 <= lat_f <= 90 and -180 <= lon_f <= 180):
|
||
|
|
return None
|
||
|
|
return (lat_f, lon_f)
|
||
|
|
|
||
|
|
|
||
|
|
def _trimmed_headline(row: dict, limit: int) -> str:
|
||
|
|
headline = row.get("headline") or ""
|
||
|
|
if not isinstance(headline, str):
|
||
|
|
headline = str(headline)
|
||
|
|
return headline.strip()[:limit]
|
||
|
|
|
||
|
|
|
||
|
|
def select_ticker(rows: list) -> list:
|
||
|
|
out = []
|
||
|
|
for row in rows:
|
||
|
|
if row.get("importance") not in _KEEP:
|
||
|
|
continue
|
||
|
|
headline = _trimmed_headline(row, TICKER_HEADLINE_MAX)
|
||
|
|
if not headline:
|
||
|
|
continue
|
||
|
|
item = dict(row)
|
||
|
|
item["headline"] = headline
|
||
|
|
out.append(item)
|
||
|
|
if len(out) >= TICKER_CAP:
|
||
|
|
break
|
||
|
|
return out
|
||
|
|
|
||
|
|
|
||
|
|
def select_map(items: list) -> list:
|
||
|
|
out = []
|
||
|
|
for row in items:
|
||
|
|
if row.get("importance") not in _KEEP:
|
||
|
|
continue
|
||
|
|
headline = _trimmed_headline(row, MAP_HEADLINE_MAX)
|
||
|
|
if not headline:
|
||
|
|
continue
|
||
|
|
coords = clamp_coords(row.get("lat"), row.get("lon"))
|
||
|
|
if coords is None:
|
||
|
|
continue
|
||
|
|
item = dict(row)
|
||
|
|
item["headline"] = headline
|
||
|
|
item["lat"], item["lon"] = coords
|
||
|
|
out.append(item)
|
||
|
|
return out
|