from collections.abc import Callable
from datetime import datetime, timezone
from typing import NoReturn

from fastapi import HTTPException, status
from sqlalchemy import delete, func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from app.core.field_crypto import encrypt_text
from app.core.masking import mask_mobile
from app.core.security import create_access_token
from app.models.rbac import (
    AdminResource,
    AdminRole,
    AdminRoleResource,
    AdminUser,
    AdminUserRole,
    utcnow,
)
from app.services.captcha import consume_captcha
from app.services.login_guard import clear_failures, is_currently_locked, record_failure
from app.services.password import hash_password, validate_password, verify_password

LOGIN_FAILURE = "用户名或密码错误"
KEEP_FULL_ADMIN = "至少保留一名拥有全部系统菜单的正常用户"
KEEP_SELF = "不能停用或注销当前登录用户"

USER_STATUSES = {"active", "disabled", "cancelled"}
ROLE_STATUSES = {"active", "disabled"}
RESOURCE_STATUSES = {"active", "disabled"}
RESOURCE_TYPES = {"directory", "menu", "button"}
MENU_TYPES = {"directory", "menu"}


def isoformat(value: datetime | None) -> str | None:
    if value is None:
        return None
    if value.tzinfo is None:
        value = value.replace(tzinfo=timezone.utc)
    return value.isoformat()


def _bad(detail: str, code: int = status.HTTP_400_BAD_REQUEST) -> NoReturn:
    raise HTTPException(status_code=code, detail=detail)


def _require_text(value: str | None, *, empty: str, limit: int, too_long: str) -> str:
    text = (value or "").strip()
    if not text:
        _bad(empty)
    if len(text) > limit:
        _bad(too_long)
    return text


def _optional_text(value: str | None, limit: int, too_long: str) -> str | None:
    if value is None:
        return None
    text = value.strip()
    if not text:
        return None
    if len(text) > limit:
        _bad(too_long)
    return text


def _clean(value: str | None, limit: int, too_long: str) -> str:
    text = (value or "").strip()
    if len(text) > limit:
        _bad(too_long)
    return text


def _exact(value: str | None) -> str | None:
    if value is None:
        return None
    text = value.strip()
    return text or None


def _page(current: int, size: int) -> tuple[int, int]:
    if current < 1:
        current = 1
    if size < 1:
        size = 10
    if size > 200:
        size = 200
    return current, size


def commit_or_raise(db: Session, message: str) -> None:
    try:
        db.commit()
    except IntegrityError:
        db.rollback()
        _bad(message)


def preserve_full_admin(db: Session, mutate: Callable[[], None]) -> None:
    before = count_full_admins(db)
    with db.begin_nested():
        mutate()
        db.flush()
        if before >= 1 and count_full_admins(db) < 1:
            _bad(KEEP_FULL_ADMIN)


def count_full_admins(db: Session) -> int:
    needed = set(db.scalars(select(AdminResource.id).where(AdminResource.status == "active")).all())
    user_ids = db.scalars(select(AdminUser.id).where(AdminUser.status == "active")).all()
    total = 0
    for user_id in user_ids:
        if needed <= owned_resource_ids(db, user_id):
            total += 1
    return total


def owned_resource_ids(db: Session, user_id: int) -> set[int]:
    rows = db.scalars(
        select(AdminRoleResource.resource_id)
        .join(AdminRole, AdminRole.id == AdminRoleResource.role_id)
        .join(AdminUserRole, AdminUserRole.role_id == AdminRole.id)
        .where(AdminUserRole.user_id == user_id, AdminRole.status == "active")
    ).all()
    return set(rows)


def apply_mobile(user: AdminUser, mobile: str) -> None:
    text = mobile.strip()
    if len(text) > 20:
        _bad("手机号过长")
    if not text:
        user.mobile_cipher = None
        user.mobile_mask = ""
        return
    user.mobile_cipher = encrypt_text(text)
    user.mobile_mask = mask_mobile(text)


def ensure_role_ids(db: Session, role_ids: list[int]) -> list[int]:
    unique = list(dict.fromkeys(role_ids))
    if not unique:
        return []
    found = set(db.scalars(select(AdminRole.id).where(AdminRole.id.in_(unique))).all())
    if found != set(unique):
        _bad("角色不存在")
    return unique


def ensure_resource_ids(db: Session, resource_ids: list[int]) -> list[int]:
    unique = list(dict.fromkeys(resource_ids))
    if not unique:
        return []
    found = set(db.scalars(select(AdminResource.id).where(AdminResource.id.in_(unique))).all())
    if found != set(unique):
        _bad("资源不存在")
    return unique


