from __future__ import annotations

from dataclasses import asdict, dataclass
from typing import Any, Mapping

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

from app.enums.credit_record import (
    CreditRecordAction,
    CreditRecordBillingScene,
    CreditRecordChargeKind,
    CreditRecordOwnerType,
    CreditRecordSourceModule,
    CreditRecordSourceStepCode,
    CreditRecordSubject,
)
from app.enums.user import FrontendUserKind
from app.models.image_engine import ImageEngine
from app.models.model_config import ModelConfig
from app.models.module_generation_step import ModuleGenerationStep
from app.models.user import User
from app.models.video_engine import VideoEngine


@dataclass(slots=True)
class CreditRecordMeta:
    owner_type: str | None = None
    owner_id: str | None = None
    attempt_no: int | None = None

    charge_kind: str | None = None
    charge_action: str | None = None
    credit_subject: str | None = None
    media_type: str | None = None
    billing_scene: str | None = None

    source_module: str | None = None
    source_project_id: str | None = None
    source_step_id: str | None = None
    source_step_code: str | None = None

    token_usage_id: str | None = None
    input_tokens: int | None = None
    output_tokens: int | None = None
    total_tokens: int | None = None

    # 统一执行配置冷备字段：
    # - engine_type=model 时，engine_* 来源于 model_configs
    # - engine_type=image 时，engine_* 来源于 image_engines
    # - engine_type=video 时，engine_* 来源于 video_engines
    engine_type: str | None = None
    engine_id: str | None = None
    engine_name: str | None = None
    engine_provider: str | None = None
    engine_model_name: str | None = None

    user_type_snapshot: str | None = None
    frontend_user_kind_snapshot: str | None = None

    def to_record_kwargs(self) -> dict[str, Any]:
        return {k: v for k, v in asdict(self).items() if v is not None}


def _safe_int(value: Any, default: int = 0) -> int:
    try:
        if value is None or value == "":
            return default
        return int(value)
    except Exception:
        return default


def _normalize_frontend_kind(value: str | None) -> str:
    return value or FrontendUserKind.EXTERNAL.value


async def with_user_snapshot(db: AsyncSession, meta: CreditRecordMeta, user_id: str) -> CreditRecordMeta:
    result = await db.execute(select(User).where(User.id == user_id).limit(1))
    user = result.scalar_one_or_none()
    if user:
        meta.user_type_snapshot = user.user_type
        meta.frontend_user_kind_snapshot = _normalize_frontend_kind(getattr(user, "frontend_user_kind", None))
    return meta


async def get_model_snapshot(db: AsyncSession, model_config_id: str | None) -> dict[str, str | None]:
    """把 ModelConfig 冷备为统一 engine_* 快照。"""
    if not model_config_id:
        return {"engine_type": "model"}
    result = await db.execute(select(ModelConfig).where(ModelConfig.id == model_config_id).limit(1))
    config = result.scalar_one_or_none()
    if not config:
        return {"engine_type": "model", "engine_id": model_config_id}
    return {
        "engine_type": "model",
        "engine_id": config.id,
        "engine_name": config.name,
        "engine_provider": config.provider,
        "engine_model_name": config.model_name,
    }


def _model_snapshot_from_usage(usage: Mapping[str, Any]) -> dict[str, str | None]:
    model_config_id = usage.get("model_config_id")
    return {
        "engine_type": "model",
        "engine_id": model_config_id,
        "engine_name": usage.get("model_config_name") or usage.get("engine_name"),
        "engine_provider": usage.get("model_provider") or usage.get("provider") or usage.get("engine_provider"),
        "engine_model_name": usage.get("model_name") or usage.get("model") or usage.get("engine_model_name"),
    }


async def _apply_model_snapshot(db: AsyncSession, meta: CreditRecordMeta, usage: Mapping[str, Any]) -> CreditRecordMeta:
    for key, value in _model_snapshot_from_usage(usage).items():
        setattr(meta, key, value)
    if meta.engine_id and not meta.engine_model_name:
        for key, value in (await get_model_snapshot(db, meta.engine_id)).items():
            setattr(meta, key, value)
    return meta


