from __future__ import annotations

from datetime import datetime
from typing import Any

from sqlalchemy import and_, case, distinct, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.enums.credit_record import (
    CREDIT_RECORD_BILLING_SCENE_LABELS,
    CREDIT_RECORD_CHARGE_KIND_LABELS,
    CREDIT_RECORD_MEDIA_TYPE_LABELS,
    CREDIT_RECORD_SOURCE_MODULE_LABELS,
    CREDIT_RECORD_SOURCE_STEP_CODE_LABELS,
    CREDIT_RECORD_SUBJECT_LABELS,
    CREDIT_RECORD_TYPE_LABELS,
    CreditRecordSubject,
)
from app.enums.user import FRONTEND_USER_KIND_LABELS, USER_TYPE_LABELS, UserType
from app.models.chat_generation_task import ChatGenerationTask
from app.models.credit_record import CreditRecord
from app.models.generation_record import GenerationRecord
from app.models.module_generation_project import ModuleGenerationProject
from app.models.module_generation_step import ModuleGenerationStep
from app.models.shot_replicate_segment import ShotReplicateSegment
from app.models.shot_replicate_task_set import ShotReplicateTaskSet
from app.models.user import User


def _iso(dt: Any) -> str | None:
    if dt is None:
        return None
    try:
        return dt.isoformat()
    except Exception:
        return str(dt)


def _round2(value: Any) -> float:
    try:
        return round(float(value or 0), 2)
    except Exception:
        return 0.0


def _as_date_start(value: str | None) -> datetime | None:
    if not value:
        return None
    try:
        return datetime.strptime(value, "%Y-%m-%d")
    except Exception:
        return None


def _as_date_end(value: str | None) -> datetime | None:
    if not value:
        return None
    try:
        return datetime.strptime(value, "%Y-%m-%d").replace(hour=23, minute=59, second=59, microsecond=999999)
    except Exception:
        return None


def _label(mapping: dict[str, str], value: str | None) -> str:
    if not value:
        return "-"
    return mapping.get(value, value)


def _build_filters(
    *,
    user_id: str | None = None,
    user_name: str | None = None,
    user_type: str | None = None,
    frontend_user_kind: str | None = None,
    record_type: str | None = None,
    credit_subject: str | None = None,
    media_type: str | None = None,
    charge_kind: str | None = None,
    source_module: str | None = None,
    source_step_code: str | None = None,
    billing_scene: str | None = None,
    start_date: str | None = None,
    end_date: str | None = None,
) -> list[Any]:
    filters: list[Any] = []
    if user_id:
        filters.append(CreditRecord.user_id == user_id)
    if user_name:
        like = f"%{user_name}%"
        filters.append(or_(User.username.ilike(like), User.phone.ilike(like), User.email.ilike(like)))
    if user_type:
        filters.append(CreditRecord.user_type_snapshot == user_type)
    if frontend_user_kind:
        filters.append(CreditRecord.frontend_user_kind_snapshot == frontend_user_kind)
        filters.append(CreditRecord.user_type_snapshot == UserType.FRONTEND.value)
    if record_type:
        filters.append(CreditRecord.type == record_type)
    if credit_subject:
        filters.append(CreditRecord.credit_subject == credit_subject)
    if media_type:
        filters.append(CreditRecord.media_type == media_type)
    if charge_kind:
        filters.append(CreditRecord.charge_kind == charge_kind)
    if source_module:
        filters.append(CreditRecord.source_module == source_module)
    if source_step_code:
        filters.append(CreditRecord.source_step_code == source_step_code)
    if billing_scene:
        filters.append(CreditRecord.billing_scene == billing_scene)
    start = _as_date_start(start_date)
    end = _as_date_end(end_date)
    if start:
        filters.append(CreditRecord.created_at >= start)
    if end:
        filters.append(CreditRecord.created_at <= end)
    return filters


async def _load_deleted_map(db: AsyncSession, records: list[CreditRecord]) -> dict[tuple[str, str], tuple[bool, str | None]]:
    grouped: dict[str, set[str]] = {}
    for record in records:
        if record.owner_type and record.owner_id:
            grouped.setdefault(record.owner_type, set()).add(record.owner_id)

    model_map: dict[str, Any] = {
        "chat_generation_task": ChatGenerationTask,
        "generation_record": GenerationRecord,
        "module_generation_project": ModuleGenerationProject,
        "module_generation_step": ModuleGenerationStep,
        "shot_replicate_task_set": ShotReplicateTaskSet,
        "shot_replicate_segment": ShotReplicateSegment,
    }
    deleted_map: dict[tuple[str, str], tuple[bool, str | None]] = {}
    for owner_type, ids in grouped.items():
        model = model_map.get(owner_type)
        if not model or not ids:
            continue
        result = await db.execute(select(model.id, model.deleted_at).where(model.id.in_(ids)))
        found = {row[0]: row[1] for row in result.all()}
        for owner_id in ids:
            deleted_at = found.get(owner_id)
            deleted_map[(owner_type, owner_id)] = (bool(deleted_at), _iso(deleted_at)) if owner_id in found else (False, None)
    return deleted_map


