import math
from dataclasses import dataclass

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

from app.models.user import User
from app.models.credit_record import CreditRecord
from app.models.video_engine import VideoEngine
from app.models.image_engine import ImageEngine
from app.models.credit_ratio import CreditRatio
from app.utils.id_gen import generate_id
from app.utils.exceptions import InsufficientCreditsError
from app.enums.common import BillingBlockEventEnum
from app.services.operation_log_service import log_operation_event
from app.services.system_config_cache import get_system_config_value
from app.services.credit_record_meta_service import CreditRecordMeta, with_user_snapshot


async def calc_text_credits(db: AsyncSession, input_tokens: int, output_tokens: int) -> float:
    """Calculate text credits based on actual token usage and cached configurable rate."""
    raw_rate = await get_system_config_value(db, "text_credits_per_1000_tokens")
    try:
        rate = float(raw_rate) if raw_rate not in (None, "") else 1.0
    except (TypeError, ValueError):
        rate = 1.0
    total_tokens = input_tokens + output_tokens
    return round(total_tokens * rate / 1000, 2)


async def _get_credit_ratio(
    db: AsyncSession,
    *,
    gen_type: str,
    resolution: str,
    engine_id: str | None = None,
) -> CreditRatio | None:
    """按引擎精确规则优先获取积分规则；找不到时回退到同类型同参数最高规则。"""
    gen_type = (gen_type or "").lower().strip()
    resolution = (resolution or "").strip()
    engine_id = (engine_id or "").strip() or None

    if engine_id:
        result = await db.execute(
            select(CreditRatio)
            .where(CreditRatio.gen_type == gen_type)
            .where(CreditRatio.model_config_id == engine_id)
            .where(CreditRatio.resolution == resolution)
            .order_by(CreditRatio.base_credits.desc(), CreditRatio.per_second_credits.desc())
            .limit(1)
        )
        ratio = result.scalar_one_or_none()
        if ratio:
            return ratio

    result = await db.execute(
        select(CreditRatio)
        .where(CreditRatio.gen_type == gen_type)
        .where(CreditRatio.resolution == resolution)
        .order_by(CreditRatio.base_credits.desc(), CreditRatio.per_second_credits.desc())
        .limit(1)
    )
    return result.scalar_one_or_none()


async def calc_video_credits(
    db: AsyncSession,
    duration: int,
    resolution: str,
    engine_id: str | None = None,
    input_video_duration: float | None = None,
    input_image_count: int | None = None,
) -> float:
    """Calculate video credits using CreditRatio table, with fallback to hardcoded.

    查询优先级：
    1. gen_type=video + engine_id + resolution 精确规则；
    2. gen_type=video + resolution 下 base_credits/per_second_credits 最高规则；
    3. 原硬编码默认算法。

    input_video_duration: 用户上传的参考视频总时长（秒），不为空时额外计费
    input_image_count: 用户上传的参考图片数量，不为空时额外计费
    """
    if not engine_id:
        video_engines_result = await db.execute(
            select(VideoEngine.id)
            .where(VideoEngine.is_active == True, VideoEngine.deleted_at.is_(None))
            .order_by(VideoEngine.priority.desc())
            .limit(1)
        )
        engine_id = video_engines_result.scalar_one_or_none()
    ratio = await _get_credit_ratio(
        db,
        gen_type="video",
        resolution=resolution,
        engine_id=engine_id,
    )
    if ratio:
        base_cost = (ratio.base_credits + ratio.per_second_credits * duration) * ratio.ratio
        if input_video_duration and input_video_duration > 0:
            input_video_cost = (
                ratio.input_video_base_credits + ratio.input_video_per_second_credits * input_video_duration
            ) * ratio.input_video_ratio
            base_cost += input_video_cost
        if input_image_count and input_image_count > 0:
            input_image_cost = (
                ratio.input_image_base_credits + ratio.input_image_per_image_credits * input_image_count
            ) * ratio.input_image_ratio
            base_cost += input_image_cost
        return round(base_cost, 2)

    base = 60.0
    duration_cost = duration * 2.0
    multiplier = {"480p": 1, "1080p": 2, "720p": 1.5}.get(resolution, 1.0)
    total = (base + duration_cost) * multiplier
    if input_video_duration and input_video_duration > 0:
        total += input_video_duration * 0.5 * multiplier
    if input_image_count and input_image_count > 0:
        total += input_image_count * 0.5 * multiplier
    return round(total, 2)


def calc_credits(duration: int, resolution: str) -> float:
    """Legacy: hardcoded credit calculation. Prefer calc_video_credits for new code."""
    base = 60.0
    duration_cost = duration * 2.0
    multiplier = {"480p": 2.5, "1080p": 1.5, "720p": 1.0}.get(resolution, 1.0)
    return round((base + duration_cost) * multiplier, 2)


