ajyy/backend/app/api/admin.py

"""管理端 API：工作台、台账办理、配置、用户/角色、审计、附件上传。"""
from __future__ import annotations

import logging
from typing import Annotated, Any, Dict, List, Optional

from fastapi import APIRouter, Depends, File, Form, Query, UploadFile
from fastapi.responses import StreamingResponse
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, _normalize_folder
from app.core.upload_validate import validate_image_upload, validate_upload
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

logger = logging.getLogger(__name__)

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,
            phones=body.phones,
            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, "请选择文件")
    try:
        folder_norm = _normalize_folder(folder or "")
    except StorageError as exc:
        raise AppHTTPError(400, exc.message) from exc
    content = await file.read()
    if folder_norm == "qrcode":
        validate_image_upload(file.filename, content)
    else:
        validate_upload(file.filename, content)
    try:
        stored = storage.upload_file(content, file.filename, folder_norm)
    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("/attachment/file")
def download_attachment_file(
    user: Annotated[SysUser, Depends(get_current_user)],
    url: str = Query(...),
):
    try:
        stream, content_type, filename, inline = storage.open_file(url)
    except StorageError as exc:
        raise AppHTTPError(400, exc.message) from exc
    logger.info("attachment_proxy user_id=%s filename=%s", user.id, filename)
    disposition = "inline" if inline else "attachment"
    # ASCII-safe filename fallback for Content-Disposition
    safe_name = filename.encode("ascii", "ignore").decode() or "download"
    headers = {
        "X-Content-Type-Options": "nosniff",
        "Content-Disposition": f'{disposition}; filename="{safe_name}"',
    }
    return StreamingResponse(stream, media_type=content_type, headers=headers)

@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(
        {
            "list": [_oper_log_to_dict(r) for r in items],
            "total": total,
            "page": page,
            "pageSize": pageSize,
        }
    )