def _record_to_item(record: CreditRecord, user: User | None, deleted_map: dict[tuple[str, str], tuple[bool, str | None]]) -> dict[str, Any]:
    owner_deleted = False
    owner_deleted_at = None
    if record.owner_type and record.owner_id:
        owner_deleted, owner_deleted_at = deleted_map.get((record.owner_type, record.owner_id), (False, None))

    user_type = record.user_type_snapshot or (user.user_type if user else None)
    frontend_kind = record.frontend_user_kind_snapshot or (getattr(user, "frontend_user_kind", None) if user else None)
    return {
        "id": record.id,
        "user_id": record.user_id,
        "username": user.username if user else None,
        "phone": user.phone if user else None,
        "email": user.email if user else None,
        "user_type": user_type,
        "user_type_label": _label(USER_TYPE_LABELS, user_type),
        "frontend_user_kind": frontend_kind,
        "frontend_user_kind_label": _label(FRONTEND_USER_KIND_LABELS, frontend_kind),
        "type": record.type,
        "record_type": record.type,
        "record_type_label": _label(CREDIT_RECORD_TYPE_LABELS, record.type),
        "amount": _round2(record.amount),
        "balance_after": _round2(record.balance_after),
        "description": record.description,
        "related_id": record.related_id,
        "biz_key": record.biz_key,
        "refund_for_biz_key": record.refund_for_biz_key,
        "owner_type": record.owner_type,
        "owner_id": record.owner_id,
        "owner_deleted": owner_deleted,
        "owner_deleted_at": owner_deleted_at,
        "attempt_no": record.attempt_no,
        "charge_kind": record.charge_kind,
        "charge_kind_label": _label(CREDIT_RECORD_CHARGE_KIND_LABELS, record.charge_kind),
        "charge_action": record.charge_action,
        "credit_subject": record.credit_subject,
        "credit_subject_label": _label(CREDIT_RECORD_SUBJECT_LABELS, record.credit_subject),
        "media_type": record.media_type,
        "media_type_label": _label(CREDIT_RECORD_MEDIA_TYPE_LABELS, record.media_type),
        "billing_scene": record.billing_scene,
        "billing_scene_label": _label(CREDIT_RECORD_BILLING_SCENE_LABELS, record.billing_scene),
        "source_module": record.source_module,
        "source_module_label": _label(CREDIT_RECORD_SOURCE_MODULE_LABELS, record.source_module),
        "source_project_id": record.source_project_id,
        "source_step_id": record.source_step_id,
        "source_step_code": record.source_step_code,
        "source_step_code_label": _label(CREDIT_RECORD_SOURCE_STEP_CODE_LABELS, record.source_step_code),
        "token_usage_id": record.token_usage_id,
        "input_tokens": record.input_tokens or 0,
        "output_tokens": record.output_tokens or 0,
        "total_tokens": record.total_tokens or 0,
        "engine_type": record.engine_type,
        "engine_id": record.engine_id,
        "engine_name": record.engine_name,
        "engine_provider": record.engine_provider,
        "engine_model_name": record.engine_model_name,
        "created_at": _iso(record.created_at),
    }


