eventflow/backend/fastapi_app/routers/ingest.py

import json
from datetime import date

from fastapi import APIRouter, HTTPException

from ..services.llm_extract import extract_event
from ..services.market_moves import fetch_moves_via_qfr
from ..services.store import (
    ensure_schema,
    get_store,
    insert_event_result,
    insert_raw_item,
)

router = APIRouter()


def _today() -> str:
    return date.today().isoformat()


@router.post("/rss")
def ingest_rss(payload: dict):
    """Ingest one or many RSS items.

    Expected payload:
      {"items": [{"title":..., "url":..., "published_at":..., "summary":..., "lang":...}, ...]}
    """
    items = payload.get("items")
    if not isinstance(items, list) or not items:
        raise HTTPException(status_code=400, detail="payload.items must be a non-empty list")

    st = get_store()
    conn = st.connect()
    ensure_schema(conn)

    n = 0
    for it in items:
        if not isinstance(it, dict):
            continue
        insert_raw_item(
            conn,
            source="rss",
            item_date=payload.get("date") or _today(),
            title=(it.get("title") or "")[:500],
            content=(it.get("summary") or it.get("content") or "")[:20_000],
            url=it.get("url"),
            published_at=it.get("published_at"),
            lang=it.get("lang"),
        )
        n += 1

    return {"ok": True, "inserted": n}


@router.post("/macro")
def ingest_macro(payload: dict):
    items = payload.get("items")
    if not isinstance(items, list) or not items:
        raise HTTPException(status_code=400, detail="payload.items must be a non-empty list")

    st = get_store()
    conn = st.connect()
    ensure_schema(conn)

    n = 0
    for it in items:
        if not isinstance(it, dict):
            continue
        insert_raw_item(
            conn,
            source="macro",
            item_date=payload.get("date") or _today(),
            title=(it.get("title") or "")[:500],
            content=(it.get("content") or "")[:20_000],
            url=it.get("url"),
            published_at=it.get("published_at"),
            lang=it.get("lang"),
        )
        n += 1

    return {"ok": True, "inserted": n}


@router.post("/market_moves")
def ingest_market_moves(payload: dict):
    items = payload.get("items")
    if not isinstance(items, list) or not items:
        raise HTTPException(status_code=400, detail="payload.items must be a non-empty list")

    st = get_store()
    conn = st.connect()
    ensure_schema(conn)

    n = 0
    for it in items:
        if not isinstance(it, dict):
            continue
        insert_raw_item(
            conn,
            source="market_moves",
            item_date=payload.get("date") or _today(),
            title=(it.get("title") or "")[:500],
            content=(it.get("content") or "")[:20_000],
            url=it.get("url"),
            published_at=it.get("published_at"),
            lang=it.get("lang"),
        )
        n += 1

    return {"ok": True, "inserted": n}


@router.post("/market_moves/run")
def run_market_moves(payload: dict | None = None):
    """Generate daily market-move items from QFR raw data and parse them into events."""

    payload = payload or {}
    day = str(payload.get("date") or _today())
    # QFR raw data uses trade_date like YYYYMMDD.
    trade_date = day.replace("-", "")

    st = get_store()
    conn = st.connect()
    ensure_schema(conn)

    data = fetch_moves_via_qfr(trade_date=trade_date, symbols=payload.get("symbols"))
    if not data.get("ok"):
        raise HTTPException(status_code=500, detail=data)

    inserted = 0
    parsed_ok = 0
    parsed_err = 0

    for mv in data.get("moves", []):
        sym = mv.get("symbol")
        td = mv.get("trade_date")
        ret_1d = mv.get("ret_1d")
        vol_20d = mv.get("vol_20d")
        z_1d = mv.get("z_1d")

        title = f"Market move {sym} {td}: ret_1d={ret_1d:.4f} z_1d={z_1d:.2f}" if isinstance(ret_1d, (int, float)) and isinstance(z_1d, (int, float)) else f"Market move {sym} {td}"
        content = (
            f"symbol={sym}\n"
            f"trade_date={td}\n"
            f"prev_trade_date={mv.get('prev_trade_date')}\n"
            f"close={mv.get('close')} prev_close={mv.get('prev_close')}\n"
            f"ret_1d={ret_1d} vol_20d={vol_20d} z_1d={z_1d}\n"
            "Interpretation task: explain the most likely macro/industry drivers for this move and which assets could be affected."
        )

        raw_item_id = insert_raw_item(
            conn,
            source="market_moves",
            item_date=day,
            title=title[:500],
            content=content[:20_000],
            url=None,
            published_at=None,
            lang="en",
        )
        inserted += 1

        try:
            res = extract_event(title=title, content=content, lang_hint="en")
        except Exception as e:
            # Network/provider errors should not abort the whole batch.
            insert_event_result(
                conn,
                raw_item_id=raw_item_id,
                model="",
                ok=False,
                event_json=None,
                error=f"llm_exception:{type(e).__name__}",
            )
            parsed_err += 1
            continue

        if res.get("ok") is True:
            insert_event_result(
                conn,
                raw_item_id=raw_item_id,
                model=str(res.get("model") or ""),
                ok=True,
                event_json=json.dumps(res.get("event"), ensure_ascii=True),
                error=None,
            )
            parsed_ok += 1
        else:
            err = str(res.get("error") or "unknown")
            if err == "llm_failed":
                # Keep a short hint to debug gateway flakiness without dumping secrets.
                exc = str(res.get("exc") or "")
                if exc:
                    err = f"{err}:{exc}"
            insert_event_result(
                conn,
                raw_item_id=raw_item_id,
                model=str(res.get("model") or ""),
                ok=False,
                event_json=None,
                error=err,
            )
            parsed_err += 1

    return {
        "ok": True,
        "date": day,
        "inserted": inserted,
        "parsed_ok": parsed_ok,
        "parsed_err": parsed_err,
        "errors": data.get("errors", []),
        "symbols": data.get("symbols"),
    }