# backend/app/plugins/advanced_stats/query.py
"""聚合查询引擎。

架构约束（重要）：
主查询只聚合 orders 表。order_items / refund_requests / member_points_ledger /
wallet_transactions 全部是一对多，JOIN 进主查询会造成行扇出——SUM(grand_total)
被行数乘倍，营收凭空翻倍。这些副表各发一条同维度聚合查询，Python 按 dim_key 合并。
查询条数固定 1 + 4，与数据量无关。
"""
from __future__ import annotations

from datetime import datetime
from decimal import Decimal
from typing import Any

from sqlalchemy import Select, and_, func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.models.customer import Customer
from app.core.models.discount_usage_log import DiscountUsageLog
from app.core.models.member import MemberPointsLedger
from app.core.models.order import Order, OrderItem
from app.core.models.payment import Payment
from app.core.models.product import Product
from app.core.models.refund import RefundRequest
from app.core.models.wallet import WalletTransaction
from app.plugins.advanced_stats.dimensions import DimCtx, get_dimension
from app.plugins.advanced_stats.schemas import ReportQuery
from app.plugins.advanced_stats.timerange import (
    local_range_to_utc_bounds,
    resolve_range,
    today_in_tz,
)

PAID_STATUSES = ["paid", "shipped", "completed"]


def date_column(basis: str):
    return Order.paid_at if basis == "paid" else Order.created_at


def resolve_effective_range(q: ReportQuery) -> tuple[Any, Any, Any]:
    """→ (本地开始日期, 本地结束日期, UTC 半开区间上下界)"""
    today = today_in_tz(now_utc=datetime.utcnow(), tz_offset_minutes=q.tz_offset)
    d_start, d_end = resolve_range(
        q.range, today=today, date_start=q.date_start, date_end=q.date_end
    )
    lo, hi = local_range_to_utc_bounds(d_start, d_end, q.tz_offset)
    return d_start, d_end, (lo, hi)


def effective_statuses(q: ReportQuery) -> list[str]:
    """状态口径：显式指定 > 维度默认 > PAID_STATUSES。

    默认与 Dashboard 保持一致（PAID_STATUSES），否则同一天同一家店，
    Dashboard 和本报表会显示不同的营收数字，看起来像 bug。
    """
    if q.statuses:
        return q.statuses
    dim = get_dimension(q.dimension)
    if dim.default_statuses:
        return list(dim.default_statuses)
    return list(PAID_STATUSES)


def base_conditions(q: ReportQuery, tenant_id: int) -> list:
    """所有查询共用的 WHERE：租户 + 状态 + UTC 时间区间。

    时间用 UTC 时间戳半开区间而非 DATE() 函数，才能走
    ix_orders_tenant_status_date 索引。
    """
    _, _, (lo, hi) = resolve_effective_range(q)
    col = date_column(q.date_basis)
    conds = [
        Order.tenant_id == tenant_id,
        Order.status.in_(effective_statuses(q)),
        col < hi,
    ]
    if lo is not None:
        conds.append(col >= lo)
    if q.date_basis == "paid":
        conds.append(Order.paid_at.isnot(None))
    return conds


def apply_filters(stmt: Select, q: ReportQuery, *, joined_customer: bool = False) -> Select:
    """把筛选条件挂上去。所有值都走 SQLAlchemy 参数绑定，不做字符串拼接。"""
    if q.order_no_from:
        stmt = stmt.where(Order.order_no >= q.order_no_from)
    if q.order_no_to:
        stmt = stmt.where(Order.order_no <= q.order_no_to)
    if q.total_min is not None:
        stmt = stmt.where(Order.grand_total >= q.total_min)
    if q.total_max is not None:
        stmt = stmt.where(Order.grand_total <= q.total_max)
    if q.currencies:
        stmt = stmt.where(func.coalesce(Order.display_currency, Order.currency).in_(q.currencies))
    if q.shipping_method_ids:
        stmt = stmt.where(Order.shipping_method_id.in_(q.shipping_method_ids))

    # 地址筛选：与维度用同一套 JSON key
    for field, value in (
        ("country", q.country), ("province", q.province), ("city", q.city),
        ("district", q.district), ("zip_code", q.postcode),
    ):
        if value:
            stmt = stmt.where(Order.shipping_address[field].as_string().like(f"%{value}%"))

    if joined_customer:
        if q.customer_name:
            stmt = stmt.where(Customer.name.like(f"%{q.customer_name}%"))
        if q.customer_email:
            stmt = stmt.where(Customer.email.like(f"%{q.customer_email}%"))
        if q.customer_phone:
            stmt = stmt.where(Customer.phone.like(f"%{q.customer_phone}%"))
        if q.member_level_ids:
            stmt = stmt.where(Customer.member_level_id.in_(q.member_level_ids))
    return stmt


