"""Asset brand + model lookup service.

Two layers:

  * **Admin curation** — full CRUD on ``asset_brands`` / ``asset_models``,
    surfaced under ``/admin/assets/settings``. Admins can deactivate
    rows the on-the-fly auto-create has created (``auto_created=true``)
    after they've been merged / consolidated.
  * **Intake auto-create** — ``ensure_brand`` / ``ensure_model`` are
    idempotent helpers used by the asset-create + intake services so
    typing a free-text value in the intake AutoComplete creates a
    lookup row at write time. The new row is flagged
    ``auto_created=true`` so admins can see what came from typed
    input vs. curated entries.

Cross-domain rule: this service NEVER imports from cmdb / approval /
hr / etc. — keeps the asset domain self-contained per the project
architecture rules in CLAUDE.md.
"""
from __future__ import annotations

from uuid import UUID

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

from app.core.exceptions import AppException, NotFoundError
from app.domains.asset.models import AssetBrand, AssetItem, AssetModel
from app.domains.asset.schemas import (
    AssetBrandCreateRequest,
    AssetBrandResponse,
    AssetBrandUpdateRequest,
    AssetModelCreateRequest,
    AssetModelResponse,
    AssetModelUpdateRequest,
)


class AssetBrandModelService:
    """All read / write operations on the asset brand + model lookup."""

    def __init__(self, db: AsyncSession) -> None:
        self.db = db

    # ── Brands ──────────────────────────────────────────────────

    async def list_brands(
        self, *, include_inactive: bool = False, with_counts: bool = True,
    ) -> list[AssetBrandResponse]:
        """Return brands sorted by ``sort_order, name``.

        ``with_counts=True`` adds ``model_count`` to each row via a
        single GROUP BY query (no N+1) so the admin settings page can
        render "Apple — 4 models". Pass False on hot paths (intake
        AutoComplete) where the count isn't needed.
        """
        stmt = select(AssetBrand).order_by(
            AssetBrand.sort_order, AssetBrand.name,
        )
        if not include_inactive:
            stmt = stmt.where(AssetBrand.is_active.is_(True))
        rows = (await self.db.execute(stmt)).scalars().all()

        counts: dict[UUID, int] = {}
        if with_counts and rows:
            count_rows = (await self.db.execute(
                select(
                    AssetModel.brand_id, func.count(AssetModel.id),
                )
                .where(AssetModel.brand_id.in_([r.id for r in rows]))
                .group_by(AssetModel.brand_id)
            )).all()
            counts = {row[0]: int(row[1]) for row in count_rows}

        out = []
        for r in rows:
            item = AssetBrandResponse.model_validate(r)
            if with_counts:
                item = item.model_copy(
                    update={"model_count": counts.get(r.id, 0)},
                )
            out.append(item)
        return out

    async def get_brand(self, brand_id: UUID) -> AssetBrand:
        row = await self.db.get(AssetBrand, brand_id)
        if row is None:
            raise NotFoundError("AssetBrand", str(brand_id))
        return row

    async def create_brand(
        self, body: AssetBrandCreateRequest,
    ) -> AssetBrandResponse:
        """Curated create from the admin settings page.

        Distinct from ``ensure_brand`` (the intake auto-create path):
        this one fails on duplicate name (409) and sets
        ``auto_created=False``. The duplicate check is case-insensitive
        because the unique key is on the ``name`` column with default
        collation; LOWER(name) keeps the comparison consistent across
        DBs.
        """
        existing = await self._find_brand_by_name(body.name)
        if existing is not None:
            raise AppException(
                code="BRAND_DUPLICATE",
                message=f"品牌 {body.name!r} 已存在。",
                status_code=409,
            )
        row = AssetBrand(
            name=body.name.strip(),
            display_name=body.display_name,
            notes=body.notes,
            sort_order=body.sort_order,
            is_active=True,
            auto_created=False,
        )
        self.db.add(row)
        await self.db.flush()
        return AssetBrandResponse.model_validate(row)

    async def update_brand(
        self, brand_id: UUID, body: AssetBrandUpdateRequest,
    ) -> AssetBrandResponse:
        row = await self.get_brand(brand_id)
        if body.name is not None and body.name.strip() != row.name:
            collision = await self._find_brand_by_name(body.name)
            if collision is not None and collision.id != row.id:
                raise AppException(
                    code="BRAND_DUPLICATE",
                    message=f"品牌 {body.name!r} 已存在,无法重命名。",
                    status_code=409,
                )
            row.name = body.name.strip()
        if body.display_name is not None:
            row.display_name = body.display_name
        if body.notes is not None:
            row.notes = body.notes
        if body.sort_order is not None:
            row.sort_order = body.sort_order
        if body.is_active is not None:
            row.is_active = body.is_active
        # Editing implicitly converts an auto-created row into a
        # curated row — admin has reviewed it.
        row.auto_created = False
        await self.db.flush()
        return AssetBrandResponse.model_validate(row)

    async def delete_brand(self, brand_id: UUID) -> None:
        """Hard-delete a brand only when no models / assets reference it.

        We deliberately fail loudly rather than silently soft-delete
        because the admin UI offers ``is_active=False`` for retirement.
        Use deactivate when you want to hide a brand from the intake
        AutoComplete without losing referential integrity.
        """
        row = await self.get_brand(brand_id)
        model_count = (await self.db.execute(
            select(func.count(AssetModel.id))
            .where(AssetModel.brand_id == brand_id)
        )).scalar() or 0
        if model_count > 0:
            raise AppException(
                code="BRAND_HAS_MODELS",
                message=(
                    f"品牌 {row.name!r} 下还有 {model_count} 个型号,"
                    "请先删除或迁移这些型号。"
                ),
                status_code=409,
            )
        asset_count = (await self.db.execute(
            select(func.count(AssetItem.id))
            .where(AssetItem.brand_id == brand_id)
        )).scalar() or 0
        if asset_count > 0:
            raise AppException(
                code="BRAND_IN_USE",
                message=(
                    f"品牌 {row.name!r} 已被 {asset_count} 个资产引用,"
                    "请改为停用 (is_active=false) 以隐藏。"
                ),
                status_code=409,
            )
        await self.db.delete(row)
        await self.db.flush()

    async def _find_brand_by_name(self, name: str) -> AssetBrand | None:
        normalised = (name or "").strip()
        if not normalised:
            return None
        return (await self.db.execute(
            select(AssetBrand).where(
                func.lower(AssetBrand.name) == normalised.lower(),
            )
        )).scalars().first()

    # ── Models ─────────────────────────────────────────────────

    async def list_models(
        self,
        *,
        brand_id: UUID | None = None,
        include_inactive: bool = False,
    ) -> list[AssetModelResponse]:
        """Return models, optionally filtered by brand.

        Brand name is joined in for display so callers don't have to
        do per-row lookups.
        """
        stmt = (
            select(AssetModel, AssetBrand.name)
            .join(AssetBrand, AssetModel.brand_id == AssetBrand.id)
            .order_by(
                AssetBrand.sort_order, AssetBrand.name,
                AssetModel.sort_order, AssetModel.name,
            )
        )
        if brand_id is not None:
            stmt = stmt.where(AssetModel.brand_id == brand_id)
        if not include_inactive:
            stmt = stmt.where(AssetModel.is_active.is_(True))
        rows = (await self.db.execute(stmt)).all()
        return [
            AssetModelResponse.model_validate(m).model_copy(
                update={"brand_name": brand_name},
            )
            for (m, brand_name) in rows
        ]

    async def get_model(self, model_id: UUID) -> AssetModel:
        row = await self.db.get(AssetModel, model_id)
        if row is None:
            raise NotFoundError("AssetModel", str(model_id))
        return row

    async def create_model(
        self, body: AssetModelCreateRequest,
    ) -> AssetModelResponse:
        # Make sure brand exists so we can return the joined name.
        brand = await self.get_brand(body.brand_id)
        existing = await self._find_model_in_brand(body.brand_id, body.name)
        if existing is not None:
            raise AppException(
                code="MODEL_DUPLICATE",
                message=(
                    f"品牌 {brand.name!r} 下已存在型号 {body.name!r}。"
                ),
                status_code=409,
            )
        row = AssetModel(
            brand_id=body.brand_id,
            name=body.name.strip(),
            notes=body.notes,
            sort_order=body.sort_order,
            is_active=True,
            auto_created=False,
        )
        self.db.add(row)
        await self.db.flush()
        return AssetModelResponse.model_validate(row).model_copy(
            update={"brand_name": brand.name},
        )

    async def update_model(
        self, model_id: UUID, body: AssetModelUpdateRequest,
    ) -> AssetModelResponse:
        row = await self.get_model(model_id)
        target_brand_id = body.brand_id or row.brand_id
        if body.brand_id is not None and body.brand_id != row.brand_id:
            await self.get_brand(body.brand_id)  # validate FK
            row.brand_id = body.brand_id
        if body.name is not None and body.name.strip() != row.name:
            collision = await self._find_model_in_brand(
                target_brand_id, body.name,
            )
            if collision is not None and collision.id != row.id:
                raise AppException(
                    code="MODEL_DUPLICATE",
                    message=(
                        f"该品牌下已存在型号 {body.name!r},无法重命名。"
                    ),
                    status_code=409,
                )
            row.name = body.name.strip()
        if body.notes is not None:
            row.notes = body.notes
        if body.sort_order is not None:
            row.sort_order = body.sort_order
        if body.is_active is not None:
            row.is_active = body.is_active
        row.auto_created = False
        await self.db.flush()
        # Reload brand name for response.
        brand = await self.get_brand(row.brand_id)
        return AssetModelResponse.model_validate(row).model_copy(
            update={"brand_name": brand.name},
        )

    async def delete_model(self, model_id: UUID) -> None:
        row = await self.get_model(model_id)
        asset_count = (await self.db.execute(
            select(func.count(AssetItem.id))
            .where(AssetItem.model_id == model_id)
        )).scalar() or 0
        if asset_count > 0:
            raise AppException(
                code="MODEL_IN_USE",
                message=(
                    f"型号 {row.name!r} 已被 {asset_count} 个资产引用,"
                    "请改为停用 (is_active=false) 以隐藏。"
                ),
                status_code=409,
            )
        await self.db.delete(row)
        await self.db.flush()

    async def _find_model_in_brand(
        self, brand_id: UUID, name: str,
    ) -> AssetModel | None:
        normalised = (name or "").strip()
        if not normalised:
            return None
        return (await self.db.execute(
            select(AssetModel).where(
                and_(
                    AssetModel.brand_id == brand_id,
                    func.lower(AssetModel.name) == normalised.lower(),
                )
            )
        )).scalars().first()

    # ── Auto-create on write (used by AssetService.intake / create) ──

    async def ensure_brand(self, name: str) -> AssetBrand:
        """Idempotently fetch-or-create a brand row by name.

        Used when the intake form sent a free-text ``brand`` string
        (combobox AutoComplete value) instead of a curated brand_id.
        Sets ``auto_created=True`` so admins can review later.
        """
        normalised = (name or "").strip()
        if not normalised:
            raise AppException(
                code="BRAND_NAME_EMPTY",
                message="品牌名称不能为空",
                status_code=400,
            )
        existing = await self._find_brand_by_name(normalised)
        if existing is not None:
            # Reactivate if previously deactivated — admin can decide
            # to re-deactivate via the settings page.
            if not existing.is_active:
                existing.is_active = True
                await self.db.flush()
            return existing
        row = AssetBrand(
            name=normalised,
            sort_order=0,
            is_active=True,
            auto_created=True,
        )
        self.db.add(row)
        await self.db.flush()
        return row

    async def ensure_model(
        self, brand: AssetBrand, name: str,
    ) -> AssetModel:
        """Idempotently fetch-or-create a model row under a brand."""
        normalised = (name or "").strip()
        if not normalised:
            raise AppException(
                code="MODEL_NAME_EMPTY",
                message="型号名称不能为空",
                status_code=400,
            )
        existing = await self._find_model_in_brand(brand.id, normalised)
        if existing is not None:
            if not existing.is_active:
                existing.is_active = True
                await self.db.flush()
            return existing
        row = AssetModel(
            brand_id=brand.id,
            name=normalised,
            sort_order=0,
            is_active=True,
            auto_created=True,
        )
        self.db.add(row)
        await self.db.flush()
        return row

    async def resolve_brand_and_model(
        self,
        *,
        brand_id: UUID | None,
        model_id: UUID | None,
        brand_name: str | None,
        model_name: str | None,
    ) -> tuple[AssetBrand | None, AssetModel | None]:
        """Used by AssetService.intake / create / update.

        Resolution rules (priority order):
          1. ``brand_id`` set — use the curated brand, ignore ``brand_name``
             unless ``model_id`` is set under that same brand.
          2. ``brand_name`` set, no ``brand_id`` — auto-create a brand row.
          3. Neither set — return None.

        Same logic for model: ``model_id`` wins, else ``model_name``
        triggers auto-create under whichever brand we resolved (or
        Unknown if no brand). If the request supplies ``model_id`` but
        a different ``brand_id``, we trust the model_id (FK already
        pins it to a specific brand) and ignore the supplied brand_id
        to avoid contradiction.
        """
        brand: AssetBrand | None = None
        if brand_id is not None:
            brand = await self.get_brand(brand_id)
        elif brand_name and brand_name.strip():
            brand = await self.ensure_brand(brand_name)

        model: AssetModel | None = None
        if model_id is not None:
            model = await self.get_model(model_id)
            # FK pins the brand — use the model's brand to keep the
            # relationship consistent. Overrides any contradicting
            # brand_id sent by the caller.
            brand = await self.get_brand(model.brand_id)
        elif model_name and model_name.strip():
            if brand is None:
                # Hang the orphan model under "Unknown" so the lookup
                # invariant ("models always have a brand") holds. The
                # admin can re-parent it later via the settings UI.
                brand = await self.ensure_brand("Unknown")
            model = await self.ensure_model(brand, model_name)

        return brand, model