api/routes/trending.py

"""
Trending causes/topics API endpoints.

Reads from the bronze ingestion tables landed by:
  - ingestion.everyorg.causes  -> bronze.bronze_everyorg_causes
  - ingestion.ntee.codes       -> bronze.bronze_ntee_codes

Falls back to the public.tag mart if the bronze migration has
not been run yet (during the Stage-2 transition).
"""
import os
from contextlib import contextmanager
from pathlib import Path
from typing import List, Optional

import psycopg2
from fastapi import APIRouter, Query
from loguru import logger
from pydantic import BaseModel

from api.routes.search_postgres import normalize_state_input

router = APIRouter(prefix="/api/trending", tags=["trending"])

LOCAL_DB_URL = os.getenv(
    "NEON_DATABASE_URL_DEV",
    "postgresql://postgres:password@localhost:5433/open_navigator",
)


@contextmanager
def _cursor():
    conn = psycopg2.connect(LOCAL_DB_URL)
    try:
        with conn.cursor() as cur:
            yield cur
    finally:
        conn.close()


_NTEE_ICON_MAP = {
    'A': '🎨', 'B': '📚', 'C': '🌍', 'D': '🐾', 'E': '⚕️',
    'F': '🏥', 'G': '🏛️', 'H': '🏥', 'I': '🔬', 'J': '👥',
    'K': '🍽️', 'L': '🏠', 'M': '🛡️', 'N': '🎯', 'O': '⚖️',
    'P': '👶', 'Q': '🌐', 'R': '🏛️', 'S': '🤝', 'T': '💼',
    'U': '🔬', 'V': '⚡', 'W': '📢', 'X': '🏛️', 'Y': '🏛️',
    'Z': '🔤',
}


class CauseItem(BaseModel):
    """A trending cause/topic"""
    # Stable slug (EveryOrg cause id, e.g. 'education') / NTEE code. The frontend
    # passes this as ?cause_id= to scope decisions/meetings to the cause.
    cause_id: Optional[str] = None
    name: str
    icon: str
    category: str
    description: Optional[str] = None
    image_url: Optional[str] = None
    popularity_rank: Optional[int] = None
    # Real count of analyzed meetings whose transcript matches this cause's
    # keyword set (gold.meeting_cause_link). 0 is honest — no fabricated figure.
    meeting_count: int = 0


class TrendingResponse(BaseModel):
    """Response with trending causes"""
    causes: List[CauseItem]
    total: int


class TrendingTopicItem(BaseModel):
    """A trending legislative topic (bill subject)."""
    name: str
    bill_count: int
    recent_bill_count: int
    state_code: Optional[str] = None


class TrendingTopicsResponse(BaseModel):
    """Response with trending legislative topics."""
    topics: List[TrendingTopicItem]
    total: int


# Causes joined to their real meeting count, ordered most-discussed first. The
# count comes from gold.meeting_cause_link (a transcript full-text match per
# cause). NULL meeting_count is coalesced to 0 (honest empty), never fabricated.
_CAUSES_WITH_COUNT_SQL = """
    SELECT c.cause_id, c.cause_name, c.description, c.icon, c.category,
           c.popularity_rank,
           COALESCE(mc.meeting_count, 0) AS meeting_count
    FROM bronze.bronze_everyorg_causes c
    LEFT JOIN (
        SELECT cause_id, COUNT(DISTINCT event_meeting_id) AS meeting_count
        FROM gold.meeting_cause_link
        WHERE (%(state)s IS NULL OR state_code = %(state)s)
          AND (%(city)s  IS NULL OR lower(jurisdiction_name) = lower(%(city)s)
                                 OR lower(city) = lower(%(city)s))
        GROUP BY cause_id
    ) mc ON mc.cause_id = c.cause_id
    -- When a place is in scope, keep only causes actually discussed there so the
    -- pills reflect that place (no global causes with a 0 local count).
    WHERE (%(state)s IS NULL AND %(city)s IS NULL) OR mc.meeting_count > 0
    ORDER BY COALESCE(mc.meeting_count, 0) DESC,
             c.popularity_rank NULLS LAST, c.cause_name
    LIMIT %(limit)s
"""

# Fallback when gold.meeting_cause_link is absent (e.g. a serving DB where the
# linkage mart hasn't been built). Causes still render, just with no count.
_CAUSES_NO_COUNT_SQL = """
    SELECT cause_id, cause_name, description, icon, category, popularity_rank, 0 AS meeting_count
    FROM bronze.bronze_everyorg_causes
    ORDER BY popularity_rank NULLS LAST, cause_name
    LIMIT %(limit)s
"""


