ajyy/backend/app/api/admin.py

"""管理端 API：工作台、台账办理、配置、用户/角色、审计、附件上传。"""
from typing import Annotated, Any, Dict, List, Optional

from fastapi import APIRouter, Depends, File, Form, Query, UploadFile
from sqlalchemy.orm import Session

from app.core import storage
from app.core.database import get_db
from app.core.deps import get_current_user, require_menu
from app.core.response import AppHTTPError, success_response
from app.core.security import hash_password
from app.core.storage import StorageError
from app.crud import attachment as attachment_crud
from app.crud import menu as menu_crud
from app.crud import oper_log as oper_log_crud
from app.crud import role as role_crud
from app.crud import user as user_crud
from app.models.system import SysUser
from app.schemas.business import HotlineUpdateBody, QrcodeUpdateBody, StatusUpdateBody
from app.schemas.system import (
    ResetPasswordBody,
    RoleMenusBody,
    UserCreateBody,
    UserUpdateBody,
)
from app.services import admin_ledger_service, auth_service, config_service, dashboard_service

router = APIRouter(prefix="/web/admin", tags=["admin"])


def _iso(dt) -> Optional[str]:
    if not dt:
        return None
    return dt.isoformat(sep=" ", timespec="seconds")


def _user_to_dict(db: Session, user: SysUser) -> Dict[str, Any]:
    role = user_crud.get_role_by_id(db, user.role_id)
    return {
        "id": user.id,
        "username": user.username,
        "displayName": user.display_name,
        "roleId": user.role_id,
        "roleCode": role.role_code if role else "",
        "roleName": role.role_name if role else "",
        "status": user.status,
        "createTime": _iso(user.create_time),
        "updateTime": _iso(user.update_time),
    }


def _role_to_dict(role) -> Dict[str, Any]:
    return {
        "id": role.id,
        "roleCode": role.role_code,
        "roleName": role.role_name,
        "remark": role.remark or "",
    }


def _oper_log_to_dict(row) -> Dict[str, Any]:
    return {
        "id": row.id,
        "userId": row.user_id,
        "username": row.username,
        "module": row.module,
        "action": row.action,
        "detail": row.detail,
        "ip": row.ip,
        "createTime": _iso(row.create_time),
    }


@router.get("/dashboard")
def dashboard(
    db: Annotated[Session, Depends(get_db)],
    _user: Annotated[SysUser, Depends(get_current_user)],
):
    return success_response(dashboard_service.get_dashboard(db))


@router.get("/feedbacks")
def list_feedbacks(
    db: Annotated[Session, Depends(get_db)],
    _user: Annotated[SysUser, Depends(get_current_user)],
    status: Optional[str] = Query(None),
    fbType: Optional[str] = Query(None),
    keyword: Optional[str] = Query(None),
    page: int = Query(1, ge=1),
    pageSize: int = Query(10, ge=1, le=100),
):
    return success_response(
        admin_ledger_service.list_feedbacks_page(
            db,
            status=status,
            fb_type=fbType,
            keyword=keyword,
            page=page,
            page_size=pageSize,
        )
    )


@router.get("/feedbacks/{feedback_id}")
def get_feedback(
    feedback_id: str,
    db: Annotated[Session, Depends(get_db)],
    _user: Annotated[SysUser, Depends(get_current_user)],
):
    return success_response(
        admin_ledger_service.get_feedback_detail(db, feedback_id)
    )


@router.put("/feedbacks/{feedback_id}/status")
def update_feedback_status(
    feedback_id: str,
    body: StatusUpdateBody,
    db: Annotated[Session, Depends(get_db)],
    user: Annotated[SysUser, Depends(require_menu("feedback:handle"))],
):
    return success_response(
        admin_ledger_service.update_feedback_status(
            db,
            user,
            feedback_id,
            status=body.status,
            handle_remark=body.handle_remark,
        )
    )