def replace_user_roles(db: Session, user_id: int, role_ids: list[int]) -> None:
    db.execute(delete(AdminUserRole).where(AdminUserRole.user_id == user_id))
    for role_id in role_ids:
        db.add(AdminUserRole(user_id=user_id, role_id=role_id))


def replace_role_links(db: Session, role_id: int, resource_ids: list[int]) -> None:
    db.execute(delete(AdminRoleResource).where(AdminRoleResource.role_id == role_id))
    for resource_id in resource_ids:
        db.add(AdminRoleResource(role_id=role_id, resource_id=resource_id))


def _role_map(db: Session, user_ids: list[int]) -> dict[int, list[AdminRole]]:
    grouped: dict[int, list[AdminRole]] = {user_id: [] for user_id in user_ids}
    if not user_ids:
        return grouped
    rows = db.execute(
        select(AdminUserRole.user_id, AdminRole)
        .join(AdminRole, AdminRole.id == AdminUserRole.role_id)
        .where(AdminUserRole.user_id.in_(user_ids))
        .order_by(AdminRole.id)
    ).all()
    for user_id, role in rows:
        grouped.setdefault(user_id, []).append(role)
    return grouped


def _resource_id_map(db: Session, role_ids: list[int]) -> dict[int, list[int]]:
    grouped: dict[int, list[int]] = {role_id: [] for role_id in role_ids}
    if not role_ids:
        return grouped
    rows = db.execute(
        select(AdminRoleResource.role_id, AdminRoleResource.resource_id)
        .where(AdminRoleResource.role_id.in_(role_ids))
        .order_by(AdminRoleResource.resource_id)
    ).all()
    for role_id, resource_id in rows:
        grouped.setdefault(role_id, []).append(resource_id)
    return grouped


def serialize_login_user(user: AdminUser) -> dict:
    return {
        "id": user.id,
        "username": user.username,
        "displayName": user.display_name,
        "mobileMask": user.mobile_mask or "",
        "status": user.status,
    }


def serialize_user(user: AdminUser, roles: list[AdminRole]) -> dict:
    return {
        "id": user.id,
        "username": user.username,
        "displayName": user.display_name,
        "mobileMask": user.mobile_mask or "",
        "status": user.status,
        "roleIds": [role.id for role in roles],
        "roles": [{"id": role.id, "code": role.code, "name": role.name} for role in roles],
        "createdAt": isoformat(user.created_at),
        "updatedAt": isoformat(user.updated_at),
    }


def serialize_role(role: AdminRole, resource_ids: list[int]) -> dict:
    return {
        "id": role.id,
        "code": role.code,
        "name": role.name,
        "description": role.description,
        "status": role.status,
        "resourceIds": resource_ids,
        "createdAt": isoformat(role.created_at),
        "updatedAt": isoformat(role.updated_at),
    }


def serialize_node(resource: AdminResource, children: list[dict]) -> dict:
    return {
        "id": resource.id,
        "parentId": resource.parent_id,
        "type": resource.type,
        "name": resource.name,
        "title": resource.title,
        "path": resource.path or "",
        "component": resource.component or "",
        "icon": resource.icon or "",
        "permission": resource.permission or "",
        "sort": resource.sort,
        "visible": bool(resource.visible),
        "status": resource.status,
        "children": children,
    }


def build_tree(resources: list[AdminResource]) -> list[dict]:
    known = {item.id for item in resources}
    grouped: dict[int | None, list[AdminResource]] = {}
    for item in resources:
        parent_id = item.parent_id if item.parent_id in known else None
        grouped.setdefault(parent_id, []).append(item)
    for nodes in grouped.values():
        nodes.sort(key=lambda node: (node.sort, node.id))

    def walk(node: AdminResource) -> dict:
        children = [walk(child) for child in grouped.get(node.id, [])]
        return serialize_node(node, children)

    return [walk(node) for node in grouped.get(None, [])]


def sibling_name_taken(db: Session, parent_id: int | None, name: str, exclude_id: int | None = None) -> bool:
    stmt = select(AdminResource.id).where(AdminResource.name == name)
    if parent_id is None:
        stmt = stmt.where(AdminResource.parent_id.is_(None))
    else:
        stmt = stmt.where(AdminResource.parent_id == parent_id)
    if exclude_id is not None:
        stmt = stmt.where(AdminResource.id != exclude_id)
    return db.scalar(stmt) is not None


