"""Pre-computed revenue forecasts.

Goal: get Prophet/ML forecasting *out of the request path*. The controller
should never call `_forecast_all_customers()` during a GET — it should SELECT
from `fpna_config.revenue_forecast_precomputed` instead. That collapses ~5–10 s
of ML work to a ~50 ms indexed read.

Two write paths:
  1. Nightly background task (started from main.py's lifespan) — keeps every
     active (company_id, year) fresh.
  2. Manual trigger via POST /revenue-forecast/{year}/precompute — for ops or
     after a large GL import.

Two read paths:
  - `load_precomputed_quarter()` — returns the 3 monthly AI values per customer
    for a given quarter. Drop-in replacement for `_forecast_all_customers()`.
  - `load_precomputed_n_months()` — returns N months ahead per customer. For
    forecast_type='12months' and 'rolling'.

Fallback: if a customer is missing from the precompute table, callers should
fall back to the synchronous Prophet path *just for that customer*.
"""
from __future__ import annotations

import asyncio
import logging
from datetime import datetime, timezone
from typing import Dict, List, Optional, Tuple

import pandas as pd
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

log = logging.getLogger(__name__)


# ── Read helpers (used by the request path) ──────────────────────────────────
async def load_precomputed_quarter(
    db: AsyncSession,
    company_id: str,
    forecast_year: int,
    quarter_months: Tuple[str, str, str],
) -> Dict[str, Tuple[float, float, float]]:
    """Return {customer_name: (m1_ai, m2_ai, m3_ai)} from the precompute table.

    Missing customers simply absent from the dict (caller decides fallback).
    """
    m1, m2, m3 = quarter_months
    rows = (await db.execute(
        text("""
            SELECT customer_name, period, ai_amount
            FROM fpna_config.revenue_forecast_precomputed
            WHERE company_id = :cid AND forecast_year = :fy AND period IN (:m1, :m2, :m3)
        """),
        {"cid": company_id, "fy": forecast_year, "m1": m1, "m2": m2, "m3": m3},
    )).mappings().all()

    result: Dict[str, Dict[str, float]] = {}
    for r in rows:
        result.setdefault(r["customer_name"], {})[r["period"]] = float(r["ai_amount"] or 0.0)

    return {
        cname: (
            round(periods.get(m1, 0.0), 2),
            round(periods.get(m2, 0.0), 2),
            round(periods.get(m3, 0.0), 2),
        )
        for cname, periods in result.items()
    }


async def load_precomputed_n_months(
    db: AsyncSession,
    company_id: str,
    forecast_year: int,
    periods: List[str],
) -> Dict[str, List[float]]:
    """Return {customer_name: [ai_amount_for_each_period_in_order]}."""
    if not periods:
        return {}
    rows = (await db.execute(
        text("""
            SELECT customer_name, period, ai_amount
            FROM fpna_config.revenue_forecast_precomputed
            WHERE company_id = :cid AND forecast_year = :fy AND period = ANY(:periods)
        """),
        {"cid": company_id, "fy": forecast_year, "periods": list(periods)},
    )).mappings().all()

    by_customer: Dict[str, Dict[str, float]] = {}
    for r in rows:
        by_customer.setdefault(r["customer_name"], {})[r["period"]] = float(r["ai_amount"] or 0.0)

    return {
        cname: [round(p_map.get(p, 0.0), 2) for p in periods]
        for cname, p_map in by_customer.items()
    }


# ── Write path (background job) ──────────────────────────────────────────────
_CUSTOMER_HISTORY_SQL = text("""
    SELECT c.customer_name, gl.period, SUM(gl.amount_credit - gl.amount_debit) AS revenue
    FROM fpna_landing.finance_general_ledger_transactions gl
    JOIN fpna_landing.finance_chart_of_accounts coa ON gl.gl_account_id = coa.account_id
    JOIN fpna_landing.finance_customers c
        ON REGEXP_REPLACE(gl.customer_id::TEXT, '\\.0$', '') = REGEXP_REPLACE(c.customer_id::TEXT, '\\.0$', '')
    WHERE coa.account_type = 'Revenue'
      AND gl.period BETWEEN :start_period AND :end_period
      AND gl.customer_id IS NOT NULL
    GROUP BY c.customer_name, gl.period
    ORDER BY c.customer_name, gl.period
""")