def get_everyorg_causes(
    limit: int = 20,
    state: Optional[str] = None,
    city: Optional[str] = None,
) -> List[CauseItem]:
    """Load EveryOrg causes (with real per-cause meeting counts) from the warehouse.

    When ``state``/``city`` are given, the per-cause meeting_count is scoped to
    that place (gold.meeting_cause_link carries state_code/jurisdiction_name/city)
    and only causes actually discussed there are returned — so Browse Causes shows
    a place-specific pill set, mirroring the Topics flow.
    """
    norm_state = normalize_state_input(state) if state else None
    norm_city = (city or "").strip() or None
    params = {"state": norm_state, "city": norm_city, "limit": limit}
    rows: list = []
    try:
        with _cursor() as cur:
            try:
                cur.execute(_CAUSES_WITH_COUNT_SQL, params)
                rows = cur.fetchall()
            except psycopg2.Error:
                # Linkage mart missing/unreadable — fall back to the plain list so
                # causes never disappear. New cursor: the prior tx is aborted.
                cur.connection.rollback()
                cur.execute(_CAUSES_NO_COUNT_SQL, {"limit": limit})
                rows = cur.fetchall()
    except Exception as exc:
        logger.warning(f"EveryOrg causes query failed (table empty or missing?): {exc}")
        return []

    return [
        CauseItem(
            cause_id=cause_id,
            name=name,
            icon=icon or '📌',
            category=category or 'general',
            description=description or '',
            image_url=f"/images/causes/everyorg_{cause_id}_square.png",
            popularity_rank=popularity_rank,
            meeting_count=int(meeting_count or 0),
        )
        for (cause_id, name, description, icon, category, popularity_rank, meeting_count) in rows
    ]


def get_ntee_causes(limit: int = 20, level: Optional[int] = None) -> List[CauseItem]:
    """Load NTEE code causes from bronze.bronze_ntee_codes (or the public.tag mart).

    Args:
        limit: max rows to return.
        level: 1 = major groups (single-letter codes), 2 = divisions (3-char codes).
    """
    # `level` maps directly to code length: A = level 1, A20 = level 2.
    bronze_where = ["cause_type = 'ntee'"]
    # public.tag filters NTEE rows via vocabulary; code lives in source_code.
    mart_where = ["vocabulary = 'ntee'"]
    params: list = []
    if level == 1:
        bronze_where.append("length(code) = 1")
        mart_where.append("length(source_code) = 1")
    elif level == 2:
        bronze_where.append("length(code) = 3")
        mart_where.append("length(source_code) = 3")

    bronze_sql = f"""
        SELECT code, name, description, category
        FROM bronze.bronze_ntee_codes
        WHERE {' AND '.join(bronze_where)}
        ORDER BY length(code), code
        LIMIT %s
    """
    # Fall back to the public.tag mart (supersedes the legacy cause_ntee table).
    mart_sql = f"""
        SELECT source_code AS code, label AS name, description, category
        FROM public.tag
        WHERE {' AND '.join(mart_where)}
        ORDER BY length(source_code), source_code
        LIMIT %s
    """
    params.append(limit)

    rows = _query_with_legacy_fallback(
        primary_sql=bronze_sql,
        legacy_sql=mart_sql,
        params=params,
    )
    if rows is None:
        return []

    out: List[CauseItem] = []
    for code, name, description, category in rows:
        icon = _NTEE_ICON_MAP.get((code or 'Z')[0], '📌')
        out.append(CauseItem(
            name=name or code,
            icon=icon,
            category=category or 'ntee',
            description=description or f"NTEE Code {code}",
            image_url=f"/images/causes/ntee_{code}_square.png",
            popularity_rank=None,
        ))
    return out


def _query_with_legacy_fallback(primary_sql: str, legacy_sql: str, params):
    """Try the bronze table first; fall back to the public.tag mart.

    Returns None if both fail (so the caller can render an empty response
    instead of 500ing).
    """
    try:
        with _cursor() as cur:
            cur.execute(primary_sql, params)
            return cur.fetchall()
    except psycopg2.errors.UndefinedTable:
        logger.info("bronze.bronze_ntee_codes not found, falling back to public.tag")
    except Exception as exc:
        logger.warning(f"Primary trending query failed: {exc}")
        return None

    try:
        with _cursor() as cur:
            cur.execute(legacy_sql, params)
            return cur.fetchall()
    except Exception as exc:
        logger.warning(f"Legacy trending query failed: {exc}")
        return None


