"""analyze_trend — time-series trend over a period (KM-608).

An analytical "family" tool: in ONE call it buckets rows into time periods
(day/week/month/quarter/year), aggregates a value per period, and summarizes
the movement (first vs last, absolute & percent change, direction, linear
slope). Answers questions like "how did revenue trend month over month?".

STATUS: compute layer only — the function takes an already-materialized
DataFrame. The wrapper layer (fetching data from the catalog via source_id,
the ToolOutput envelope, ToolSpec registration) is added once the Planner
seam (KM-418) is settled. Keeping compute separate from data-fetching makes
this function easy to unit-test in isolation and stable when wrapped.
"""

from __future__ import annotations

import numpy as np
import pandas as pd

from src.tools.analytics.descriptive import ColumnNotFoundError

# Friendly period name -> pandas resample rule. Using the non-deprecated
# pandas 2.2 codes (ME/QE/YE) avoids FutureWarnings.
FREQ_MAP = {
    "day": "D",
    "week": "W",
    "month": "ME",
    "quarter": "QE",
    "year": "YE",
}

# How to aggregate the value within each period.
SUPPORTED_AGGS = ("sum", "mean", "count", "min", "max", "median")


class InvalidFrequencyError(ValueError):
    """The requested period is not in FREQ_MAP (maps to error_code INVALID_FREQUENCY)."""


class UnsupportedAggregationError(ValueError):
    """The requested aggregation is not supported (maps to error_code UNSUPPORTED_AGG)."""


class InvalidDateColumnError(ValueError):
    """date_column holds numeric values that aren't a recognizable date/year/month
    (maps to error_code INVALID_DATE_COLUMN)."""


def _clean(value: object) -> object:
    """Convert numpy scalars to plain Python; NaN -> None for JSON-clean output."""
    if value is None:
        return None
    if isinstance(value, float) and np.isnan(value):
        return None
    if hasattr(value, "item"):
        value = value.item()
        return None if isinstance(value, float) and np.isnan(value) else value
    return value


def _parse_date_column(df: pd.DataFrame, date_column: str) -> pd.Series:
    """Parse date_column into datetimes, guarding against numeric epoch misparsing.

    pd.to_datetime() treats bare numeric input as epoch-nanoseconds, so bare
    month numbers (1-12) or calendar years (e.g. 2025) silently collapse to a
    single 1970 timestamp instead of raising. Numeric columns are resolved
    explicitly here rather than falling through to pd.to_datetime().
    """
    col = df[date_column]
    if not pd.api.types.is_numeric_dtype(col):
        return pd.to_datetime(col)

    non_null = col.dropna()
    is_whole = non_null.empty or (non_null == non_null.astype(int)).all()

    if is_whole and non_null.between(1, 12).all():
        year_col = next((c for c in df.columns if c.lower() == "year"), None)
        year_series = df[year_col] if year_col is not None else None
        year_non_null = year_series.dropna() if year_series is not None else pd.Series(dtype=float)
        year_ok = (
            year_series is not None
            and pd.api.types.is_numeric_dtype(year_series)
            and not year_non_null.empty
            and (year_non_null == year_non_null.astype(int)).all()
            and year_non_null.between(1900, 2100).all()
        )
        if not year_ok:
            raise InvalidDateColumnError(
                f"date_column '{date_column}' holds bare month numbers (1-12) and no "
                "'year' column is present in the data — retrieve a year column "
                "alongside month, or use a real date column."
            )
        valid = col.notna() & year_series.notna()
        result = pd.Series(pd.NaT, index=col.index, dtype="datetime64[ns]")
        result.loc[valid] = pd.to_datetime(
            {
                "year": year_series.loc[valid].astype(int),
                "month": col.loc[valid].astype(int),
                "day": 1,
            }
        )
        return result

    if is_whole and non_null.between(1900, 2100).all():
        result = pd.Series(pd.NaT, index=col.index, dtype="datetime64[ns]")
        valid = col.notna()
        result.loc[valid] = pd.to_datetime(
            col.loc[valid].astype(int).astype(str), format="%Y"
        )
        return result

    raise InvalidDateColumnError(
        f"date_column '{date_column}' is numeric but is not a recognizable date, "
        "year, or month column."
    )


def _period_label(ts: pd.Timestamp, freq: str) -> str:
    """Human-readable period label keyed off the friendly frequency name."""
    if freq == "month":
        return str(ts.strftime("%Y-%m"))
    if freq == "quarter":
        return f"{ts.year}-Q{ts.quarter}"
    if freq == "year":
        return str(ts.strftime("%Y"))
    return str(ts.strftime("%Y-%m-%d"))  # day / week