@router.get("/suggestions")
def list_suggestions(
    db: Annotated[Session, Depends(get_db)],
    _user: Annotated[SysUser, Depends(get_current_user)],
    status: Optional[str] = Query(None),
    keyword: Optional[str] = Query(None),
    page: int = Query(1, ge=1),
    pageSize: int = Query(10, ge=1, le=100),
):
    return success_response(
        admin_ledger_service.list_suggestions_page(
            db,
            status=status,
            keyword=keyword,
            page=page,
            page_size=pageSize,
        )
    )


@router.get("/suggestions/{suggestion_id}")
def get_suggestion(
    suggestion_id: str,
    db: Annotated[Session, Depends(get_db)],
    _user: Annotated[SysUser, Depends(get_current_user)],
):
    return success_response(
        admin_ledger_service.get_suggestion_detail(db, suggestion_id)
    )


@router.put("/suggestions/{suggestion_id}/status")
def update_suggestion_status(
    suggestion_id: str,
    body: StatusUpdateBody,
    db: Annotated[Session, Depends(get_db)],
    user: Annotated[SysUser, Depends(require_menu("suggestion:handle"))],
):
    return success_response(
        admin_ledger_service.update_suggestion_status(
            db,
            user,
            suggestion_id,
            status=body.status,
            handle_remark=body.handle_remark,
        )
    )


@router.get("/hotline")
def get_hotline(
    db: Annotated[Session, Depends(get_db)],
    _user: Annotated[SysUser, Depends(get_current_user)],
):
    return success_response(config_service.get_hotline_payload(db))


@router.put("/hotline")
def put_hotline(
    body: HotlineUpdateBody,
    db: Annotated[Session, Depends(get_db)],
    user: Annotated[SysUser, Depends(require_menu("hotline:write"))],
):
    return success_response(
        config_service.update_hotline(
            db,
            user,
            phone=body.phone,
            email=body.email,
            address=body.address,
            work_time=body.work_time,
        )
    )


@router.get("/qrcode")
def get_qrcode(
    db: Annotated[Session, Depends(get_db)],
    _user: Annotated[SysUser, Depends(get_current_user)],
):
    return success_response(config_service.get_qrcode_payload(db))


@router.put("/qrcode")
def put_qrcode(
    body: QrcodeUpdateBody,
    db: Annotated[Session, Depends(get_db)],
    user: Annotated[SysUser, Depends(require_menu("qrcode:write"))],
):
    return success_response(
        config_service.update_qrcode(
            db,
            user,
            land_url=body.land_url,
            title=body.title,
            image_url=body.image_url,
        )
    )


@router.post("/attachment/upload")
async def admin_upload(
    db: Annotated[Session, Depends(get_db)],
    _user: Annotated[SysUser, Depends(get_current_user)],
    file: UploadFile = File(...),
    folder: Optional[str] = Form(""),
):
    if not file or not file.filename:
        raise AppHTTPError(400, "请选择文件")
    content = await file.read()
    try:
        stored = storage.upload_file(content, file.filename, folder or "")
    except StorageError as exc:
        raise AppHTTPError(400, exc.message) from exc
    row = attachment_crud.create_attachment(
        db,
        file_name=file.filename,
        object_key=stored["objectKey"],
        url=stored["url"],
        size=len(content),
    )
    return success_response(
        {
            "id": row.id,
            "url": row.url,
            "fileName": row.file_name,
            "size": row.size,
        }
    )


@router.get("/users")
def list_users(
    db: Annotated[Session, Depends(get_db)],
    _user: Annotated[SysUser, Depends(require_menu("user:manage"))],
):
    rows = user_crud.list_users(db)
    return success_response([_user_to_dict(db, u) for u in rows])


@router.post("/users")
def create_user(
    body: UserCreateBody,
    db: Annotated[Session, Depends(get_db)],
    _user: Annotated[SysUser, Depends(require_menu("user:manage"))],
):
    username = (body.username or "").strip()
    password = body.password or ""
    display_name = (body.display_name or "").strip()
    if not username or not password or not display_name:
        raise AppHTTPError(400, "用户名、密码、显示名不能为空")
    if user_crud.get_user_by_username(db, username):
        raise AppHTTPError(400, "用户名已存在")
    role = user_crud.get_role_by_id(db, body.role_id)
    if not role:
        raise AppHTTPError(400, "角色不存在")
    row = user_crud.create_user(
        db,
        username=username,
        password_hash=hash_password(password),
        display_name=display_name,
        role_id=role.id,
    )
    return success_response(_user_to_dict(db, row))


