shanxia/crm/py-backend/app/routers/auth.py

import json
from datetime import datetime, timedelta
from typing import Optional, List

from fastapi import APIRouter, Depends, HTTPException, status, Request
from pydantic import BaseModel, ConfigDict, Field
from pydantic.alias_generators import to_camel
from sqlalchemy.orm import Session

from app.database import get_db
from app.models.system import SysUser, SysTenant, SysUserRole, SysRole, SysRoleMenu, SysMenu
from app.schemas import R
from app.utils.security import verify_password, hash_password
from app.utils import next_id
from app.middleware import create_access_token, get_current_user
from app.utils.redis_client import get_redis

router = APIRouter()


class LoginBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)

    tenant_id: str = "000000"
    username: str
    password: str
    client_id: str = "e5cd7e4891bf95d1d19206ce24a7b32e"
    grant_type: str = "password"


class LoginVo(BaseModel):
    model_config = ConfigDict(populate_by_name=True, alias_generator=to_camel)

    access_token: str
    refresh_token: str = ""
    expire_in: int
    refresh_expire_in: int = 0
    client_id: str = ""
    scope: str = ""
    openid: str = ""


@router.post("/login")
def login(body: LoginBody, db: Session = Depends(get_db)):
    user = (
        db.query(SysUser)
        .filter(
            SysUser.user_name == body.username,
            SysUser.del_flag == "0",
        )
        .first()
    )
    if not user:
        return R.fail("用户不存在或密码错误")
    if not verify_password(body.password, user.password):
        return R.fail("用户不存在或密码错误")
    if user.status != "0":
        return R.fail("用户已被停用")

    tenant_id = body.tenant_id or "000000"
    permissions = _get_user_permissions(db, user.user_id, tenant_id)
    roles = _get_user_roles(db, user.user_id, tenant_id)

    token_data = {
        "login_id": str(user.user_id),
        "tenant_id": tenant_id,
        "user_id": user.user_id,
        "user_name": user.user_name,
        "nick_name": user.nick_name,
        "dept_id": user.dept_id,
    }
    expire_minutes = 1440
    access_token = create_access_token(
        token_data,
        timedelta(minutes=expire_minutes),
    )
    refresh_token = create_access_token(
        token_data,
        timedelta(minutes=expire_minutes * 2),
    )

    redis = get_redis()
    session_key = f"login:session:{user.user_id}:{tenant_id}"
    session_data = {
        "user_id": user.user_id,
        "user_name": user.user_name,
        "nick_name": user.nick_name,
        "dept_id": user.dept_id or 0,
        "tenant_id": tenant_id,
        "permissions": list(permissions),
        "roles": list(roles),
    }
    redis.setex(session_key, expire_minutes * 60, json.dumps(session_data, ensure_ascii=False))

    user.login_ip = "127.0.0.1"
    user.login_date = datetime.now()
    db.commit()

    return R.ok(
        data={
            "access_token": access_token,
            "refresh_token": refresh_token,
            "expire_in": expire_minutes * 60,
            "refresh_expire_in": 0,
            "client_id": body.client_id or "",
            "scope": "",
            "openid": "",
        }
    )


@router.post("/logout")
def logout(current_user: dict = Depends(get_current_user)):
    user_id = current_user.get("user_id")
    tenant_id = current_user.get("tenant_id", "000000")
    redis = get_redis()
    session_key = f"login:session:{user_id}:{tenant_id}"
    redis.delete(session_key)
    return R.ok(msg="退出成功")


@router.get("/tenant/list")
def tenant_list(db: Session = Depends(get_db)):
    tenants = db.query(SysTenant).filter(SysTenant.status == "0", SysTenant.del_flag == "0").all()
    return R.ok(data=[
        {
            "tenantId": t.tenant_id,
            "companyName": t.company_name,
            "domain": t.domain,
        }
        for t in tenants
    ])


def _get_user_permissions(db: Session, user_id: int, tenant_id: str) -> set:
    role_ids = [r.role_id for r in db.query(SysUserRole).filter(SysUserRole.user_id == user_id).all()]
    if not role_ids:
        return set()
    menu_ids = [m.menu_id for m in db.query(SysRoleMenu).filter(SysRoleMenu.role_id.in_(role_ids)).all()]
    if not menu_ids:
        return set()
    menus = db.query(SysMenu).filter(SysMenu.menu_id.in_(menu_ids), SysMenu.status == "0").all()
    return {m.perms for m in menus if m.perms}


def _get_user_roles(db: Session, user_id: int, tenant_id: str) -> set:
    role_ids = [r.role_id for r in db.query(SysUserRole).filter(SysUserRole.user_id == user_id).all()]
    if not role_ids:
        return set()
    roles = db.query(SysRole).filter(SysRole.role_id.in_(role_ids), SysRole.del_flag == "0").all()
    return {r.role_key for r in roles}