# Prompt-style description read by the Planner to decide WHEN to pick this tool.
# Final destination is ToolSpec.description once the wrapper layer is built.
DESCRIPTION = """\
Summary: Time-series trend of one metric over evenly-spaced periods (day, week, \
month, quarter, year). Reports per-period points plus direction, absolute and \
percent change, and a linear slope.

USE WHEN the question is about movement over time — growth, decline, trend, \
seasonality. Trigger words: "over time" (dari waktu ke waktu), "trend" (tren), \
"monthly/yearly" (bulanan/tahunan), "growth" (pertumbuhan), "since/last N months".

DON'T USE WHEN:
  - it groups by a non-time category -> analyze_aggregate
  - it compares two specific groups (A vs B) -> analyze_comparison
  - it summarizes a column with no time axis -> analyze_descriptive

Example questions:
  - "how did monthly revenue change this year?"
  - "show the sales trend over the last 12 months"
  - "is the number of signups growing quarter over quarter?"
  - "yearly profit from 2019 to 2024"
"""


def analyze_trend(
    df: pd.DataFrame,
    date_column: str,
    value_column: str,
    freq: str = "month",
    agg: str = "sum",
) -> dict[str, object]:
    """Time-series trend of one value over evenly-spaced periods.

    Args:
        df: already-materialized data (in the real system the wrapper fetches
            this from a source_id).
        date_column: column holding dates/timestamps.
        value_column: numeric column to aggregate per period.
        freq: period granularity — one of FREQ_MAP keys (default "month").
        agg: how to aggregate within a period — one of SUPPORTED_AGGS.

    Returns:
        dict with:
            freq, agg          — echo of the chosen settings
            points             — [{"period": str, "value": number|None}, ...]
            first, last        — value of the first/last non-empty period
            change_abs         — last - first
            change_pct         — (last - first) / first, or None if first == 0
            direction          — "up" | "down" | "flat"
            slope              — linear slope across periods, or None if < 2 points

    Raises:
        ColumnNotFoundError: if date_column or value_column is absent.
        InvalidFrequencyError: if freq is not a known period.
        UnsupportedAggregationError: if agg is not supported.
        InvalidDateColumnError: if date_column is numeric but not a recognizable
            date, year, or bare month number (needing a companion 'year' column).
    """
    missing = [c for c in (date_column, value_column) if c not in df.columns]
    if missing:
        raise ColumnNotFoundError(f"columns not found: {missing}")
    if freq not in FREQ_MAP:
        raise InvalidFrequencyError(
            f"unknown frequency '{freq}'; supported: {list(FREQ_MAP)}"
        )
    if agg not in SUPPORTED_AGGS:
        raise UnsupportedAggregationError(
            f"unsupported aggregation '{agg}'; supported: {list(SUPPORTED_AGGS)}"
        )

    # Build a clean datetime-indexed series, then resample into periods.
    s = df[[date_column, value_column]].copy()
    s[date_column] = _parse_date_column(df, date_column)
    s = s.dropna(subset=[date_column]).set_index(date_column).sort_index()
    resampled = s[value_column].resample(FREQ_MAP[freq]).agg(agg)

    points = [
        {"period": _period_label(ts, freq), "value": _clean(val)}
        for ts, val in resampled.items()
    ]

    # Summary stats are computed over non-empty periods only.
    non_null = resampled.dropna()
    first: float | None
    last: float | None
    change_abs: float | None
    change_pct: float | None
    slope: float | None
    if non_null.empty:
        first = last = change_abs = change_pct = slope = None
        direction = "flat"
    else:
        first = float(non_null.iloc[0])
        last = float(non_null.iloc[-1])
        change_abs = last - first
        change_pct = (change_abs / first) if first != 0 else None
        if change_abs > 0:
            direction = "up"
        elif change_abs < 0:
            direction = "down"
        else:
            direction = "flat"
        if non_null.shape[0] > 1:
            x = np.arange(non_null.shape[0])
            slope = float(np.polyfit(x, non_null.to_numpy(dtype=float), 1)[0])
        else:
            slope = None

    return {
        "freq": freq,
        "agg": agg,
        "points": points,
        "first": first,
        "last": last,
        "change_abs": change_abs,
        "change_pct": change_pct,
        "direction": direction,
        "slope": slope,
    }