osint-dashboard/news/summerizer/intel.py

"""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
MAP_CAP = 20


def _impact_to_importance(level) -> str | None:
    lv = str(level or "").strip().lower()
    if lv in {"high", "critical"}:
        return "critical" if lv == "critical" else "high"
    if lv == "medium":
        return "high"
    return None


def _from_market_intel(data: dict) -> dict:
    """Map the k8s deepseek reduce JSON onto summary_en / ticker / map_items."""
    raw_ov = data.get("market_overview")
    ov: dict = raw_ov if isinstance(raw_ov, dict) else {}
    raw_geo = data.get("geopolitical_osint")
    geo: dict = raw_geo if isinstance(raw_geo, dict) else {}
    raw_events = ov.get("critical_events")
    events: list = raw_events if isinstance(raw_events, list) else []
    raw_themes = ov.get("key_themes")
    themes: list = raw_themes if isinstance(raw_themes, list) else []
    raw_conflicts = geo.get("active_conflicts")
    conflicts: list = raw_conflicts if isinstance(raw_conflicts, list) else []

    parts: list[str] = []
    sentiment = ov.get("overall_sentiment")
    if sentiment:
        score = ov.get("sentiment_score", "")
        parts.append(f"**Sentiment:** {sentiment}" + (f" ({score})" if score != "" else ""))
    if themes:
        parts.append("**Themes:** " + ", ".join(str(t) for t in themes[:5] if t))
    for ev in events[:8]:
        if not isinstance(ev, dict):
            continue
        headline = (ev.get("headline") or "").strip()
        desc = (ev.get("description") or "").strip()
        if headline and desc:
            parts.append(f"### {headline}\n{desc}")
        elif headline:
            parts.append(f"### {headline}")
        elif desc:
            parts.append(desc)
    for c in conflicts[:4]:
        if not isinstance(c, dict):
            continue
        intel = (c.get("intelligence") or "").strip()
        region = (c.get("region") or "").strip()
        if intel:
            parts.append(f"**{region or 'Conflict'}:** {intel}")
    summary_en = "\n\n".join(parts)

    ticker: list[dict] = []
    for ev in events:
        if not isinstance(ev, dict):
            continue
        importance = _impact_to_importance(ev.get("impact_level"))
        if not importance:
            continue
        headline = (ev.get("headline") or "").strip()
        if not headline:
            continue
        ticker.append({
            "headline": headline,
            "importance": importance,
            "url": ev.get("url") or "",
            "location_name": ev.get("location_name") or ev.get("region") or "",
        })

    map_items: list[dict] = []
    for c in conflicts:
        if not isinstance(c, dict):
            continue
        status = str(c.get("status") or "").lower()
        importance = "critical" if status == "escalating" else "high"
        headline = (c.get("intelligence") or c.get("region") or "").strip()
        if not headline:
            continue
        map_items.append({
            "headline": headline,
            "importance": importance,
            "location_name": c.get("location_name") or c.get("region") or "",
            "lat": c.get("lat"),
            "lon": c.get("lon"),
            "location_confidence": "region",
            "category": "military/conflict",
            "url": "",
        })
    for ev in events:
        if not isinstance(ev, dict):
            continue
        importance = _impact_to_importance(ev.get("impact_level"))
        if not importance:
            continue
        if clamp_coords(ev.get("lat"), ev.get("lon")) is None:
            continue
        headline = (ev.get("headline") or "").strip()
        if not headline:
            continue
        map_items.append({
            "headline": headline,
            "importance": importance,
            "location_name": ev.get("location_name") or "",
            "lat": ev.get("lat"),
            "lon": ev.get("lon"),
            "location_confidence": "city",
            "category": ev.get("category") or "other",
            "url": ev.get("url") or "",
        })
    return {"summary_en": summary_en, "ticker": ticker, "map_items": map_items}


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)
        if isinstance(data.get("summary_en"), str) or isinstance(data.get("ticker"), list):
            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 [],
            }
        if "market_overview" in data or "geopolitical_osint" in data:
            return _from_market_intel(data)
        return dict(_EMPTY)
    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)
        if len(out) >= MAP_CAP:
            break
    return out