async def get_engine_snapshot(db: AsyncSession, *, gen_type: str, engine_id: str | None) -> dict[str, str | None]:
    if not engine_id:
        return {"engine_type": (gen_type or None)}
    gen_type = (gen_type or "").lower().strip()
    if gen_type == "image":
        result = await db.execute(select(ImageEngine).where(ImageEngine.id == engine_id).limit(1))
    else:
        result = await db.execute(select(VideoEngine).where(VideoEngine.id == engine_id).limit(1))
    engine = result.scalar_one_or_none()
    if not engine:
        return {"engine_type": gen_type or None, "engine_id": engine_id}
    return {
        "engine_type": gen_type or None,
        "engine_id": engine.id,
        "engine_name": getattr(engine, "name", None),
        "engine_provider": getattr(engine, "provider", None),
        "engine_model_name": getattr(engine, "model_name", None),
    }


def infer_module_billing_scene(*, source_module: str | None, source_step_code: str | None, media_type: str | None = None) -> str | None:
    if source_module == CreditRecordSourceModule.HOT_OPENING_REPLICATE.value:
        mapping = {
            "image_prompt_optimize": CreditRecordBillingScene.HOT_OPENING_IMAGE_PROMPT_OPTIMIZE.value,
            "image_generate": CreditRecordBillingScene.HOT_OPENING_IMAGE_GENERATE.value,
            "video_prompt_optimize": CreditRecordBillingScene.HOT_OPENING_VIDEO_PROMPT_OPTIMIZE.value,
            "video_generate": CreditRecordBillingScene.HOT_OPENING_VIDEO_GENERATE.value,
        }
        return mapping.get(source_step_code)
    if source_module == CreditRecordSourceModule.SHOT_REPLICATE.value:
        mapping = {
            "image_prompt_optimize": CreditRecordBillingScene.SHOT_IMAGE_PROMPT_OPTIMIZE.value,
            "image_generate": CreditRecordBillingScene.SHOT_IMAGE_GENERATE.value,
            "video_prompt_optimize": CreditRecordBillingScene.SHOT_VIDEO_PROMPT_OPTIMIZE.value,
            "video_generate": CreditRecordBillingScene.SHOT_VIDEO_GENERATE.value,
            CreditRecordSourceStepCode.VIDEO_ANALYSIS.value: CreditRecordBillingScene.SHOT_VIDEO_ANALYSIS.value,
        }
        return mapping.get(source_step_code)
    if source_module == CreditRecordSourceModule.AI_CREATION.value:
        return CreditRecordBillingScene.AI_CREATION_IMAGE_GENERATE.value if media_type == "image" else CreditRecordBillingScene.AI_CREATION_VIDEO_GENERATE.value
    if source_module == CreditRecordSourceModule.GENERATION_RECORD.value:
        return CreditRecordBillingScene.GENERATION_RECORD_IMAGE_GENERATE.value if media_type == "image" else CreditRecordBillingScene.GENERATION_RECORD_VIDEO_GENERATE.value
    return None


def build_admin_adjust_meta(*, action: str = CreditRecordAction.CHARGE.value) -> CreditRecordMeta:
    return CreditRecordMeta(
        owner_type=CreditRecordOwnerType.ADMIN_ADJUST.value,
        charge_kind=CreditRecordChargeKind.ADMIN_ADJUST.value,
        charge_action=action,
        credit_subject=CreditRecordSubject.ADMIN_ADJUST.value,
        source_module=CreditRecordSourceModule.ADMIN.value,
        billing_scene=CreditRecordBillingScene.ADMIN_ADJUST.value,
    )


def build_recharge_meta(*, owner_id: str | None = None) -> CreditRecordMeta:
    return CreditRecordMeta(
        owner_type=CreditRecordOwnerType.PAYMENT_ORDER.value if owner_id else None,
        owner_id=owner_id,
        charge_kind=CreditRecordChargeKind.RECHARGE.value,
        charge_action=CreditRecordAction.CHARGE.value,
        credit_subject=CreditRecordSubject.RECHARGE.value,
        source_module=CreditRecordSourceModule.PAYMENT.value,
        billing_scene=CreditRecordBillingScene.RECHARGE.value,
    )