@router.put("/users/{user_id}")
def update_user(
    user_id: str,
    body: UserUpdateBody,
    db: Annotated[Session, Depends(get_db)],
    _user: Annotated[SysUser, Depends(require_menu("user:manage"))],
):
    row = user_crud.get_user_by_id(db, user_id)
    if not row:
        raise AppHTTPError(400, "用户不存在")
    role_id = body.role_id
    if role_id is not None:
        role = user_crud.get_role_by_id(db, role_id)
        if not role:
            raise AppHTTPError(400, "角色不存在")
    status = body.status
    if status is not None and status not in ("0", "1"):
        raise AppHTTPError(400, "状态无效")
    display_name = body.display_name
    if display_name is not None:
        display_name = display_name.strip()
        if not display_name:
            raise AppHTTPError(400, "显示名不能为空")
    updated = user_crud.update_user(
        db,
        row,
        display_name=display_name,
        role_id=role_id,
        status=status,
    )
    return success_response(_user_to_dict(db, updated))


@router.put("/users/{user_id}/reset-password")
def reset_user_password(
    user_id: str,
    body: ResetPasswordBody,
    db: Annotated[Session, Depends(get_db)],
    _user: Annotated[SysUser, Depends(require_menu("user:manage"))],
):
    row = user_crud.get_user_by_id(db, user_id)
    if not row:
        raise AppHTTPError(400, "用户不存在")
    password = body.password or ""
    if not password:
        raise AppHTTPError(400, "密码不能为空")
    user_crud.update_user(db, row, password_hash=hash_password(password))
    return success_response({"id": user_id})


@router.get("/roles")
def list_roles(
    db: Annotated[Session, Depends(get_db)],
    _user: Annotated[SysUser, Depends(get_current_user)],
):
    return success_response([_role_to_dict(r) for r in role_crud.list_roles(db)])


@router.get("/menus/tree")
def menus_tree(
    db: Annotated[Session, Depends(get_db)],
    _user: Annotated[SysUser, Depends(get_current_user)],
):
    menus = menu_crud.list_all_menus(db)
    return success_response(auth_service.build_menu_tree(menus))


@router.get("/roles/{role_id}/menus")
def get_role_menus(
    role_id: str,
    db: Annotated[Session, Depends(get_db)],
    _user: Annotated[SysUser, Depends(get_current_user)],
):
    role = role_crud.get_role_by_id(db, role_id)
    if not role:
        raise AppHTTPError(400, "角色不存在")
    return success_response({"menuIds": role_crud.list_menu_ids_for_role(db, role_id)})


@router.put("/roles/{role_id}/menus")
def put_role_menus(
    role_id: str,
    body: RoleMenusBody,
    db: Annotated[Session, Depends(get_db)],
    _user: Annotated[SysUser, Depends(require_menu("role:grant"))],
):
    role = role_crud.get_role_by_id(db, role_id)
    if not role:
        raise AppHTTPError(400, "角色不存在")
    menu_ids: List[str] = list(body.menu_ids or [])
    if menu_ids:
        valid_ids = {m.id for m in menu_crud.list_all_menus(db)}
        invalid = [mid for mid in menu_ids if mid not in valid_ids]
        if invalid:
            raise AppHTTPError(400, "存在无效菜单")
    role_crud.replace_role_menus(db, role_id, menu_ids)
    return success_response({"menuIds": menu_ids})


@router.get("/oper-logs")
def list_oper_logs(
    db: Annotated[Session, Depends(get_db)],
    _user: Annotated[SysUser, Depends(require_menu("audit"))],
    page: int = Query(1, ge=1),
    pageSize: int = Query(10, ge=1, le=100),
):
    items, total = oper_log_crud.list_oper_logs(db, page=page, page_size=pageSize)
    return success_response(
        {
            "items": [_oper_log_to_dict(r) for r in items],
            "total": total,
            "page": page,
            "pageSize": pageSize,
        }
    )