async def list_admin_credit_records(
    db: AsyncSession,
    *,
    page: int,
    page_size: int,
    user_id: str | None = None,
    user_name: str | None = None,
    user_type: str | None = None,
    frontend_user_kind: str | None = None,
    record_type: str | None = None,
    credit_subject: str | None = None,
    media_type: str | None = None,
    charge_kind: str | None = None,
    source_module: str | None = None,
    source_step_code: str | None = None,
    billing_scene: str | None = None,
    start_date: str | None = None,
    end_date: str | None = None,
) -> dict[str, Any]:
    page = max(int(page or 1), 1)
    page_size = min(max(int(page_size or 20), 1), 1000)
    filters = _build_filters(
        user_id=user_id,
        user_name=user_name,
        user_type=user_type,
        frontend_user_kind=frontend_user_kind,
        record_type=record_type,
        credit_subject=credit_subject,
        media_type=media_type,
        charge_kind=charge_kind,
        source_module=source_module,
        source_step_code=source_step_code,
        billing_scene=billing_scene,
        start_date=start_date,
        end_date=end_date,
    )
    where_clause = and_(*filters) if filters else None

    base_query = select(CreditRecord, User).join(User, CreditRecord.user_id == User.id, isouter=True)
    count_query = select(func.count(CreditRecord.id)).select_from(CreditRecord).join(User, CreditRecord.user_id == User.id, isouter=True)
    if where_clause is not None:
        base_query = base_query.where(where_clause)
        count_query = count_query.where(where_clause)
    total = (await db.execute(count_query)).scalar() or 0

    result = await db.execute(
        base_query.order_by(CreditRecord.created_at.desc(), CreditRecord.id.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
    )
    rows = result.all()
    records = [row[0] for row in rows]
    deleted_map = await _load_deleted_map(db, records)
    items = [_record_to_item(record, user, deleted_map) for record, user in rows]

    summary_query = select(
        func.coalesce(func.sum(case((CreditRecord.type == "recharge", CreditRecord.amount), else_=0)), 0),
        func.coalesce(func.sum(case((CreditRecord.type == "consume", func.abs(CreditRecord.amount)), else_=0)), 0),
        func.coalesce(func.sum(case((CreditRecord.type == "refund", CreditRecord.amount), else_=0)), 0),
        func.count(CreditRecord.id),
        func.count(distinct(case((and_(CreditRecord.credit_subject == CreditRecordSubject.MEDIA.value, CreditRecord.type == "consume"), func.concat(CreditRecord.owner_type, ":", CreditRecord.owner_id)), else_=None))),
        func.count(case((and_(CreditRecord.credit_subject == CreditRecordSubject.MEDIA.value, CreditRecord.type == "consume"), 1), else_=None)),
        func.count(distinct(case((and_(CreditRecord.credit_subject == CreditRecordSubject.MEDIA.value, CreditRecord.media_type == "image", CreditRecord.type == "consume"), func.concat(CreditRecord.owner_type, ":", CreditRecord.owner_id)), else_=None))),
        func.count(distinct(case((and_(CreditRecord.credit_subject == CreditRecordSubject.MEDIA.value, CreditRecord.media_type == "video", CreditRecord.type == "consume"), func.concat(CreditRecord.owner_type, ":", CreditRecord.owner_id)), else_=None))),
        func.coalesce(func.sum(case((and_(CreditRecord.credit_subject == CreditRecordSubject.MEDIA.value, CreditRecord.media_type == "image", CreditRecord.type == "consume"), func.abs(CreditRecord.amount)), else_=0)), 0),
        func.coalesce(func.sum(case((and_(CreditRecord.credit_subject == CreditRecordSubject.MEDIA.value, CreditRecord.media_type == "video", CreditRecord.type == "consume"), func.abs(CreditRecord.amount)), else_=0)), 0),
        func.coalesce(func.sum(case((and_(CreditRecord.credit_subject == CreditRecordSubject.TEXT.value, CreditRecord.type == "consume"), func.abs(CreditRecord.amount)), else_=0)), 0),
        func.coalesce(func.sum(case((and_(CreditRecord.credit_subject == CreditRecordSubject.ANALYSIS.value, CreditRecord.type == "consume"), func.abs(CreditRecord.amount)), else_=0)), 0),
        func.coalesce(func.sum(CreditRecord.total_tokens), 0),
        func.coalesce(func.sum(CreditRecord.input_tokens), 0),
        func.coalesce(func.sum(CreditRecord.output_tokens), 0),
    ).select_from(CreditRecord).join(User, CreditRecord.user_id == User.id, isouter=True)
    if where_clause is not None:
        summary_query = summary_query.where(where_clause)
    s = (await db.execute(summary_query)).one()
    summary = {
        "total_recharge": _round2(s[0]),
        "total_consume": _round2(s[1]),
        "total_refund": _round2(s[2]),
        "transaction_count": int(s[3] or 0),
        "generation_count": int(s[4] or 0),
        "generation_attempt_count": int(s[5] or 0),
        "image_generation_count": int(s[6] or 0),
        "video_generation_count": int(s[7] or 0),
        "image_consume": _round2(s[8]),
        "video_consume": _round2(s[9]),
        "text_consume": _round2(s[10]),
        "analysis_consume": _round2(s[11]),
        "total_tokens": int(s[12] or 0),
        "input_tokens": int(s[13] or 0),
        "output_tokens": int(s[14] or 0),
    }
    return {"items": items, "total": total, "summary": summary}
