"""One revenue forecast collection and calculation cycle."""

from __future__ import annotations

from datetime import datetime, timedelta
from decimal import Decimal
from typing import Any
from zoneinfo import ZoneInfo

from odps_source import build_odps_client
from revenue_forecast import (
    RevenueForecast,
    RevenueObservation,
)
from revenue_forecast_config import RevenueForecastConfig
from revenue_forecast_repository import (
    load_speed_parameter,
    load_previous_observation,
    save_forecast,
    upsert_observation,
)
from revenue_speed_forecast import (
    SpeedForecastUnavailable,
    calculate_speed_forecast,
)
from revenue_forecast_source import (
    MINIAPP_CHANNEL,
    fetch_daily_channel_costs,
    fetch_revenue_source_snapshot,
)
from storage import advisory_lock


SHANGHAI = ZoneInfo("Asia/Shanghai")


def _in_runtime_window(now: datetime, config: RevenueForecastConfig) -> bool:
    local_time = now.timetz().replace(tzinfo=None)
    return config.start_time <= local_time <= config.stop_time


def _quality_status(
    observation: RevenueObservation,
    previous: RevenueObservation | None,
    now: datetime,
    config: RevenueForecastConfig,
) -> tuple[str, str | None, int]:
    lag_seconds = int((now - observation.report_time).total_seconds())
    if observation.report_time.date() != now.date():
        return "INVALID", "report_time is not from the current day", lag_seconds
    if lag_seconds < -60:
        return "INVALID", "report_time is in the future", lag_seconds
    if lag_seconds > config.max_lag_minutes * 60:
        return "STALE", "source data exceeds maximum lag", lag_seconds
    if observation.today_revenue < 0:
        return "INVALID", "today cumulative revenue is negative", lag_seconds
    if previous and observation.today_revenue < previous.today_revenue:
        return "CORRECTED", "cumulative revenue moved backwards", lag_seconds
    return "VALID", None, lag_seconds


def _decimal_json(value: Decimal | None) -> str | None:
    return str(value) if value is not None else None


def _forecast_payload(forecast: RevenueForecast) -> dict[str, Any]:
    return {
        "forecast_status": forecast.status,
        "forecast_version": forecast.forecast_version,
        "forecast_method": forecast.forecast_method,
        "parameter_version": forecast.parameter_version,
        "current_cumulative_revenue": str(
            forecast.current_cumulative_revenue
        ),
        "forecast_revenue": str(forecast.forecast_revenue),
        "forecast_lower": str(forecast.forecast_lower),
        "forecast_upper": str(forecast.forecast_upper),
        "target_cost_ratio": str(forecast.target_cost_ratio),
        "target_daily_cost": str(forecast.target_daily_cost),
        "cost_reserve_date": (
            forecast.cost_reserve_date.isoformat()
            if forecast.cost_reserve_date is not None
            else None
        ),
        "channel_costs": {
            channel: str(cost) for channel, cost in forecast.channel_costs
        },
        "non_miniapp_reserved_cost": str(
            forecast.non_miniapp_reserved_cost
        ),
        "miniapp_target_daily_cost": str(
            forecast.miniapp_target_daily_cost
        ),
        "speed_latest_weight": _decimal_json(forecast.speed_latest_weight),
        "latest_interval_revenue": _decimal_json(
            forecast.latest_interval_revenue
        ),
        "previous_interval_revenue": _decimal_json(
            forecast.previous_interval_revenue
        ),
        "weighted_speed": _decimal_json(forecast.weighted_speed),
        "remaining_multiplier_p10": _decimal_json(
            forecast.remaining_multiplier_p10
        ),
        "remaining_multiplier_p50": _decimal_json(
            forecast.remaining_multiplier_p50
        ),
        "remaining_multiplier_p90": _decimal_json(
            forecast.remaining_multiplier_p90
        ),
        "parameter_sample_count": forecast.parameter_sample_count,
    }


def run_revenue_forecast_cycle(
    *,
    now: datetime | None = None,
    config: RevenueForecastConfig | None = None,
    ignore_runtime_window: bool = False,
) -> dict[str, Any]:
    now = (now or datetime.now(SHANGHAI)).astimezone(SHANGHAI)
    config = config or RevenueForecastConfig.from_env()
    if not ignore_runtime_window and not _in_runtime_window(now, config):
        return {"status": "OFF_HOURS", "evaluated_at": now.isoformat()}

    with advisory_lock(config.lock_name) as acquired:
        if not acquired:
            return {"status": "LOCKED", "evaluated_at": now.isoformat()}

        client = build_odps_client()
        source_snapshot = fetch_revenue_source_snapshot(
            client,
            now.date(),
            window_minutes=config.trend_window_minutes,
            window_count=config.trend_window_count,
            as_of=now,
        )
        if source_snapshot is None:
            return {"status": "WAIT_DATA", "evaluated_at": now.isoformat()}
        observation = source_snapshot.observation

        previous = load_previous_observation(
            observation.report_time,
            config.forecast_version,
        )
        quality_status, quality_message, lag_seconds = _quality_status(
            observation,
            previous,
            now,
            config,
        )
        upsert_observation(
            observation,
            source_version=config.forecast_version,
            lag_seconds=lag_seconds,
            quality_status=quality_status,
            quality_message=quality_message,
        )
        payload: dict[str, Any] = {
            "status": quality_status,
            "quality_status": quality_status,
            "partition": observation.partition,
            "report_time": observation.report_time.isoformat(),
            "lag_seconds": lag_seconds,
            "today_revenue": str(observation.today_revenue),
            "quality_message": quality_message,
            "evaluated_at": now.isoformat(),
        }
        if quality_status != "VALID":
            return payload

        trend_windows = list(source_snapshot.trend_windows)
        cost_reserve_date = now.date() - timedelta(days=1)
        channel_costs = fetch_daily_channel_costs(client, cost_reserve_date)
        if not channel_costs:
            payload["status"] = "WAIT_COST_DATA"
            payload["cost_reserve_date"] = cost_reserve_date.isoformat()
            return payload
        non_miniapp_reserved_cost = sum(
            (
                cost
                for channel, cost in channel_costs
                if channel != MINIAPP_CHANNEL
            ),
            Decimal("0"),
        )
        parameter = load_speed_parameter(
            observation.report_time,
            config.speed_parameter_version,
        )
        try:
            forecast = calculate_speed_forecast(
                observation,
                trend_windows,
                parameter,
                forecast_version=config.forecast_version,
                parameter_version=config.speed_parameter_version,
                target_cost_ratio=config.target_cost_ratio,
                minimum_samples=config.speed_minimum_samples,
                cost_reserve_date=cost_reserve_date,
                channel_costs=channel_costs,
                non_miniapp_reserved_cost=non_miniapp_reserved_cost,
            )
        except SpeedForecastUnavailable as exc:
            payload["status"] = "WAIT_SPEED_MODEL"
            payload["forecast_status"] = str(exc)
            payload["parameter_version"] = config.speed_parameter_version
            return payload
        save_forecast(forecast)
        payload["status"] = "FORECASTED"
        payload.update(_forecast_payload(forecast))
        return payload