async def calc_image_credits(
    db: AsyncSession,
    image_size: str,
    engine_id: str | None = None,
    input_image_count: int | None = None,
) -> float:
    """Calculate image credits using CreditRatio table, with fallback to hardcoded.

    查询优先级：
    1. gen_type=image + engine_id + image_size 精确规则；
    2. gen_type=image + image_size 下 base_credits/per_second_credits 最高规则；
    3. 原硬编码默认算法。

    input_image_count: 用户上传的参考图片数量，不为空时额外计费
    """
    # 如果engine_id为空，默认查询权重最高的图片引擎积分规则
    if not engine_id:
        image_engines_result = await db.execute(
            select(ImageEngine.id)
            .where(ImageEngine.is_active == True, ImageEngine.deleted_at.is_(None))
            .order_by(ImageEngine.priority.desc())
            .limit(1)
        )
        engine_id = image_engines_result.scalar_one_or_none()
    ratio = await _get_credit_ratio(
        db,
        gen_type="image",
        resolution=image_size,
        engine_id=engine_id,
    )
    if ratio:
        base_cost = ratio.base_credits * ratio.ratio
        if input_image_count and input_image_count > 0:
            input_image_cost = (
                ratio.input_image_base_credits + ratio.input_image_per_image_credits * input_image_count
            ) * ratio.input_image_ratio
            base_cost += input_image_cost
        return round(base_cost, 2)

    # Fallback
    multiplier = {"4K": 2.0, "2K": 1.0}.get(image_size, 1.0)
    base_cost = 4.0
    total = base_cost * multiplier
    if input_image_count and input_image_count > 0:
        total += input_image_count * 0.5 * multiplier
    return round(total, 2)


@dataclass(slots=True)
class CreditMutationResult:
    user: User
    record: CreditRecord | None
    created: bool
    amount: float
    balance_before: float
    balance_after: float


async def _get_existing_credit_record_by_biz_key(
    db: AsyncSession,
    *,
    user_id: str,
    biz_key: str | None,
) -> CreditRecord | None:
    """按正式业务幂等键查找已有积分流水。"""
    if not biz_key:
        return None
    result = await db.execute(
        select(CreditRecord)
        .where(CreditRecord.user_id == user_id, CreditRecord.biz_key == biz_key)
        .limit(1)
    )
    return result.scalar_one_or_none()


async def deduct_credits_result(
    db: AsyncSession,
    user_id: str,
    amount: float,
    description: str,
    related_id: str | None = None,
    *,
    biz_key: str | None = None,
    refund_for_biz_key: str | None = None,
    record_meta: CreditRecordMeta | dict | None = None,
    record_type: str = "consume",
    allow_negative: bool = False,
    create_zero_record: bool = False,
) -> CreditMutationResult:
    """并发安全且可观察幂等结果的积分扣减。"""
    amount = round(float(amount or 0), 2)
    result = await db.execute(select(User).where(User.id == user_id).with_for_update().limit(1))
    user = result.scalar_one_or_none()
    if not user:
        raise ValueError("User not found")

    before_balance = round(float(user.credits or 0), 2)
    if biz_key:
        existing = await _get_existing_credit_record_by_biz_key(db, user_id=user_id, biz_key=biz_key)
        if existing:
            return CreditMutationResult(
                user=user,
                record=existing,
                created=False,
                amount=abs(round(float(existing.amount or 0), 2)),
                balance_before=before_balance,
                balance_after=before_balance,
            )

    if amount <= 0 and not create_zero_record:
        return CreditMutationResult(
            user=user,
            record=None,
            created=False,
            amount=0.0,
            balance_before=before_balance,
            balance_after=before_balance,
        )

    if amount > 0 and not allow_negative and before_balance < amount:
        event_type = (
            BillingBlockEventEnum.NEGATIVE_BALANCE.value
            if before_balance < 0
            else BillingBlockEventEnum.INSUFFICIENT_CREDITS.value
        )
        log_operation_event(
            domain="billing",
            module="credits",
            event_type=event_type,
            event_status="failed",
            source="app.services.credits.deduct_credits_result",
            user_id=user_id,
            task_id=related_id,
            message="积分不足，已拦截新的扣费请求",
            detail={
                "user_id": user_id,
                "amount": amount,
                "before_balance": before_balance,
                "allow_negative": allow_negative,
                "biz_key": biz_key,
                "refund_for_biz_key": refund_for_biz_key,
                "description": description,
                "record_type": record_type,
            },
        )
        raise InsufficientCreditsError()

    user.credits = round(before_balance - max(0.0, amount), 2)
    meta_kwargs: dict = {}
    if record_meta:
        if isinstance(record_meta, CreditRecordMeta):
            record_meta = await with_user_snapshot(db, record_meta, user_id, user=user)
            meta_kwargs = record_meta.to_record_kwargs()
        elif isinstance(record_meta, dict):
            meta_kwargs = {k: v for k, v in record_meta.items() if v is not None}
    record = CreditRecord(
        id=generate_id(),
        user_id=user_id,
        type=record_type,
        amount=-amount,
        balance_after=user.credits,
        description=description,
        related_id=related_id,
        biz_key=biz_key,
        refund_for_biz_key=refund_for_biz_key,
        **meta_kwargs,
    )
    db.add(record)
    await db.flush()
    return CreditMutationResult(
        user=user,
        record=record,
        created=True,
        amount=amount,
        balance_before=before_balance,
        balance_after=round(float(user.credits or 0), 2),
    )