def build_payment_refund_meta(*, owner_id: str | None = None) -> CreditRecordMeta:
    return CreditRecordMeta(
        owner_type=CreditRecordOwnerType.PAYMENT_ORDER.value if owner_id else None,
        owner_id=owner_id,
        charge_kind=CreditRecordChargeKind.REFUND.value,
        charge_action=CreditRecordAction.REFUND.value,
        credit_subject=CreditRecordSubject.REFUND.value,
        source_module=CreditRecordSourceModule.PAYMENT.value,
        billing_scene=CreditRecordBillingScene.REFUND.value,
    )


async def build_generation_media_meta(
    db: AsyncSession,
    *,
    owner_type: str,
    owner_id: str,
    attempt_no: int,
    gen_type: str,
    engine_id: str | None,
    source_module: str | None = None,
    source_project_id: str | None = None,
    source_step_id: str | None = None,
    source_step_code: str | None = None,
    billing_scene: str | None = None,
) -> CreditRecordMeta:
    media_type = (gen_type or "").lower().strip() or None
    if source_module is None:
        source_module = CreditRecordSourceModule.GENERATION_RECORD.value if owner_type == CreditRecordOwnerType.GENERATION_RECORD.value else CreditRecordSourceModule.AI_CREATION.value
    if billing_scene is None:
        billing_scene = infer_module_billing_scene(source_module=source_module, source_step_code=source_step_code, media_type=media_type)
    meta = CreditRecordMeta(
        owner_type=owner_type,
        owner_id=owner_id,
        attempt_no=attempt_no,
        charge_kind=CreditRecordChargeKind.MEDIA.value,
        charge_action=CreditRecordAction.CHARGE.value,
        credit_subject=CreditRecordSubject.MEDIA.value,
        media_type=media_type,
        billing_scene=billing_scene,
        source_module=source_module,
        source_project_id=source_project_id,
        source_step_id=source_step_id,
        source_step_code=source_step_code,
    )
    for key, value in (await get_engine_snapshot(db, gen_type=media_type or "", engine_id=engine_id)).items():
        setattr(meta, key, value)
    return meta


async def build_generation_record_prompt_meta(
    db: AsyncSession,
    *,
    record_id: str,
    attempt_no: int,
    charge_kind: str,
    usage: Mapping[str, Any],
) -> CreditRecordMeta:
    scene_map = {
        CreditRecordChargeKind.TEXT_PROMPT.value: CreditRecordBillingScene.GENERATION_RECORD_TEXT_PROMPT_OPTIMIZE.value,
        CreditRecordChargeKind.FILE_PARSE.value: CreditRecordBillingScene.GENERATION_RECORD_FILE_PARSE.value,
        CreditRecordChargeKind.VISION_INPUT.value: CreditRecordBillingScene.GENERATION_RECORD_VISION_INPUT.value,
    }
    meta = CreditRecordMeta(
        owner_type=CreditRecordOwnerType.GENERATION_RECORD.value,
        owner_id=record_id,
        attempt_no=attempt_no,
        charge_kind=charge_kind,
        charge_action=CreditRecordAction.CHARGE.value,
        credit_subject=CreditRecordSubject.TEXT.value,
        billing_scene=scene_map.get(charge_kind),
        source_module=CreditRecordSourceModule.GENERATION_RECORD.value,
        token_usage_id=usage.get("token_usage_id"),
        input_tokens=_safe_int(usage.get("input_tokens")),
        output_tokens=_safe_int(usage.get("output_tokens")),
        total_tokens=_safe_int(usage.get("total_tokens"), _safe_int(usage.get("input_tokens")) + _safe_int(usage.get("output_tokens"))),
    )
    return await _apply_model_snapshot(db, meta, usage)