async def precompute_revenue_forecasts(
    db: AsyncSession,
    company_id: str,
    forecast_year: int,
) -> Dict[str, int]:
    """Recompute forecasts for every customer for the given (company_id, year)
    and upsert into fpna_config.revenue_forecast_precomputed.

    This is the slow operation we want to keep OFF the request path. Imports
    are local to avoid a circular import with controller.py at module load.
    """
    from .controller import (   # noqa: WPS433  (intentional local import)
        HISTORY_MONTHS_SQL,
        _forecast_all_customers_n_months,
    )

    started = datetime.now(timezone.utc)
    log.info("precompute: start company=%s year=%s", company_id, forecast_year)

    # Pull the same revenue history slice the request path uses, but for the
    # whole 12-month forecast window so we can serve quarter/12m/rolling all
    # from the same precomputed rows.
    forecast_start = pd.Timestamp(year=forecast_year, month=1, day=1)
    history_start = (forecast_start - pd.DateOffset(months=HISTORY_MONTHS_SQL)).strftime("%Y-%m")
    end_period = (forecast_start - pd.DateOffset(months=1)).strftime("%Y-%m")

    cust_hist_result = await db.execute(
        _CUSTOMER_HISTORY_SQL,
        {"start_period": history_start, "end_period": end_period},
    )
    cust_hist_df = pd.DataFrame(cust_hist_result.mappings().all())
    if cust_hist_df.empty:
        log.info("precompute: no history rows, skip")
        return {"customers": 0, "periods": 0}

    cust_hist_df = cust_hist_df.rename(columns={"revenue": "y"})
    all_customers = cust_hist_df["customer_name"].unique().tolist()

    # 12 months — gives us coverage for any forecast_type (quarter/12months/rolling).
    periods = [(forecast_start + pd.DateOffset(months=i)).strftime("%Y-%m") for i in range(12)]

    per_customer = await _forecast_all_customers_n_months(
        cust_hist_df, all_customers, forecast_start, n_ahead=12,
    )

    rows_to_upsert: List[dict] = []
    for cname in all_customers:
        amounts = per_customer.get(cname) or [0.0] * 12
        for period, amount in zip(periods, amounts):
            rows_to_upsert.append({
                "company_id":     company_id,
                "customer_name":  cname,
                "forecast_year":  forecast_year,
                "period":         period,
                "ai_amount":      float(amount or 0.0),
                "model_metadata": None,
            })

    # Upsert in batches to keep statements small.
    BATCH = 1000
    inserted = 0
    for i in range(0, len(rows_to_upsert), BATCH):
        batch = rows_to_upsert[i:i + BATCH]
        await db.execute(
            text("""
                INSERT INTO fpna_config.revenue_forecast_precomputed
                  (company_id, customer_name, forecast_year, period, ai_amount, model_metadata, computed_at)
                VALUES (:company_id, :customer_name, :forecast_year, :period, :ai_amount, CAST(:model_metadata AS JSONB), now())
                ON CONFLICT (company_id, customer_name, forecast_year, period)
                DO UPDATE SET ai_amount = EXCLUDED.ai_amount,
                              model_metadata = EXCLUDED.model_metadata,
                              computed_at = now()
            """),
            batch,
        )
        inserted += len(batch)
    await db.commit()

    elapsed = (datetime.now(timezone.utc) - started).total_seconds()
    log.info(
        "precompute: done company=%s year=%s customers=%s rows=%s elapsed=%.1fs",
        company_id, forecast_year, len(all_customers), inserted, elapsed,
    )
    return {"customers": len(all_customers), "periods": inserted, "elapsed_seconds": elapsed}


# ── Coverage check (used by request path to decide fallback) ────────────────
async def precompute_has_coverage(
    db: AsyncSession, company_id: str, forecast_year: int,
) -> bool:
    """Cheap existence check: do we have ANY precomputed rows for this slice?"""
    row = (await db.execute(
        text("""
            SELECT 1
            FROM fpna_config.revenue_forecast_precomputed
            WHERE company_id = :cid AND forecast_year = :fy
            LIMIT 1
        """),
        {"cid": company_id, "fy": forecast_year},
    )).first()
    return bool(row)


# ── Background scheduler loop ────────────────────────────────────────────────
_PRECOMPUTE_TASK: Optional[asyncio.Task] = None


async def _precompute_all_companies_once() -> None:
    """Find every (company_id, forecast_year) that has *any* GL activity and
    refresh its precomputed forecast. Cheap to run nightly."""
    from src.core.database import AsyncSessionLocal

    async with AsyncSessionLocal() as db:
        targets = (await db.execute(
            text("""
                SELECT DISTINCT t.company_id,
                       CAST(LEFT(t.period, 4) AS INTEGER) AS y
                FROM fpna_landing.finance_general_ledger_transactions t
                WHERE t.period IS NOT NULL
            """ if _has_company_id_column() else """
                SELECT DISTINCT CAST(LEFT(period, 4) AS INTEGER) AS y
                FROM fpna_landing.finance_general_ledger_transactions
                WHERE period IS NOT NULL
            """)
        )).mappings().all()

    # The current GL table doesn't carry company_id; treat company_id="default"
    # as the single-tenant key. If/when GL is multi-tenant this loop should
    # iterate company_id too.
    seen_years = sorted({int(t["y"]) for t in targets if t.get("y")})
    for y in seen_years:
        try:
            async with AsyncSessionLocal() as db:
                await precompute_revenue_forecasts(db, company_id="default", forecast_year=y)
        except Exception:    # noqa: BLE001  (must keep loop alive)
            log.exception("precompute: company=default year=%s failed", y)


def _has_company_id_column() -> bool:
    """The GL table may or may not have a `company_id` column depending on
    deployment. Return True if the column exists. Default False for safety."""
    return False  # current schema (see forecasting/models.py) has no company_id


async def precompute_scheduler(interval_seconds: int = 24 * 3600) -> None:
    """Background loop. Sleeps first, then runs forever. Cancellable."""
    # Initial small delay so app startup isn't blocked / contended.
    await asyncio.sleep(60)
    while True:
        try:
            await _precompute_all_companies_once()
        except asyncio.CancelledError:
            raise
        except Exception:    # noqa: BLE001
            log.exception("precompute scheduler iteration failed")
        await asyncio.sleep(interval_seconds)


def start_background_precompute() -> Optional[asyncio.Task]:
    """Idempotent: returns the existing task if already started."""
    global _PRECOMPUTE_TASK
    if _PRECOMPUTE_TASK and not _PRECOMPUTE_TASK.done():
        return _PRECOMPUTE_TASK
    _PRECOMPUTE_TASK = asyncio.create_task(precompute_scheduler())
    return _PRECOMPUTE_TASK


def stop_background_precompute() -> None:
    global _PRECOMPUTE_TASK
    if _PRECOMPUTE_TASK and not _PRECOMPUTE_TASK.done():
        _PRECOMPUTE_TASK.cancel()