def needs_customer_join(q: ReportQuery) -> bool:
    return bool(
        q.customer_name or q.customer_email or q.customer_phone
        or q.member_level_ids or q.dimension == "member_level"
    )


def build_order_aggregate(q: ReportQuery, tenant_id: int) -> Select:
    """订单级主聚合。只碰 orders（+ 按需 customers / payments），不碰任何一对多副表。"""
    dim = get_dimension(q.dimension)
    ctx = DimCtx(tz=q.tz_offset, date_col=date_column(q.date_basis), granularity=q.granularity)
    dim_expr = dim.expr(ctx)

    stmt = select(
        dim_expr.label("dim_key"),
        func.count(func.distinct(Order.id)).label("orders"),
        func.count(func.distinct(Order.customer_id)).label("customers"),
        func.coalesce(func.sum(Order.subtotal), 0).label("subtotal"),
        func.coalesce(func.sum(Order.discount_total), 0).label("discount"),
        func.coalesce(func.sum(Order.shipping_total), 0).label("shipping"),
        func.coalesce(func.sum(Order.tax_total), 0).label("tax"),
        func.coalesce(func.sum(Order.grand_total), 0).label("total"),
    ).where(and_(*base_conditions(q, tenant_id)))

    join_customer = needs_customer_join(q)
    if join_customer:
        stmt = stmt.join(Customer, Customer.id == Order.customer_id)
    if "payment" in dim.joins or q.payment_gateways:
        # 只取已完成的支付，避免一单多次失败尝试造成重复计数
        stmt = stmt.join(
            Payment,
            and_(Payment.order_id == Order.id, Payment.status == "completed"),
        )
        if q.payment_gateways:
            stmt = stmt.where(Payment.gateway.in_(q.payment_gateways))

    stmt = apply_filters(stmt, q, joined_customer=join_customer)
    return stmt.group_by(dim_expr)


def build_item_aggregate(q: ReportQuery, tenant_id: int) -> Select:
    """行级聚合。base='order_item' 的维度（商品/分类/品牌/税率）用它作主查询；
    base='order' 的维度用它取 items_qty 副指标。"""
    dim = get_dimension(q.dimension)
    ctx = DimCtx(tz=q.tz_offset, date_col=date_column(q.date_basis), granularity=q.granularity)
    dim_expr = dim.expr(ctx)

    stmt = (
        select(
            dim_expr.label("dim_key"),
            func.coalesce(func.sum(OrderItem.quantity), 0).label("quantity"),
            func.count(func.distinct(OrderItem.order_id)).label("orders"),
            func.coalesce(func.sum(OrderItem.total_price), 0).label("revenue"),
            func.coalesce(func.sum(OrderItem.tax_amount), 0).label("tax"),
            func.coalesce(func.avg(OrderItem.unit_price), 0).label("avg_unit_price"),
        )
        .join(Order, Order.id == OrderItem.order_id)
        .where(and_(*base_conditions(q, tenant_id)), OrderItem.tenant_id == tenant_id)
    )

    if "product" in dim.joins or q.product_keyword or q.category_ids or q.brand_ids:
        stmt = stmt.join(Product, Product.id == OrderItem.product_id)
        if q.product_keyword:
            kw = f"%{q.product_keyword}%"
            stmt = stmt.where((Product.name.like(kw)) | (Product.sku.like(kw)))
        if q.category_ids:
            stmt = stmt.where(Product.category_id.in_(q.category_ids))
        if q.brand_ids:
            stmt = stmt.where(Product.brand_id.in_(q.brand_ids))
    if q.product_ids:
        stmt = stmt.where(OrderItem.product_id.in_(q.product_ids))

    join_customer = needs_customer_join(q)
    if join_customer:
        stmt = stmt.join(Customer, Customer.id == Order.customer_id)
    stmt = apply_filters(stmt, q, joined_customer=join_customer)
    return stmt.group_by(dim_expr)


