zkxw/backend/app/api/public.py

"""公共接口（Web 登录、刷新 Token、登出；用户模型对齐 docs/sql/zkxw.sql 基线）"""
import base64
import os
from datetime import datetime
from typing import Optional

from fastapi import APIRouter, Depends, HTTPException, Query, Request, UploadFile, File
from sqlalchemy.exc import OperationalError
from sqlalchemy.orm import Session

from app.core.database import get_db
from app.core.security import verify_password
from app.core.auth import (
    create_access_token,
    create_refresh_token,
    decode_token,
    get_current_user_by_token,
    extend_session,
    is_refresh_token_valid,
    is_session_valid,
    revoke_refresh_token,
    revoke_session,
    store_refresh_token,
    store_session,
)
from app.core.response import success_response, error_response
from app.core.config import settings
from app.core.storage import storage
from app.core.avatar_url import extract_avatar_storage_key
from app.models.models import SysUser
from app.crud.user import user_crud
from app.core.utils import generate_uuid
from app.core import login_crypto
from app.services.web_auth_roles import (
    get_user_role_names,
    iter_user_menu_perm_codes,
    resolve_web_role_for_user,
    user_has_active_roles,
)
from app.schemas.auth import ProfileUpdateRequest
from app.crud.user import update_user, change_password
from app.services.issue_user_tokens import build_user_info_payload
from app.services import wx_qr_session
from app.services.wechat_mini_auth import get_unlimited_wxacode, WechatMiniAuthError

router = APIRouter(prefix="", tags=["公共接口"])


def _build_current_user_payload(current_user: dict) -> dict:
    """组装 current-user 接口响应。"""
    au = current_user.get("avatar_url")
    avatar_url = str(au).strip() if au else None
    return {
        "id": current_user.get("user_id"),
        "username": current_user.get("username"),
        "realName": current_user.get("real_name"),
        "phone": current_user.get("phone"),
        "email": current_user.get("email"),
        "sex": current_user.get("sex") or "0",
        "remark": current_user.get("remark"),
        "role": current_user.get("role"),
        "roleNames": current_user.get("role_names") or current_user.get("roleNames") or [],
        "isSuperAdmin": current_user.get("role") == "super_admin",
        "regionId": current_user.get("region_id"),
        "regionName": current_user.get("region_name") or current_user.get("dept_name"),
        "deptName": current_user.get("dept_name") or current_user.get("region_name"),
        "postNames": current_user.get("post_names") or [],
        "regionIds": [],
        "isActive": current_user.get("is_active"),
        "avatarUrl": avatar_url if avatar_url else None,
    }


async def _resolve_current_user_from_request(request: Request, db: Session) -> dict:
    authorization = request.headers.get("Authorization", "")
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="未授权")
    token = authorization.replace("Bearer ", "")
    return await get_current_user_by_token(token=token, db=db)


def _jwt_user_payload(user: SysUser, web_role: str) -> dict:
    return {
        "sub": user.id,
        "username": user.user_no,
        "role": web_role,
        "region_id": user.dept_id,
        "real_name": user.nick_name,
    }


@router.get("/web/auth/login-rsa-public-key", response_model=dict)
async def web_login_rsa_public_key():
    """Web 端登录用 RSA 公钥（PKCS#1 v1.5，与 JSEncrypt 一致）"""
    login_crypto.ensure_initialized()
    return success_response({"publicKeyPem": login_crypto.get_public_pem()})