async def deduct_credits(
    db: AsyncSession,
    user_id: str,
    amount: float,
    description: str,
    related_id: str | None = None,
    *,
    biz_key: str | None = None,
    refund_for_biz_key: str | None = None,
    record_meta: CreditRecordMeta | dict | None = None,
    record_type: str = "consume",
    allow_negative: bool = False,
    create_zero_record: bool = False,
) -> User:
    """兼容旧调用：返回 User；精确幂等状态请使用 deduct_credits_result。"""
    mutation = await deduct_credits_result(
        db,
        user_id=user_id,
        amount=amount,
        description=description,
        related_id=related_id,
        biz_key=biz_key,
        refund_for_biz_key=refund_for_biz_key,
        record_meta=record_meta,
        record_type=record_type,
        allow_negative=allow_negative,
        create_zero_record=create_zero_record,
    )
    return mutation.user


async def add_credits_result(
    db: AsyncSession,
    user_id: str,
    amount: float,
    description: str,
    related_id: str | None = None,
    *,
    record_type: str = "recharge",
    biz_key: str | None = None,
    refund_for_biz_key: str | None = None,
    record_meta: CreditRecordMeta | dict | None = None,
) -> CreditMutationResult:
    """并发安全且可观察幂等结果的积分增加。"""
    amount = round(float(amount or 0), 2)
    result = await db.execute(select(User).where(User.id == user_id).with_for_update().limit(1))
    user = result.scalar_one_or_none()
    if not user:
        raise ValueError("User not found")

    before_balance = round(float(user.credits or 0), 2)
    if biz_key:
        existing = await _get_existing_credit_record_by_biz_key(db, user_id=user_id, biz_key=biz_key)
        if existing:
            return CreditMutationResult(
                user=user,
                record=existing,
                created=False,
                amount=abs(round(float(existing.amount or 0), 2)),
                balance_before=before_balance,
                balance_after=before_balance,
            )

    if amount <= 0:
        return CreditMutationResult(
            user=user,
            record=None,
            created=False,
            amount=0.0,
            balance_before=before_balance,
            balance_after=before_balance,
        )

    user.credits = round(before_balance + amount, 2)
    meta_kwargs: dict = {}
    if record_meta:
        if isinstance(record_meta, CreditRecordMeta):
            record_meta = await with_user_snapshot(db, record_meta, user_id, user=user)
            meta_kwargs = record_meta.to_record_kwargs()
        elif isinstance(record_meta, dict):
            meta_kwargs = {k: v for k, v in record_meta.items() if v is not None}
    record = CreditRecord(
        id=generate_id(),
        user_id=user_id,
        type=record_type,
        amount=amount,
        balance_after=user.credits,
        description=description,
        related_id=related_id,
        biz_key=biz_key,
        refund_for_biz_key=refund_for_biz_key,
        **meta_kwargs,
    )
    db.add(record)
    await db.flush()
    return CreditMutationResult(
        user=user,
        record=record,
        created=True,
        amount=amount,
        balance_before=before_balance,
        balance_after=round(float(user.credits or 0), 2),
    )


async def add_credits(
    db: AsyncSession,
    user_id: str,
    amount: float,
    description: str,
    related_id: str | None = None,
    *,
    record_type: str = "recharge",
    biz_key: str | None = None,
    refund_for_biz_key: str | None = None,
    record_meta: CreditRecordMeta | dict | None = None,
) -> User:
    """兼容旧调用：返回 User；精确幂等状态请使用 add_credits_result。"""
    mutation = await add_credits_result(
        db,
        user_id=user_id,
        amount=amount,
        description=description,
        related_id=related_id,
        record_type=record_type,
        biz_key=biz_key,
        refund_for_biz_key=refund_for_biz_key,
        record_meta=record_meta,
    )
    return mutation.user


async def refund_credits(
    db: AsyncSession,
    user_id: str,
    amount: float,
    description: str,
    related_id: str | None = None,
    *,
    biz_key: str | None = None,
    refund_for_biz_key: str | None = None,
    record_meta: CreditRecordMeta | dict | None = None,
) -> User:
    """生成失败积分回退。"""
    return await add_credits(
        db,
        user_id=user_id,
        amount=amount,
        description=description,
        related_id=related_id,
        record_type="refund",
        biz_key=biz_key,
        refund_for_biz_key=refund_for_biz_key,
        record_meta=record_meta,
    )


async def get_records(db: AsyncSession, user_id: str, page: int = 1, page_size: int = 20) -> tuple[list[CreditRecord], int]:
    count_query = select(func.count(CreditRecord.id)).where(CreditRecord.user_id == user_id)
    total = (await db.execute(count_query)).scalar() or 0
    
    result = await db.execute(
        select(CreditRecord)
        .where(CreditRecord.user_id == user_id)
        .order_by(CreditRecord.created_at.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
    )
    return list(result.scalars().all()), total