def assert_parent(db: Session, resource_id: int | None, parent_id: int | None) -> None:
    if parent_id is None:
        return
    if resource_id is not None and parent_id == resource_id:
        _bad("不能把资源移动到自己的下级")
    seen: set[int] = set()
    current: int | None = parent_id
    while current is not None:
        if current in seen:
            break
        if resource_id is not None and current == resource_id:
            _bad("不能把资源移动到自己的下级")
        seen.add(current)
        parent = db.get(AdminResource, current)
        if parent is None:
            _bad("上级资源不存在")
        current = parent.parent_id


def granted_resources(db: Session, user_id: int) -> list[AdminResource]:
    stmt = select(AdminResource).where(
        AdminResource.status == "active",
        AdminResource.id.in_(
            select(AdminRoleResource.resource_id)
            .join(AdminRole, AdminRole.id == AdminRoleResource.role_id)
            .join(AdminUserRole, AdminUserRole.role_id == AdminRole.id)
            .where(AdminUserRole.user_id == user_id, AdminRole.status == "active")
        ),
    )
    return list(db.scalars(stmt).all())


def menus_and_permissions(db: Session, user: AdminUser) -> tuple[list[dict], list[str]]:
    resources = granted_resources(db, user.id)
    menus = build_tree([item for item in resources if item.type in MENU_TYPES])
    permissions = sorted({item.permission for item in resources if item.type == "button" and item.permission})
    return menus, permissions


def _reject_login(db: Session, user: AdminUser | None, *, record: bool) -> NoReturn:
    if record and user is not None:
        record_failure(db, user)
        db.commit()
    _bad(LOGIN_FAILURE, status.HTTP_401_UNAUTHORIZED)


def authenticate(db: Session, username: str, password: str, captcha_id: str, captcha_code: str) -> dict:
    captcha_ok = consume_captcha(captcha_id, captcha_code)
    user = db.scalar(select(AdminUser).where(AdminUser.username == (username or "").strip()))
    if user is None:
        _reject_login(db, None, record=False)
    if is_currently_locked(user):
        _reject_login(db, user, record=False)
    if not captcha_ok:
        _reject_login(db, user, record=True)
    if not verify_password(password, user.password_hash):
        _reject_login(db, user, record=True)
    if user.status != "active":
        _reject_login(db, user, record=True)
    clear_failures(db, user)
    token = create_access_token(str(user.id), "staff")
    menus, permissions = menus_and_permissions(db, user)
    db.commit()
    return {
        "token": token,
        "user": serialize_login_user(user),
        "menus": menus,
        "permissions": permissions,
    }


def change_own_password(db: Session, user_id: int, old_password: str, new_password: str) -> dict:
    user = db.get(AdminUser, user_id)
    if user is None:
        _bad("登录已失效", status.HTTP_401_UNAUTHORIZED)
    if not verify_password(old_password, user.password_hash):
        _bad("旧密码错误")
    validate_password(new_password)
    user.password_hash = hash_password(new_password)
    user.updated_at = utcnow()
    commit_or_raise(db, "保存密码失败")
    return {"ok": True}


def query_users(
    db: Session,
    *,
    current: int,
    size: int,
    username: str | None,
    display_name: str | None,
    status_value: str | None,
) -> dict:
    current, size = _page(current, size)
    filters = []
    username_text = _exact(username)
    display_text = _exact(display_name)
    status_text = _exact(status_value)
    if username_text:
        filters.append(AdminUser.username.contains(username_text))
    if display_text:
        filters.append(AdminUser.display_name.contains(display_text))
    if status_text:
        filters.append(AdminUser.status == status_text)
    total = db.scalar(select(func.count()).select_from(AdminUser).where(*filters)) or 0
    rows = db.scalars(
        select(AdminUser).where(*filters).order_by(AdminUser.id.desc()).offset((current - 1) * size).limit(size)
    ).all()
    role_map = _role_map(db, [row.id for row in rows])
    return {
        "records": [serialize_user(row, role_map.get(row.id, [])) for row in rows],
        "current": current,
        "size": size,
        "total": total,
    }