@router.post("/web/auth/login", response_model=dict)
async def web_login(
    request: Request,
    body: dict,
    db: Session = Depends(get_db),
):
    """Web 端登录：账号为 sys_user.user_no，密码为 sys_user.password（bcrypt）。"""
    login_crypto.ensure_initialized()
    user_no, password, cred_err = login_crypto.extract_login_credentials(body)
    if cred_err:
        return error_response(400, cred_err)

    user = user_crud.get_by_user_no(db, user_no)

    if not user:
        return error_response(401, "用户名或密码错误")

    pwd_hash = getattr(user, "password", None) or ""
    if not verify_password(password, pwd_hash):
        return error_response(401, "用户名或密码错误")

    if user.status != "1":
        return error_response(403, "账号已被停用，请联系管理员")

    tenant_no = (getattr(user, "tenant_no", None) or "").strip() or None
    if not user_has_active_roles(db, user.id, tenant_no=tenant_no):
        return error_response(403, "账号未分配角色，请联系管理员")

    web_role = resolve_web_role_for_user(
        db, user.id, tenant_no=tenant_no
    )
    role_names = get_user_role_names(db, user.id, tenant_no=tenant_no)

    session_id = generate_uuid()
    access_jti = generate_uuid()
    refresh_jti = generate_uuid()

    access_token = create_access_token(
        data=_jwt_user_payload(user, web_role),
        session_id=session_id,
        token_jti=access_jti,
    )
    refresh_token = create_refresh_token(
        data={
            "sub": user.id,
            "username": user.user_no,
            "role": web_role,
        },
        session_id=session_id,
        refresh_jti=refresh_jti,
    )

    await store_session(session_id=session_id, user_id=user.id)
    await store_refresh_token(refresh_jti=refresh_jti, user_id=user.id)

    return success_response(
        {
            "token": access_token,
            "refreshToken": refresh_token,
            "expiresIn": settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60,
            "userInfo": build_user_info_payload(user, web_role, role_names, db=db),
        }
    )


@router.post("/web/auth/wx-qr/create", response_model=dict)
async def wx_qr_create():
    session_id, short = wx_qr_session.create_session()
    page = (settings.WX_QR_PAGE or "pages/login/web-qr/index").strip()
    try:
        raw = get_unlimited_wxacode(
            scene=short,
            page=page,
            env_version=settings.WX_QR_ENV_VERSION,
        )
    except WechatMiniAuthError as exc:
        wx_qr_session.invalidate(session_id)
        return error_response(400, str(exc))
    b64 = base64.b64encode(raw).decode("ascii")
    mime = "image/jpeg" if raw[:2] == b"\xff\xd8" else "image/png"
    return success_response(
        {
            "sessionId": session_id,
            "expiresIn": settings.WX_QR_SESSION_TTL_SEC,
            "qrImageBase64": f"data:{mime};base64,{b64}",
        }
    )


@router.get("/web/auth/wx-qr/status", response_model=dict)
async def wx_qr_status(sessionId: str = ""):
    sid = (sessionId or "").strip()
    data = wx_qr_session.get_session(sid)
    if not data:
        return success_response({"status": "expired", "message": "二维码已过期，请刷新"})
    status = data.get("status") or "expired"
    if status == "success":
        tokens = wx_qr_session.consume_success(sid)
        if not tokens:
            return success_response({"status": "expired", "message": "登录凭证已失效，请刷新二维码"})
        return success_response({"status": "success", **tokens})
    if status == "failed":
        return success_response({"status": "failed", "message": data.get("message") or "登录失败"})
    if status in ("expired", "consumed"):
        return success_response({"status": "expired", "message": "二维码已过期，请刷新"})
    return success_response({"status": status})