# ── 副表聚合 ─────────────────────────────────────────────────
# 每个副表一条同维度聚合查询，内部 JOIN orders 取维度表达式。
# 绝不把它们并进主查询——一对多 JOIN 会让主查询的金额按行数翻倍。
#
# 生效范围：下面三条只对 base == "order" 且不是 coupon / wallet 的维度生效。
#   - base == "order_item"：三个副表都挂在订单上，按商品行分组时一笔订单的
#     退款/积分/余额分不到具体某个商品行，硬摊只会造出假数字，所以这些维度
#     输出 ITEM_METRICS，本来就不含这三项
#   - coupon：有自己的主查询 build_coupon_aggregate
#   - wallet：有自己的主查询 build_wallet_dimension_aggregate（查钱包流水，
#     含不挂订单的充值），指标是 WALLET_METRICS
#
# 筛选口径：三个 builder 都挂 apply_filters(stmt, q, joined_customer=...)，
# 与主查询一致——用户在前端勾选「客户姓名」「订单金额」「支付方式」时，
# 退款/积分/钱包三者按与主查询同一订单集聚合，对账对得上。


def _dim_expr_for_side(q: ReportQuery):
    dim = get_dimension(q.dimension)
    ctx = DimCtx(tz=q.tz_offset, date_col=date_column(q.date_basis), granularity=q.granularity)
    return dim.expr(ctx)


def _apply_side_joins(stmt, q: ReportQuery, dim_expr):
    """副表聚合要按维度补齐主聚合一致的 JOIN：
    member_level 维度表达式来自 Customer，payment_method 来自 Payment。
    不补齐会报 Unknown column。"""
    dim = get_dimension(q.dimension)
    if dim.base == "order" and (needs_customer_join(q) or q.dimension == "member_level"):
        stmt = stmt.join(Customer, Customer.id == Order.customer_id)
    if "payment" in dim.joins or q.payment_gateways:
        stmt = stmt.join(
            Payment,
            and_(Payment.order_id == Order.id, Payment.status == "completed"),
        )
        if q.payment_gateways:
            # 顺带把 gateway 筛选挂上——主查询的 payment_gateways 处理在
            # build_order_aggregate，这里需要复制一份，否则副表覆盖范围会
            # 比主查询宽。
            stmt = stmt.where(Payment.gateway.in_(q.payment_gateways))
    return stmt


def build_refund_aggregate(q: ReportQuery, tenant_id: int) -> Select:
    dim_expr = _dim_expr_for_side(q)
    stmt = (
        select(
            dim_expr.label("dim_key"),
            func.coalesce(func.sum(RefundRequest.refund_amount), 0).label("refunds"),
        )
        .join(Order, Order.id == RefundRequest.order_id)
    )
    stmt = _apply_side_joins(stmt, q, dim_expr)
    stmt = apply_filters(stmt, q, joined_customer=needs_customer_join(q))
    return stmt.where(
        and_(*base_conditions(q, tenant_id)),
        RefundRequest.tenant_id == tenant_id,
        RefundRequest.status == "completed",
    ).group_by(dim_expr)


def build_points_aggregate(q: ReportQuery, tenant_id: int) -> Select:
    """积分赚取/使用。change_amount 正=赚 负=用，用 CASE 一次查出两列。
    只统计挂在订单上的积分变动，管理员手工调分（order_id IS NULL）不计入。"""
    dim_expr = _dim_expr_for_side(q)
    earned = func.sum(
        func.if_(MemberPointsLedger.change_amount > 0, MemberPointsLedger.change_amount, 0)
    )
    used = func.sum(
        func.if_(MemberPointsLedger.change_amount < 0, -MemberPointsLedger.change_amount, 0)
    )
    stmt = (
        select(
            dim_expr.label("dim_key"),
            func.coalesce(earned, 0).label("points_earned"),
            func.coalesce(used, 0).label("points_used"),
        )
        .join(Order, Order.id == MemberPointsLedger.order_id)
    )
    stmt = _apply_side_joins(stmt, q, dim_expr)
    stmt = apply_filters(stmt, q, joined_customer=needs_customer_join(q))
    return stmt.where(
        and_(*base_conditions(q, tenant_id)),
        MemberPointsLedger.tenant_id == tenant_id,
        MemberPointsLedger.order_id.isnot(None),
    ).group_by(dim_expr)


def build_wallet_aggregate(q: ReportQuery, tenant_id: int) -> Select:
    """余额支付金额。amount 为负（扣款），取绝对值。"""
    dim_expr = _dim_expr_for_side(q)
    stmt = (
        select(
            dim_expr.label("dim_key"),
            func.coalesce(func.sum(func.abs(WalletTransaction.amount)), 0).label("wallet_paid"),
        )
        .join(
            Order,
            and_(Order.id == WalletTransaction.source_id,
                 WalletTransaction.source_type == "order"),
        )
    )
    stmt = _apply_side_joins(stmt, q, dim_expr)
    stmt = apply_filters(stmt, q, joined_customer=needs_customer_join(q))
    return stmt.where(
        and_(*base_conditions(q, tenant_id)),
        WalletTransaction.tenant_id == tenant_id,
        WalletTransaction.type == "order_pay",
    ).group_by(dim_expr)