def create_user(
    db: Session,
    *,
    username: str,
    password: str,
    display_name: str,
    mobile: str | None,
    role_ids: list[int],
) -> dict:
    username_text = _require_text(username, empty="请填写用户名", limit=64, too_long="用户名过长")
    display_text = _require_text(display_name, empty="请填写姓名", limit=64, too_long="姓名过长")
    validate_password(password)
    checked_roles = ensure_role_ids(db, role_ids)
    if db.scalar(select(AdminUser.id).where(AdminUser.username == username_text)):
        _bad("用户名已存在")
    now = utcnow()
    user = AdminUser(
        username=username_text,
        password_hash=hash_password(password),
        display_name=display_text,
        mobile_mask="",
        status="active",
        failed_login_count=0,
        created_at=now,
        updated_at=now,
    )
    if mobile:
        apply_mobile(user, mobile)
    db.add(user)
    try:
        db.flush()
    except IntegrityError:
        db.rollback()
        _bad("用户名已存在")
    replace_user_roles(db, user.id, checked_roles)
    commit_or_raise(db, "用户名已存在")
    roles = _role_map(db, [user.id]).get(user.id, [])
    return serialize_user(user, roles)


def update_user(
    db: Session,
    user_id: int,
    actor_id: int,
    *,
    display_name: str,
    status_value: str,
    role_ids: list[int],
    mobile: str | None,
    mobile_provided: bool,
) -> dict:
    user = db.get(AdminUser, user_id)
    if user is None:
        _bad("用户不存在", status.HTTP_404_NOT_FOUND)
    display_text = _require_text(display_name, empty="请填写姓名", limit=64, too_long="姓名过长")
    if status_value not in USER_STATUSES:
        _bad("用户状态不正确")
    if status_value in {"disabled", "cancelled"} and user.id == actor_id:
        _bad(KEEP_SELF)
    checked_roles = ensure_role_ids(db, role_ids)
    normalized_mobile = mobile
    if mobile_provided and mobile is not None:
        text = mobile.strip()
        if len(text) > 20:
            _bad("手机号过长")
        normalized_mobile = text

    def mutate() -> None:
        user.display_name = display_text
        user.status = status_value
        if mobile_provided and normalized_mobile is not None:
            apply_mobile(user, normalized_mobile)
        replace_user_roles(db, user.id, checked_roles)
        user.updated_at = utcnow()

    preserve_full_admin(db, mutate)
    commit_or_raise(db, "保存用户失败")
    roles = _role_map(db, [user.id]).get(user.id, [])
    return serialize_user(user, roles)


def reset_user_password(db: Session, user_id: int, password: str) -> dict:
    user = db.get(AdminUser, user_id)
    if user is None:
        _bad("用户不存在", status.HTTP_404_NOT_FOUND)
    validate_password(password)
    user.password_hash = hash_password(password)
    user.updated_at = utcnow()
    commit_or_raise(db, "保存密码失败")
    return {"ok": True}


def query_roles(
    db: Session,
    *,
    current: int,
    size: int,
    name: str | None,
    code: str | None,
    status_value: str | None,
) -> dict:
    current, size = _page(current, size)
    filters = []
    name_text = _exact(name)
    code_text = _exact(code)
    status_text = _exact(status_value)
    if name_text:
        filters.append(AdminRole.name.contains(name_text))
    if code_text:
        filters.append(AdminRole.code.contains(code_text))
    if status_text:
        filters.append(AdminRole.status == status_text)
    total = db.scalar(select(func.count()).select_from(AdminRole).where(*filters)) or 0
    rows = db.scalars(
        select(AdminRole).where(*filters).order_by(AdminRole.id.desc()).offset((current - 1) * size).limit(size)
    ).all()
    resource_map = _resource_id_map(db, [row.id for row in rows])
    return {
        "records": [serialize_role(row, resource_map.get(row.id, [])) for row in rows],
        "current": current,
        "size": size,
        "total": total,
    }


def create_role(db: Session, *, code: str, name: str, description: str | None) -> dict:
    code_text = _require_text(code, empty="请填写角色编码", limit=64, too_long="角色编码过长")
    name_text = _require_text(name, empty="请填写角色名称", limit=64, too_long="角色名称过长")
    description_text = _optional_text(description, 2000, "说明过长")
    if db.scalar(select(AdminRole.id).where(AdminRole.code == code_text)):
        _bad("角色编码已存在")
    now = utcnow()
    role = AdminRole(
        code=code_text,
        name=name_text,
        description=description_text,
        status="active",
        created_at=now,
        updated_at=now,
    )
    db.add(role)
    commit_or_raise(db, "角色编码已存在")
    return serialize_role(role, [])