@router.post("/web/auth/refresh-token", response_model=dict)
async def refresh_token(body: dict, db: Session = Depends(get_db)):
    """刷新 Token"""
    refresh_token_val = body.get("refreshToken")
    if not refresh_token_val:
        return error_response(400, "refreshToken不能为空")

    try:
        payload = decode_token(refresh_token_val)
        if payload.get("type") != "refresh":
            return error_response(401, "无效的refreshToken")

        session_id = payload.get("sid")
        refresh_jti = payload.get("jti")

        if not await is_session_valid(session_id=session_id):
            return error_response(401, "refreshToken已失效，请重新登录")
        if not await is_refresh_token_valid(refresh_jti=refresh_jti):
            return error_response(401, "无效的refreshToken")

        user_id = payload.get("sub")
        user = user_crud.get(db, user_id)

        if not user or user.del_flag != 0 or user.status != "1":
            return error_response(401, "用户不存在或已禁用")

        tenant_no = (getattr(user, "tenant_no", None) or "").strip() or None
        web_role = resolve_web_role_for_user(
            db, user.id, tenant_no=tenant_no
        )

        await revoke_refresh_token(refresh_jti=refresh_jti)
        await extend_session(session_id=session_id)

        new_access_jti = generate_uuid()
        new_refresh_jti = generate_uuid()

        new_access_token = create_access_token(
            data=_jwt_user_payload(user, web_role),
            session_id=session_id,
            token_jti=new_access_jti,
        )
        new_refresh_token = create_refresh_token(
            data={
                "sub": user.id,
                "username": user.user_no,
                "role": web_role,
            },
            session_id=session_id,
            refresh_jti=new_refresh_jti,
        )

        await store_refresh_token(refresh_jti=new_refresh_jti, user_id=user.id)

        return success_response(
            {
                "token": new_access_token,
                "refreshToken": new_refresh_token,
                "expiresIn": settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60,
            }
        )
    except HTTPException:
        return error_response(401, "refreshToken已过期")


@router.post("/web/auth/logout", response_model=dict)
async def logout(request: Request):
    """登出：吊销 Redis 会话（若已配置 REDIS_HOST）。"""
    authorization = request.headers.get("Authorization", "")
    if not authorization.startswith("Bearer "):
        return error_response(401, "未授权")

    token = authorization.replace("Bearer ", "")
    try:
        payload = decode_token(token)
        session_id = payload.get("sid")
        if session_id:
            await revoke_session(session_id=session_id)
    except HTTPException:
        return error_response(401, "Token无效或已过期")

    return success_response({"message": "登出成功"})


@router.get("/web/auth/current-user", response_model=dict)
async def get_current_user_profile(request: Request, db: Session = Depends(get_db)):
    """获取当前用户信息（与 sys_user 字段对应）。"""
    try:
        current_user = await _resolve_current_user_from_request(request, db)
        return success_response(_build_current_user_payload(current_user))
    except HTTPException:
        return error_response(401, "Token无效或已过期")


@router.put("/web/auth/profile", response_model=dict)
async def update_current_user_profile(
    request: Request,
    body: ProfileUpdateRequest,
    db: Session = Depends(get_db),
):
    """当前用户更新个人信息（手机号、邮箱、性别、备注）。"""
    try:
        current_user = await _resolve_current_user_from_request(request, db)
        user_id = current_user.get("user_id")
        if not user_id:
            return error_response(401, "未授权")

        payload = body.model_dump(exclude_unset=True, by_alias=False)
        update_user(
            db,
            user_id,
            phone_no=payload.get("phone_no"),
            email=payload.get("email"),
            sex=payload.get("sex"),
            remark=payload.get("remark"),
            avatar=payload.get("avatar"),
            user_id_updater=user_id,
        )

        user = user_crud.get(db, user_id)
        if not user:
            return error_response(404, "用户不存在")
        tenant_no = (getattr(user, "tenant_no", None) or "").strip() or None
        web_role = resolve_web_role_for_user(
            db, user.id, tenant_no=tenant_no
        )
        role_names = get_user_role_names(
            db, user.id, tenant_no=tenant_no
        )
        return success_response(build_user_info_payload(user, web_role, role_names, db=db))
    except HTTPException:
        return error_response(401, "Token无效或已过期")
    except ValueError as exc:
        return error_response(400, str(exc))