def build_coupon_aggregate(q: ReportQuery, tenant_id: int) -> Select:
    """券使用聚合。coupon 维度的主查询就是它——按 discount_id 分组。

    口径缺口（前端必须标注）：
    - 管理员建单不写 discount_usage_logs（既有缺陷，另有任务修）
    - POS 单促销记在 extra_attributes.pos_promotion_applied，不进本表
    """
    stmt = (
        select(
            DiscountUsageLog.discount_id.label("dim_key"),
            func.count(func.distinct(DiscountUsageLog.order_id)).label("orders"),
            func.count(func.distinct(DiscountUsageLog.customer_id)).label("customers"),
            func.coalesce(func.sum(Order.grand_total), 0).label("total"),
            func.coalesce(func.sum(DiscountUsageLog.discount_amount), 0).label("coupon_amount"),
            func.min(DiscountUsageLog.created_at).label("first_used"),
            func.max(DiscountUsageLog.created_at).label("last_used"),
        )
        .join(Order, Order.id == DiscountUsageLog.order_id)
        .where(
            and_(*base_conditions(q, tenant_id)),
            DiscountUsageLog.tenant_id == tenant_id,
        )
        .group_by(DiscountUsageLog.discount_id)
    )
    if q.discount_ids:
        stmt = stmt.where(DiscountUsageLog.discount_id.in_(q.discount_ids))
    return stmt


def build_wallet_dimension_aggregate(q: ReportQuery, tenant_id: int) -> Select:
    """余额**维度**的主查询：按流水类型分组（充值/消费/退款/后台调整）。

    注意这里**不能复用 base_conditions**——充值（type='topup'）根本不挂在订单上，
    套用订单的时间与状态过滤会把充值全部滤掉，"充值总额"永远是 0。
    所以时间过滤走 wallet_transactions.created_at，且不过滤订单状态。

    与 build_wallet_aggregate 的分工：
    - build_wallet_aggregate：副表，给别的维度提供 wallet_paid 一列（只算 order_pay）
    - 本函数：余额维度自己的主查询，四种流水类型全都要
    """
    _, _, (lo, hi) = resolve_effective_range(q)
    conds = [
        WalletTransaction.tenant_id == tenant_id,
        WalletTransaction.created_at < hi,
    ]
    if lo is not None:
        conds.append(WalletTransaction.created_at >= lo)
    return (
        select(
            WalletTransaction.type.label("dim_key"),
            func.count().label("wallet_txns"),
            func.count(func.distinct(WalletTransaction.customer_id)).label("wallet_customers"),
            func.coalesce(func.sum(func.abs(WalletTransaction.amount)), 0).label("wallet_amount"),
        )
        .where(and_(*conds))
        .group_by(WalletTransaction.type)
    )


def compute_burn_rate(rows: dict[str, dict[str, Any]]) -> float | None:
    """余额消耗率 = 消费金额 / 充值金额。没有充值时返回 None（不是 0，也不除零）。"""
    topup = Decimal(str(rows.get("topup", {}).get("wallet_amount") or 0))
    spent = Decimal(str(rows.get("order_pay", {}).get("wallet_amount") or 0))
    if not topup:
        return None
    return float((spent / topup).quantize(Decimal("0.0001")))


# ── 合并 ─────────────────────────────────────────────────────


def merge_side_metrics(
    main: dict[str, dict[str, Any]],
    side: dict[str, dict[str, Any]],
    *,
    zero_keys: tuple[str, ...],
) -> dict[str, dict[str, Any]]:
    """把副表结果按 dim_key 并进主结果。

    主表没有的分组一律丢弃——副表有而主表没有的键（例如订单状态被排除但
    退款记录还在），凭空造行会让合计对不上主查询。
    """
    for row in main.values():
        for k in zero_keys:
            row.setdefault(k, 0)
    for metric_name, by_key in side.items():
        for dim_key, value in by_key.items():
            if dim_key in main:
                main[dim_key][metric_name] = value
    return main


def compute_aov(total: Decimal, orders: int) -> Decimal:
    """客单价。订单数为 0 时返回 0，不抛 ZeroDivisionError。"""
    if not orders:
        return Decimal("0")
    return (Decimal(total) / Decimal(orders)).quantize(Decimal("0.01"))