"""Common query utilities for metrics module."""

from datetime import datetime, timedelta, UTC
from typing import Any, TypedDict

from sqlalchemy import Column, case, func
from sqlalchemy.orm import Session


# Single source of truth for period → days across all metrics endpoints.
# Covers BOTH UI vocabularies: the main dashboard / context-overflow pages
# send '3m'/'1y'; the standalone link-analytics page sends '90d'/'365d'.
# Both alias to the same day counts so every metrics endpoint agrees.
# 'all' maps to None (no time limit).
PERIOD_DAYS_MAP = {
    "7d": 7,
    "30d": 30,
    "3m": 90,
    "90d": 90,
    "1y": 365,
    "365d": 365,
    "all": None,
}


def get_period_days(period: str, default: int = 30) -> int | None:
    """Convert a period string to number of days.

    Returns None for 'all' (no time limit); unknown values use ``default``.
    """
    return PERIOD_DAYS_MAP.get(period, default)


def get_period_cutoff(period: str) -> datetime | None:
    """UTC cutoff datetime for a period string, or None for 'all'.

    Delegates to :func:`get_period_days` so there is exactly one period
    vocabulary. Unknown values default to 30 days.
    """
    days = get_period_days(period)
    if days is None:  # 'all'
        return None
    return datetime.now(UTC) - timedelta(days=days)


def get_time_filter_condition(period: str, timestamp_column: Column) -> Any:
    """Get SQLAlchemy condition for time filtering.

    Args:
        period: Time period ('7d', '30d', '3m', '1y', 'all')
        timestamp_column: SQLAlchemy timestamp column to filter on

    Returns:
        SQLAlchemy condition object or None for 'all'
    """
    cutoff = get_period_cutoff(period)
    if cutoff is None:
        return None
    return timestamp_column >= cutoff


class TruncationSummary(TypedDict):
    """Aggregated context-overflow stats over a time window.

    Token-summary fields (total_tokens, prompt/completion sums and avgs,
    max_prompt_tokens) are computed in the same query because they share
    the same row set and time window — one scan instead of three.
    """

    total_requests: int
    requests_with_context: int
    truncated_requests: int
    truncation_rate: float  # raw percentage, unrounded
    avg_tokens_truncated: float  # raw, unrounded
    total_tokens: int
    total_prompt_tokens: int
    total_completion_tokens: int
    avg_prompt_tokens: float  # raw, unrounded
    avg_completion_tokens: float  # raw, unrounded
    max_prompt_tokens: int


def get_context_overflow_truncation_summary(
    session: Session, period: str, research_mode: str = "all"
) -> TruncationSummary:
    """Single-source aggregation of truncation + token-summary stats.

    Both /metrics/api/metrics and /metrics/api/context-overflow surface
    the same truncation rate; computing it in two places risks the two
    summaries silently disagreeing for the same time window.

    Truncation fields and token-summary fields are computed in one merged
    query (CASE-based AVG isolates truncated rows; SQL AVG ignores NULLs).
    Returns raw values — callers round/cast for their own display contract.

    research_mode defaults to "all" (no mode filter). Pass "quick" or
    "detailed" to scope to that mode — matches the rest of api_metrics()
    so the dashboard's mode toggle stays in sync with this panel.
    """
    from ..database.models import TokenUsage

    time_condition = get_time_filter_condition(period, TokenUsage.timestamp)
    mode_condition = get_research_mode_condition(
        research_mode, TokenUsage.research_mode
    )

    base = session.query(TokenUsage)
    if time_condition is not None:
        base = base.filter(time_condition)
    if mode_condition is not None:
        base = base.filter(mode_condition)

    row = base.with_entities(
        func.count(TokenUsage.id).label("total_requests"),
        func.sum(
            case((TokenUsage.context_limit.isnot(None), 1), else_=0)
        ).label("requests_with_context"),
        func.sum(
            case((TokenUsage.context_truncated.is_(True), 1), else_=0)
        ).label("truncated_requests"),
        func.avg(
            case(
                (
                    TokenUsage.context_truncated.is_(True),
                    TokenUsage.tokens_truncated,
                ),
                else_=None,
            )
        ).label("avg_tokens_truncated"),
        func.coalesce(func.sum(TokenUsage.total_tokens), 0).label(
            "total_tokens"
        ),
        func.coalesce(func.sum(TokenUsage.prompt_tokens), 0).label(
            "total_prompt_tokens"
        ),
        func.coalesce(func.sum(TokenUsage.completion_tokens), 0).label(
            "total_completion_tokens"
        ),
        func.avg(TokenUsage.prompt_tokens).label("avg_prompt_tokens"),
        func.avg(TokenUsage.completion_tokens).label("avg_completion_tokens"),
        func.max(TokenUsage.prompt_tokens).label("max_prompt_tokens"),
    ).first()

    total_requests = int(row.total_requests or 0) if row else 0
    requests_with_context = int(row.requests_with_context or 0) if row else 0
    truncated_requests = int(row.truncated_requests or 0) if row else 0

    truncation_rate = (
        (truncated_requests / requests_with_context) * 100
        if requests_with_context > 0
        else 0.0
    )

    avg_tokens_truncated = float(row.avg_tokens_truncated or 0) if row else 0.0

    return TruncationSummary(
        total_requests=total_requests,
        requests_with_context=requests_with_context,
        truncated_requests=truncated_requests,
        truncation_rate=truncation_rate,
        avg_tokens_truncated=avg_tokens_truncated,
        total_tokens=int(row.total_tokens or 0) if row else 0,
        total_prompt_tokens=int(row.total_prompt_tokens or 0) if row else 0,
        total_completion_tokens=(
            int(row.total_completion_tokens or 0) if row else 0
        ),
        avg_prompt_tokens=float(row.avg_prompt_tokens or 0) if row else 0.0,
        avg_completion_tokens=(
            float(row.avg_completion_tokens or 0) if row else 0.0
        ),
        max_prompt_tokens=int(row.max_prompt_tokens or 0) if row else 0,
    )


def get_research_mode_condition(research_mode: str, mode_column: Column) -> Any:
    """Get SQLAlchemy condition for research mode filtering.

    Args:
        research_mode: Research mode ('quick', 'detailed', 'all')
        mode_column: SQLAlchemy column to filter on

    Returns:
        SQLAlchemy condition object or None for 'all'
    """
    if research_mode == "all":
        return None
    if research_mode in ["quick", "detailed"]:
        return mode_column == research_mode
    return None