@router.post("/web/auth/change-password", response_model=dict)
async def change_current_user_password(
    request: Request,
    body: dict,
    db: Session = Depends(get_db),
):
    """当前用户修改密码。"""
    try:
        current_user = await _resolve_current_user_from_request(request, db)
        user_id = current_user.get("user_id")
        if not user_id:
            return error_response(401, "未授权")

        old_password = (body.get("oldPassword") or body.get("old_password") or "").strip()
        new_password = (body.get("newPassword") or body.get("new_password") or "").strip()
        if not old_password:
            return error_response(400, "请输入原密码")
        if len(new_password) < 6:
            return error_response(400, "新密码长度不能少于6位")

        change_password(db, user_id, old_password, new_password)
        return success_response({"message": "密码修改成功"})
    except HTTPException:
        return error_response(401, "Token无效或已过期")
    except ValueError as exc:
        return error_response(400, str(exc))


_AVATAR_MAX_BYTES = 2 * 1024 * 1024
_AVATAR_ALLOWED_EXT = {".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp"}


@router.post("/web/auth/upload-avatar", response_model=dict)
async def upload_current_user_avatar(
    request: Request,
    file: UploadFile = File(...),
    db: Session = Depends(get_db),
):
    """当前用户上传头像（仅图片，最大 2MB）。"""
    try:
        await _resolve_current_user_from_request(request, db)
    except HTTPException:
        return error_response(401, "Token无效或已过期")

    if not file or not file.filename:
        return error_response(400, "请选择图片")

    ext = os.path.splitext(file.filename)[1].lower()
    if ext not in _AVATAR_ALLOWED_EXT:
        return error_response(400, "仅支持 JPG、PNG、GIF、WEBP、BMP 格式")

    content_type = (file.content_type or "").lower()
    if content_type and not content_type.startswith("image/"):
        return error_response(400, "仅支持上传图片文件")

    try:
        file_content = await file.read()
        if not file_content:
            return error_response(400, "文件内容为空")
        if len(file_content) > _AVATAR_MAX_BYTES:
            return error_response(400, "头像大小不能超过 2MB")

        url = storage.upload_file(file_content, file.filename, "avatar")
        stored_url = extract_avatar_storage_key(url)
        return success_response({"url": stored_url})
    except Exception as exc:
        return error_response(500, f"上传失败: {str(exc)}")


@router.get("/web/auth/menu-codes", response_model=dict)
async def get_web_auth_menu_codes(request: Request, db: Session = Depends(get_db)):
    """获取当前用户菜单权限标识（sys_menu.perms，经角色继承）。"""
    authorization = request.headers.get("Authorization", "")
    if not authorization.startswith("Bearer "):
        return error_response(401, "未授权")

    token = authorization.replace("Bearer ", "")
    try:
        current_user = await get_current_user_by_token(token=token, db=db)
        role = current_user.get("role")
        if role == "super_admin":
            return success_response(
                {
                    "isSuperAdmin": True,
                    "all": True,
                    "menuCodes": [],
                }
            )

        try:
            codes = iter_user_menu_perm_codes(
                db,
                current_user.get("user_id") or "",
                tenant_no=current_user.get("tenant_no"),
            )
            menu_codes = sorted(codes)
        except OperationalError:
            menu_codes = []

        return success_response(
            {
                "isSuperAdmin": False,
                "all": False,
                "menuCodes": menu_codes,
            }
        )
    except HTTPException:
        return error_response(401, "Token无效或已过期")


@router.get("/cn-holiday/range", response_model=dict)
def cn_holiday_range(
    start: str = Query(..., description="YYYY-MM-DD"),
    end: str = Query(..., description="YYYY-MM-DD"),
    db: Session = Depends(get_db),
):
    """查询闭区间内每日节假日/休息日标志（只读本地缓存表）。"""
    try:
        start_d = datetime.strptime(start.strip(), "%Y-%m-%d").date()
        end_d = datetime.strptime(end.strip(), "%Y-%m-%d").date()
    except ValueError:
        raise HTTPException(status_code=400, detail="start/end 须为 YYYY-MM-DD")
    if end_d < start_d:
        raise HTTPException(status_code=400, detail="end 不能早于 start")
    from app.services.cn_holiday import list_day_flags

    days = list_day_flags(db, start_d, end_d)
    return success_response({"days": days})