@router.get("", response_model=TrendingResponse)
async def get_trending_causes(
    source: str = Query("everyorg", description="Source: 'everyorg', 'ntee', or 'mixed'"),
    limit: int = Query(12, ge=1, le=100, description="Max number of causes to return"),
    level: Optional[int] = Query(None, description="NTEE level filter (1 or 2)"),
    state: Optional[str] = Query(None, description="2-letter code or full state name; scopes EveryOrg cause meeting counts to that state (omit for national)"),
    city: Optional[str] = Query(None, description="City/jurisdiction name; further scopes EveryOrg cause meeting counts to that place"),
) -> TrendingResponse:
    """
    Get trending causes for homepage.

    Reads from bronze tables landed by ingestion.everyorg.causes and
    ingestion.ntee.codes.

    **Sources:**
    - `everyorg`: Curated popular causes
    - `ntee`: IRS nonprofit categories
    - `mixed`: Interleaved from both
    """
    if source == "ntee":
        causes = get_ntee_causes(limit=limit, level=level)
    elif source == "mixed":
        half = limit // 2
        everyorg = get_everyorg_causes(limit=half, state=state, city=city)
        ntee = get_ntee_causes(limit=half, level=1)
        causes = []
        for i in range(max(len(everyorg), len(ntee))):
            if i < len(everyorg):
                causes.append(everyorg[i])
            if i < len(ntee):
                causes.append(ntee[i])
    else:
        causes = get_everyorg_causes(limit=limit, state=state, city=city)

    return TrendingResponse(causes=causes, total=len(causes))


@router.get("/topics", response_model=TrendingTopicsResponse)
async def get_trending_topics(
    state: Optional[str] = Query(
        None, description="2-letter code or full state name; omit for national"
    ),
    limit: int = Query(12, ge=1, le=50, description="Max number of topics to return"),
) -> TrendingTopicsResponse:
    """
    Get trending legislative topics for the homepage trending pills.

    Reads the modeled ``public.rpt_bill_map_aggregate`` mart. The trending
    signal is ``recent_bill_count`` (bills with recent action).

    - **state** given: per-state subjects filtered by ``state_code``.
    - **state** omitted: subjects aggregated across all jurisdictions.
    """
    norm_state = normalize_state_input(state)

    if norm_state:
        sql = """
            SELECT subject, bill_count, recent_bill_count, state_code
            FROM public.rpt_bill_map_aggregate
            WHERE state_code = %s
              AND subject IS NOT NULL
              AND btrim(subject) <> ''
            ORDER BY recent_bill_count DESC, bill_count DESC
            LIMIT %s
        """
        params: tuple = (norm_state, limit)
    else:
        sql = """
            SELECT subject,
                   sum(bill_count) AS bill_count,
                   sum(recent_bill_count) AS recent_bill_count,
                   NULL AS state_code
            FROM public.rpt_bill_map_aggregate
            WHERE subject IS NOT NULL
              AND btrim(subject) <> ''
            GROUP BY subject
            ORDER BY sum(recent_bill_count) DESC, sum(bill_count) DESC
            LIMIT %s
        """
        params = (limit,)

    try:
        with _cursor() as cur:
            cur.execute(sql, params)
            rows = cur.fetchall()
    except psycopg2.errors.UndefinedTable:
        logger.warning(
            "public.rpt_bill_map_aggregate not found; returning empty trending topics"
        )
        return TrendingTopicsResponse(topics=[], total=0)
    except Exception as exc:
        logger.warning(f"Trending topics query failed: {exc}")
        return TrendingTopicsResponse(topics=[], total=0)

    topics = [
        TrendingTopicItem(
            name=subject,
            bill_count=int(bill_count or 0),
            recent_bill_count=int(recent_bill_count or 0),
            state_code=state_code,
        )
        for (subject, bill_count, recent_bill_count, state_code) in rows
    ]
    return TrendingTopicsResponse(topics=topics, total=len(topics))


@router.get("/stats")
async def get_trending_stats():
    """Get stats about available causes."""
    everyorg_count = 0
    ntee_count = 0

    try:
        with _cursor() as cur:
            cur.execute("SELECT count(*) FROM bronze.bronze_everyorg_causes")
            everyorg_count = cur.fetchone()[0]
    except Exception as exc:
        logger.warning(f"EveryOrg count failed: {exc}")

    try:
        with _cursor() as cur:
            cur.execute(
                "SELECT count(*) FROM bronze.bronze_ntee_codes WHERE cause_type = 'ntee'"
            )
            ntee_count = cur.fetchone()[0]
    except psycopg2.errors.UndefinedTable:
        # Stage-2 transition: bronze table not yet created, fall back to the
        # public.tag mart (NTEE rows filtered via vocabulary).
        try:
            with _cursor() as cur:
                cur.execute("SELECT count(*) FROM public.tag WHERE vocabulary = 'ntee'")
                ntee_count = cur.fetchone()[0]
        except Exception as exc:
            logger.warning(f"NTEE tag-mart count failed: {exc}")
    except Exception as exc:
        logger.warning(f"NTEE count failed: {exc}")

    images_dir = Path("data/media/causes")
    images_count = (
        len(list(images_dir.glob("*_square.png"))) if images_dir.exists() else 0
    )

    return {
        "everyorg_causes": everyorg_count,
        "ntee_causes": ntee_count,
        "total_causes": everyorg_count + ntee_count,
        "generated_images": images_count,
    }