async def build_module_step_prompt_meta(
    db: AsyncSession,
    *,
    step_id: str,
    attempt_no: int,
    usage: Mapping[str, Any],
) -> CreditRecordMeta:
    result = await db.execute(select(ModuleGenerationStep).where(ModuleGenerationStep.id == step_id).limit(1))
    step = result.scalar_one_or_none()
    source_module = getattr(step, "module", None) if step else None
    source_step_code = getattr(step, "step_code", None) if step else None
    meta = CreditRecordMeta(
        owner_type=CreditRecordOwnerType.MODULE_GENERATION_STEP.value,
        owner_id=step_id,
        attempt_no=attempt_no,
        charge_kind=CreditRecordChargeKind.TEXT_PROMPT.value,
        charge_action=CreditRecordAction.CHARGE.value,
        credit_subject=CreditRecordSubject.TEXT.value,
        billing_scene=infer_module_billing_scene(source_module=source_module, source_step_code=source_step_code),
        source_module=source_module,
        source_project_id=getattr(step, "project_id", None) if step else None,
        source_step_id=step_id,
        source_step_code=source_step_code,
        token_usage_id=usage.get("token_usage_id"),
        input_tokens=_safe_int(usage.get("input_tokens")),
        output_tokens=_safe_int(usage.get("output_tokens")),
        total_tokens=_safe_int(usage.get("total_tokens"), _safe_int(usage.get("input_tokens")) + _safe_int(usage.get("output_tokens"))),
    )
    return await _apply_model_snapshot(db, meta, usage)


async def build_shot_video_analysis_meta(
    db: AsyncSession,
    *,
    owner_type: str,
    owner_id: str,
    attempt_no: int,
    usage: Mapping[str, Any],
    billing_scene: str,
    source_project_id: str | None = None,
    source_step_id: str | None = None,
) -> CreditRecordMeta:
    meta = CreditRecordMeta(
        owner_type=owner_type,
        owner_id=owner_id,
        attempt_no=attempt_no,
        charge_kind=CreditRecordChargeKind.VIDEO_ANALYSIS.value,
        charge_action=CreditRecordAction.CHARGE.value,
        credit_subject=CreditRecordSubject.ANALYSIS.value,
        media_type="video",
        billing_scene=billing_scene,
        source_module=CreditRecordSourceModule.SHOT_REPLICATE.value,
        source_project_id=source_project_id,
        source_step_id=source_step_id,
        source_step_code=CreditRecordSourceStepCode.VIDEO_ANALYSIS.value,
        token_usage_id=usage.get("token_usage_id"),
        input_tokens=_safe_int(usage.get("input_tokens")),
        output_tokens=_safe_int(usage.get("output_tokens")),
        total_tokens=_safe_int(usage.get("total_tokens"), _safe_int(usage.get("input_tokens")) + _safe_int(usage.get("output_tokens"))),
    )
    return await _apply_model_snapshot(db, meta, usage)


def build_refund_meta_from_charge(charge: Any, *, attempt_no: int | None = None) -> CreditRecordMeta:
    return CreditRecordMeta(
        owner_type=getattr(charge, "owner_type", None),
        owner_id=getattr(charge, "owner_id", None),
        attempt_no=attempt_no or getattr(charge, "attempt_no", None),
        charge_kind=getattr(charge, "charge_kind", None),
        charge_action=CreditRecordAction.REFUND.value,
        credit_subject=getattr(charge, "credit_subject", None) or CreditRecordSubject.REFUND.value,
        media_type=getattr(charge, "media_type", None),
        billing_scene=CreditRecordBillingScene.REFUND.value,
        source_module=getattr(charge, "source_module", None),
        source_project_id=getattr(charge, "source_project_id", None),
        source_step_id=getattr(charge, "source_step_id", None),
        source_step_code=getattr(charge, "source_step_code", None),
        token_usage_id=getattr(charge, "token_usage_id", None),
        input_tokens=getattr(charge, "input_tokens", None),
        output_tokens=getattr(charge, "output_tokens", None),
        total_tokens=getattr(charge, "total_tokens", None),
        engine_type=getattr(charge, "engine_type", None),
        engine_id=getattr(charge, "engine_id", None),
        engine_name=getattr(charge, "engine_name", None),
        engine_provider=getattr(charge, "engine_provider", None),
        engine_model_name=getattr(charge, "engine_model_name", None),
        user_type_snapshot=getattr(charge, "user_type_snapshot", None),
        frontend_user_kind_snapshot=getattr(charge, "frontend_user_kind_snapshot", None),
    )