def update_role(
    db: Session,
    role_id: int,
    *,
    name: str,
    description: str | None,
    status_value: str,
) -> dict:
    role = db.get(AdminRole, role_id)
    if role is None:
        _bad("角色不存在", status.HTTP_404_NOT_FOUND)
    name_text = _require_text(name, empty="请填写角色名称", limit=64, too_long="角色名称过长")
    description_text = _optional_text(description, 2000, "说明过长")
    if status_value not in ROLE_STATUSES:
        _bad("角色状态不正确")

    def mutate() -> None:
        role.name = name_text
        role.description = description_text
        role.status = status_value
        role.updated_at = utcnow()

    preserve_full_admin(db, mutate)
    commit_or_raise(db, "保存角色失败")
    resource_ids = _resource_id_map(db, [role.id]).get(role.id, [])
    return serialize_role(role, resource_ids)


def replace_role_resources(db: Session, role_id: int, resource_ids: list[int]) -> dict:
    role = db.get(AdminRole, role_id)
    if role is None:
        _bad("角色不存在", status.HTTP_404_NOT_FOUND)
    checked = ensure_resource_ids(db, resource_ids)

    def mutate() -> None:
        replace_role_links(db, role.id, checked)
        role.updated_at = utcnow()

    preserve_full_admin(db, mutate)
    commit_or_raise(db, "保存角色资源失败")
    return {"ok": True}


def list_resources(db: Session) -> list[dict]:
    rows = list(db.scalars(select(AdminResource)).all())
    return build_tree(rows)


def _resource_fields(
    *,
    resource_type: str,
    name: str,
    title: str,
    path: str | None,
    component: str | None,
    icon: str | None,
    permission: str | None,
    sort: int,
    visible: bool,
    status_value: str,
) -> dict:
    if resource_type not in RESOURCE_TYPES:
        _bad("资源类型不正确")
    if status_value not in RESOURCE_STATUSES:
        _bad("资源状态不正确")
    return {
        "type": resource_type,
        "name": _require_text(name, empty="请填写名称", limit=64, too_long="名称过长"),
        "title": _require_text(title, empty="请填写标题", limit=64, too_long="标题过长"),
        "path": _clean(path, 255, "路由过长"),
        "component": _clean(component, 255, "组件路径过长"),
        "icon": _clean(icon, 64, "图标过长"),
        "permission": _clean(permission, 64, "权限码过长"),
        "sort": sort,
        "visible": bool(visible),
        "status": status_value,
    }


def create_resource(db: Session, payload: dict) -> dict:
    parent_id = payload["parent_id"]
    fields = _resource_fields(
        resource_type=payload["type"],
        name=payload["name"],
        title=payload["title"],
        path=payload.get("path"),
        component=payload.get("component"),
        icon=payload.get("icon"),
        permission=payload.get("permission"),
        sort=payload.get("sort") or 0,
        visible=payload.get("visible", True),
        status_value=payload.get("status") or "active",
    )
    assert_parent(db, None, parent_id)
    if sibling_name_taken(db, parent_id, fields["name"]):
        _bad("同级名称已存在")
    resource = AdminResource(parent_id=parent_id, **fields)
    db.add(resource)
    commit_or_raise(db, "同级名称已存在")
    return serialize_node(resource, [])


def update_resource(db: Session, resource_id: int, payload: dict) -> dict:
    resource = db.get(AdminResource, resource_id)
    if resource is None:
        _bad("资源不存在", status.HTTP_404_NOT_FOUND)
    parent_id = payload["parent_id"]
    fields = _resource_fields(
        resource_type=payload["type"],
        name=payload["name"],
        title=payload["title"],
        path=payload.get("path"),
        component=payload.get("component"),
        icon=payload.get("icon"),
        permission=payload.get("permission"),
        sort=payload.get("sort") if payload.get("sort") is not None else resource.sort,
        visible=payload.get("visible", resource.visible),
        status_value=payload.get("status") or resource.status,
    )
    assert_parent(db, resource.id, parent_id)
    if sibling_name_taken(db, parent_id, fields["name"], exclude_id=resource.id):
        _bad("同级名称已存在")
    resource.parent_id = parent_id
    for key, value in fields.items():
        setattr(resource, key, value)
    commit_or_raise(db, "同级名称已存在")
    return serialize_node(resource, [])


def delete_resource(db: Session, resource_id: int) -> dict:
    resource = db.get(AdminResource, resource_id)
    if resource is None:
        _bad("资源不存在", status.HTTP_404_NOT_FOUND)
    child_id = db.scalar(select(AdminResource.id).where(AdminResource.parent_id == resource_id).limit(1))
    if child_id is not None:
        _bad("存在子节点，不能删除")
    db.execute(delete(AdminRoleResource).where(AdminRoleResource.resource_id == resource_id))
    db.delete(resource)
    commit_or_raise(db, "存在子节点，不能删除")
    return {"ok": True}