zlwuandCursor
10c3c5470c
Bundle pending backend/web-admin feature updates, data-migration scripts, and related docs so the V1.5 branch on the Tencent Cloud Gogs matches the current workspace (excluding local secrets). Co-authored-by: Cursor <cursoragent@cursor.com>
1435 lines
49 KiB
Python
1435 lines
49 KiB
Python
"""运营端管理接口（租户、角色下拉等）"""
|
||
import os
|
||
import uuid
|
||
from typing import Any, List, Optional
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, Query, UploadFile, File, Form
|
||
from pydantic import BaseModel, ConfigDict, Field
|
||
from sqlalchemy import and_, or_
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.config import settings
|
||
from app.core.data_scope import apply_data_scope, assert_row_in_data_scope
|
||
from app.core.database import get_db
|
||
from app.core.menu_guard import require_admin_or_super_admin, require_menu_code
|
||
from app.core.response import success_response, error_response
|
||
from app.core.attachment_biz import resolve_upload_folder
|
||
from app.core.storage import storage
|
||
from app.crud import tenant as tenant_svc
|
||
from app.crud import dict_type as dict_type_svc
|
||
from app.crud import dict_data as dict_data_svc
|
||
from app.crud import dept as dept_svc
|
||
from app.crud import site as site_svc
|
||
from app.crud import post as post_svc
|
||
from app.crud import menu as menu_svc
|
||
from app.crud import role as role_svc
|
||
from app.crud import user as user_svc
|
||
from app.crud import role as role_svc
|
||
from app.crud import user as user_svc
|
||
from app.crud import oss as oss_svc
|
||
from app.models.models import SysDept, SysDictData, SysRole, SysUser
|
||
from app.services.web_auth_roles import _map_role_name_to_web
|
||
|
||
router = APIRouter(prefix="/web/admin", tags=["运营端管理"])
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 角色下拉（用户弹窗等）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@router.get("/roles")
|
||
def list_roles_for_select(
|
||
_user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
"""返回运营端角色下拉选项（label=库中角色名，value=Web 角色码）。"""
|
||
rows: List[SysRole] = (
|
||
db.query(SysRole)
|
||
.filter(SysRole.del_flag == 0, SysRole.status == "1")
|
||
.order_by(SysRole.create_time.desc())
|
||
.all()
|
||
)
|
||
seen: set[str] = set()
|
||
out: List[dict] = []
|
||
for r in rows:
|
||
name = (r.role_name or "").strip()
|
||
if not name:
|
||
continue
|
||
code = _map_role_name_to_web(name)
|
||
key = f"{code}:{name}"
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
out.append({"label": name, "value": code})
|
||
if not out:
|
||
out = [
|
||
{"label": "超级管理员", "value": "super_admin"},
|
||
{"label": "运营管理员", "value": "admin"},
|
||
]
|
||
return success_response(out)
|
||
|
||
|
||
@router.get("/users/options")
|
||
def list_users_for_select(
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
"""返回用户下拉选项（label=昵称，value=用户ID，附带 nickName/userNo）。
|
||
|
||
供申请人/审批人等选择场景使用，返回本租户全部有效用户，
|
||
不做业务数据范围过滤（否则「仅本人」角色会按 create_by 滤空，连自己都选不到）。
|
||
"""
|
||
tenant_no = _get_tenant_no(db, user)
|
||
q = db.query(SysUser).filter(SysUser.del_flag == 0, SysUser.status == "1")
|
||
q = q.filter(SysUser.tenant_no == tenant_no)
|
||
rows = q.order_by(SysUser.create_time.desc()).all()
|
||
out = [
|
||
{
|
||
"label": (r.nick_name or r.user_no or ""),
|
||
"value": r.id,
|
||
"nickName": r.nick_name or "",
|
||
"userNo": r.user_no or "",
|
||
}
|
||
for r in rows
|
||
]
|
||
return success_response(out)
|
||
|
||
|
||
@router.get("/roles/options")
|
||
def list_roles_for_designer(
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
"""返回角色下拉选项（label=角色名，value=角色ID UUID），供流程设计器审批人选择使用。
|
||
与 /roles 不同，此端点返回角色 UUID 作为 value（引擎解析审批人需要 UUID）。
|
||
角色为租户级主数据选项，不做 create_by 数据范围过滤。
|
||
"""
|
||
tenant_no = _get_tenant_no(db, user)
|
||
q = db.query(SysRole).filter(SysRole.del_flag == 0, SysRole.status == "1")
|
||
q = q.filter(SysRole.tenant_no == tenant_no)
|
||
rows = q.order_by(SysRole.create_time.desc()).all()
|
||
out = [{"label": (r.role_name or "").strip(), "value": r.id} for r in rows if (r.role_name or "").strip()]
|
||
return success_response(out)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 租户管理
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TenantCreateBody(BaseModel):
|
||
model_config = ConfigDict(populate_by_name=True)
|
||
|
||
tenant_no: str = Field(..., alias="tenantNo")
|
||
tenant_name: str = Field(..., alias="tenantName")
|
||
contact_name: Optional[str] = Field(None, alias="contactName")
|
||
contact_phone: Optional[str] = Field(None, alias="contactPhone")
|
||
order_num: Optional[int] = Field(0, alias="orderNum")
|
||
remark: Optional[str] = None
|
||
|
||
|
||
class TenantUpdateBody(BaseModel):
|
||
model_config = ConfigDict(populate_by_name=True)
|
||
|
||
tenant_no: Optional[str] = Field(None, alias="tenantNo")
|
||
tenant_name: Optional[str] = Field(None, alias="tenantName")
|
||
contact_name: Optional[str] = Field(None, alias="contactName")
|
||
contact_phone: Optional[str] = Field(None, alias="contactPhone")
|
||
order_num: Optional[int] = Field(None, alias="orderNum")
|
||
remark: Optional[str] = None
|
||
|
||
|
||
@router.get("/tenants")
|
||
def list_tenants(
|
||
page: int = Query(1, ge=1),
|
||
pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
|
||
keyword: Optional[str] = None,
|
||
_user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
items, total = tenant_svc.list_tenants(db, page=page, page_size=pageSize, keyword=keyword, user_info=_user)
|
||
total_pages = tenant_svc.total_pages(total, pageSize)
|
||
return success_response(
|
||
{
|
||
"list": items,
|
||
"total": total,
|
||
"page": page,
|
||
"pageSize": pageSize,
|
||
"totalPages": total_pages,
|
||
}
|
||
)
|
||
|
||
|
||
@router.post("/tenants")
|
||
def create_tenant(
|
||
body: TenantCreateBody,
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
obj = tenant_svc.create_tenant(
|
||
db,
|
||
tenant_no=body.tenant_no,
|
||
tenant_name=body.tenant_name,
|
||
contact_name=body.contact_name,
|
||
contact_phone=body.contact_phone,
|
||
order_num=body.order_num,
|
||
remark=body.remark,
|
||
user_id=user.get("user_id"),
|
||
)
|
||
return success_response(tenant_svc._row_to_dict(obj), message="创建成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
@router.put("/tenants/{tenant_id}")
|
||
def update_tenant(
|
||
tenant_id: str,
|
||
body: TenantUpdateBody,
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
|
||
obj = tenant_svc.update_tenant(
|
||
db,
|
||
tenant_id,
|
||
tenant_no=payload.get("tenant_no"),
|
||
tenant_name=payload.get("tenant_name"),
|
||
contact_name=payload.get("contact_name"),
|
||
contact_phone=payload.get("contact_phone"),
|
||
order_num=payload.get("order_num"),
|
||
remark=payload.get("remark"),
|
||
user_id=user.get("user_id"),
|
||
)
|
||
return success_response(tenant_svc._row_to_dict(obj), message="更新成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
@router.delete("/tenants/{tenant_id}")
|
||
def delete_tenant(
|
||
tenant_id: str,
|
||
_user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
tenant_svc.soft_delete_tenant(db, tenant_id, user_info=_user)
|
||
return success_response(None, message="删除成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helper: resolve tenant_no from user context
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _get_tenant_no(db: Session, user: dict) -> str:
|
||
user_id = user.get("user_id")
|
||
if not user_id:
|
||
return "shanxia2.0"
|
||
sys_user = db.query(SysUser).filter(SysUser.id == user_id).first()
|
||
return (sys_user.tenant_no if sys_user else None) or "shanxia2.0"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 用户管理
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class UserCreateBody(BaseModel):
|
||
model_config = ConfigDict(populate_by_name=True)
|
||
user_no: str = Field(..., alias="userNo")
|
||
nick_name: str = Field(..., alias="nickName")
|
||
password: str = Field(..., alias="password")
|
||
dept_id: Optional[str] = Field(None, alias="deptId")
|
||
email: Optional[str] = Field("", alias="email")
|
||
phone_no: Optional[str] = Field("", alias="phoneNo")
|
||
sex: Optional[str] = Field("0", alias="sex")
|
||
avatar: Optional[str] = Field(None, alias="avatar")
|
||
status: Optional[str] = Field("1", alias="status")
|
||
order_num: Optional[int] = Field(None, alias="orderNum")
|
||
remark: Optional[str] = None
|
||
role_ids: Optional[List[str]] = Field(None, alias="roleIds")
|
||
post_ids: Optional[List[str]] = Field(None, alias="postIds")
|
||
|
||
|
||
class UserUpdateBody(BaseModel):
|
||
model_config = ConfigDict(populate_by_name=True)
|
||
nick_name: Optional[str] = Field(None, alias="nickName")
|
||
dept_id: Optional[str] = Field(None, alias="deptId")
|
||
email: Optional[str] = Field(None, alias="email")
|
||
phone_no: Optional[str] = Field(None, alias="phoneNo")
|
||
sex: Optional[str] = Field(None, alias="sex")
|
||
avatar: Optional[str] = Field(None, alias="avatar")
|
||
status: Optional[str] = Field(None, alias="status")
|
||
order_num: Optional[int] = Field(None, alias="orderNum")
|
||
remark: Optional[str] = None
|
||
role_ids: Optional[List[str]] = Field(None, alias="roleIds")
|
||
post_ids: Optional[List[str]] = Field(None, alias="postIds")
|
||
|
||
|
||
@router.get("/users")
|
||
def list_users(
|
||
page: int = Query(1, ge=1),
|
||
pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
|
||
keyword: Optional[str] = None,
|
||
deptId: Optional[str] = Query(None, alias="deptId"),
|
||
status: Optional[str] = None,
|
||
user: dict = Depends(require_menu_code("system:user:list")),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
tenant_no = _get_tenant_no(db, user)
|
||
query = db.query(SysUser).filter(
|
||
SysUser.del_flag == 0,
|
||
SysUser.tenant_no == tenant_no,
|
||
)
|
||
kw = (keyword or "").strip()
|
||
if kw:
|
||
like = f"%{kw}%"
|
||
query = query.filter(
|
||
or_(
|
||
SysUser.user_no.like(like),
|
||
SysUser.nick_name.like(like),
|
||
SysUser.phone_no.like(like),
|
||
)
|
||
)
|
||
if deptId:
|
||
query = query.filter(SysUser.dept_id == deptId)
|
||
if status:
|
||
query = query.filter(SysUser.status == status)
|
||
query = apply_data_scope(query, SysUser, user, db, tenant_no)
|
||
total = query.count()
|
||
items = query.order_by(SysUser.create_time.desc()).offset((page - 1) * pageSize).limit(pageSize).all()
|
||
dept_ids = list({r.dept_id for r in items if r.dept_id})
|
||
dept_map: dict[str, str] = {}
|
||
if dept_ids:
|
||
for d in db.query(SysDept).filter(SysDept.id.in_(dept_ids), SysDept.del_flag == 0).all():
|
||
dept_map[d.id] = d.dept_name or ""
|
||
result_list = [
|
||
user_svc._row_to_dict(r, dept_name=dept_map.get(r.dept_id or "", "")) for r in items
|
||
]
|
||
tp = user_svc.total_pages(total, pageSize)
|
||
return success_response({"list": result_list, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
||
|
||
|
||
@router.get("/users/{user_id}")
|
||
def get_user_by_id(
|
||
user_id: str,
|
||
user: dict = Depends(require_menu_code("system:user:list")),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
tenant_no = _get_tenant_no(db, user)
|
||
row = user_svc.get_user(db, user_id)
|
||
if not row:
|
||
return error_response(404, "用户不存在")
|
||
if row.tenant_no != tenant_no and user.get("role") != "super_admin":
|
||
return error_response(403, "无权查看该用户")
|
||
try:
|
||
assert_row_in_data_scope(db, row, SysUser, user, tenant_no)
|
||
except HTTPException:
|
||
return error_response(403, "无权查看该用户")
|
||
data = user_svc.get_user_with_relations(db, user_id)
|
||
return success_response(data)
|
||
|
||
|
||
@router.post("/users")
|
||
def create_user(
|
||
body: UserCreateBody,
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
tenant_no = _get_tenant_no(db, user)
|
||
if user_svc.user_no_exists(db, body.user_no, tenant_no):
|
||
return error_response(400, "用户账号已存在")
|
||
obj = user_svc.create_user(
|
||
db,
|
||
user_no=body.user_no,
|
||
nick_name=body.nick_name,
|
||
password=body.password,
|
||
dept_id=body.dept_id,
|
||
email=body.email,
|
||
phone_no=body.phone_no,
|
||
sex=body.sex,
|
||
avatar=body.avatar,
|
||
status=body.status,
|
||
tenant_no=tenant_no,
|
||
order_num=body.order_num,
|
||
remark=body.remark,
|
||
user_id=user.get("user_id"),
|
||
role_ids=body.role_ids,
|
||
post_ids=body.post_ids,
|
||
)
|
||
return success_response(user_svc._row_to_dict(obj), message="创建成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
@router.put("/users/{user_id}")
|
||
def update_user(
|
||
user_id: str,
|
||
body: UserUpdateBody,
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
|
||
obj = user_svc.update_user(
|
||
db,
|
||
user_id,
|
||
nick_name=payload.get("nick_name"),
|
||
dept_id=payload.get("dept_id"),
|
||
email=payload.get("email"),
|
||
phone_no=payload.get("phone_no"),
|
||
sex=payload.get("sex"),
|
||
avatar=payload.get("avatar"),
|
||
status=payload.get("status"),
|
||
order_num=payload.get("order_num"),
|
||
remark=payload.get("remark"),
|
||
user_id_updater=user.get("user_id"),
|
||
role_ids=payload.get("role_ids"),
|
||
post_ids=payload.get("post_ids"),
|
||
)
|
||
return success_response(user_svc._row_to_dict(obj), message="更新成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
@router.delete("/users/{user_id}")
|
||
def delete_user(
|
||
user_id: str,
|
||
_user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
user_svc.soft_delete_user(db, user_id, user_info=_user)
|
||
return success_response(None, message="删除成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
@router.post("/users/{user_id}/reset-password")
|
||
def reset_user_password(
|
||
user_id: str,
|
||
_user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
user_svc.reset_password(db, user_id, "111111")
|
||
return success_response(None, message="密码已重置为 111111")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 字典类型管理
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class DictTypeCreateBody(BaseModel):
|
||
model_config = ConfigDict(populate_by_name=True)
|
||
|
||
dict_name: str = Field(..., alias="dictName")
|
||
dict_type: str = Field(..., alias="dictType")
|
||
order_num: Optional[int] = Field(0, alias="orderNum")
|
||
remark: Optional[str] = None
|
||
|
||
|
||
class DictTypeUpdateBody(BaseModel):
|
||
model_config = ConfigDict(populate_by_name=True)
|
||
|
||
dict_name: Optional[str] = Field(None, alias="dictName")
|
||
dict_type: Optional[str] = Field(None, alias="dictType")
|
||
order_num: Optional[int] = Field(None, alias="orderNum")
|
||
remark: Optional[str] = None
|
||
|
||
|
||
@router.get("/dict-types")
|
||
def list_dict_types(
|
||
page: int = Query(1, ge=1),
|
||
pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
|
||
keyword: Optional[str] = None,
|
||
user: dict = Depends(require_menu_code("system:dict:list")),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
tenant_no = _get_tenant_no(db, user)
|
||
items, total = dict_type_svc.list_dict_types(
|
||
db, page=page, page_size=pageSize, keyword=keyword, tenant_no=tenant_no, user_info=user
|
||
)
|
||
tp = dict_type_svc.total_pages(total, pageSize)
|
||
return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
||
|
||
|
||
@router.post("/dict-types")
|
||
def create_dict_type(
|
||
body: DictTypeCreateBody,
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
tenant_no = _get_tenant_no(db, user)
|
||
obj = dict_type_svc.create_dict_type(
|
||
db,
|
||
dict_name=body.dict_name,
|
||
dict_type=body.dict_type,
|
||
tenant_no=tenant_no,
|
||
order_num=body.order_num,
|
||
remark=body.remark,
|
||
user_id=user.get("user_id"),
|
||
)
|
||
return success_response(dict_type_svc._row_to_dict(obj), message="创建成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
@router.put("/dict-types/{dict_type_id}")
|
||
def update_dict_type(
|
||
dict_type_id: str,
|
||
body: DictTypeUpdateBody,
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
tenant_no = _get_tenant_no(db, user)
|
||
payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
|
||
obj = dict_type_svc.update_dict_type(
|
||
db,
|
||
dict_type_id,
|
||
dict_name=payload.get("dict_name"),
|
||
dict_type=payload.get("dict_type"),
|
||
order_num=payload.get("order_num"),
|
||
remark=payload.get("remark"),
|
||
user_id=user.get("user_id"),
|
||
tenant_no=tenant_no,
|
||
)
|
||
return success_response(dict_type_svc._row_to_dict(obj), message="更新成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
@router.delete("/dict-types/{dict_type_id}")
|
||
def delete_dict_type(
|
||
dict_type_id: str,
|
||
_user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
dict_type_svc.soft_delete_dict_type(db, dict_type_id, user_info=_user)
|
||
return success_response(None, message="删除成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 字典数据管理
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class DictDataCreateBody(BaseModel):
|
||
model_config = ConfigDict(populate_by_name=True)
|
||
|
||
dict_label: str = Field(..., alias="dictLabel")
|
||
dict_value: str = Field(..., alias="dictValue")
|
||
dict_type: str = Field(..., alias="dictType")
|
||
tag_type: Optional[str] = Field("", alias="tagType")
|
||
order_num: Optional[int] = Field(0, alias="orderNum")
|
||
remark: Optional[str] = None
|
||
|
||
|
||
class DictDataUpdateBody(BaseModel):
|
||
model_config = ConfigDict(populate_by_name=True)
|
||
|
||
dict_label: Optional[str] = Field(None, alias="dictLabel")
|
||
dict_value: Optional[str] = Field(None, alias="dictValue")
|
||
tag_type: Optional[str] = Field(None, alias="tagType")
|
||
order_num: Optional[int] = Field(None, alias="orderNum")
|
||
remark: Optional[str] = None
|
||
|
||
|
||
@router.get("/dict-data")
|
||
def list_dict_data(
|
||
page: int = Query(1, ge=1),
|
||
pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
|
||
dictType: Optional[str] = None,
|
||
keyword: Optional[str] = None,
|
||
user: dict = Depends(require_menu_code("system:dict:list")),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
tenant_no = _get_tenant_no(db, user)
|
||
items, total = dict_data_svc.list_dict_data(
|
||
db, page=page, page_size=pageSize, dict_type_code=dictType, keyword=keyword, tenant_no=tenant_no, user_info=user
|
||
)
|
||
tp = dict_data_svc.total_pages(total, pageSize)
|
||
return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
||
|
||
|
||
@router.post("/dict-data")
|
||
def create_dict_data(
|
||
body: DictDataCreateBody,
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
tenant_no = _get_tenant_no(db, user)
|
||
obj = dict_data_svc.create_dict_data(
|
||
db,
|
||
dict_label=body.dict_label,
|
||
dict_value=body.dict_value,
|
||
dict_type=body.dict_type,
|
||
tag_type=body.tag_type,
|
||
tenant_no=tenant_no,
|
||
order_num=body.order_num,
|
||
remark=body.remark,
|
||
user_id=user.get("user_id"),
|
||
)
|
||
return success_response(dict_data_svc._row_to_dict(obj), message="创建成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
@router.put("/dict-data/{data_id}")
|
||
def update_dict_data(
|
||
data_id: str,
|
||
body: DictDataUpdateBody,
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
|
||
obj = dict_data_svc.update_dict_data(
|
||
db,
|
||
data_id,
|
||
dict_label=payload.get("dict_label"),
|
||
dict_value=payload.get("dict_value"),
|
||
tag_type=payload.get("tag_type"),
|
||
order_num=payload.get("order_num"),
|
||
remark=payload.get("remark"),
|
||
user_id=user.get("user_id"),
|
||
)
|
||
return success_response(dict_data_svc._row_to_dict(obj), message="更新成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
@router.delete("/dict-data/{data_id}")
|
||
def delete_dict_data(
|
||
data_id: str,
|
||
_user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
dict_data_svc.soft_delete_dict_data(db, data_id, user_info=_user)
|
||
return success_response(None, message="删除成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
@router.get("/dict-data/options")
|
||
def dict_data_options(
|
||
dictType: str = Query(..., alias="dictType"),
|
||
user: dict = Depends(require_menu_code("system:dict:list")),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
tenant_no = _get_tenant_no(db, user)
|
||
# 字典选项为租户主数据，表单共用；不做数据范围过滤
|
||
items = dict_data_svc.list_dict_data_options(
|
||
db, dict_type_code=dictType, tenant_no=tenant_no, user_info=None
|
||
)
|
||
return success_response(items)
|
||
|
||
|
||
@router.get("/dict-data/all")
|
||
def dict_data_all(
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
tenant_no = _get_tenant_no(db, user)
|
||
q = db.query(SysDictData).filter(
|
||
and_(SysDictData.tenant_no == tenant_no, SysDictData.del_flag == 0)
|
||
)
|
||
rows = q.order_by(SysDictData.order_num.asc()).all()
|
||
grouped = {}
|
||
for r in rows:
|
||
key = r.dict_type
|
||
if key not in grouped:
|
||
grouped[key] = []
|
||
grouped[key].append({
|
||
"label": r.dict_label,
|
||
"value": r.dict_value,
|
||
"tagType": r.tag_type or ""
|
||
})
|
||
return success_response(grouped)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 组织管理
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class DeptCreateBody(BaseModel):
|
||
model_config = ConfigDict(populate_by_name=True)
|
||
|
||
parent_id: str = Field("0", alias="parentId")
|
||
dept_name: str = Field(..., alias="deptName")
|
||
dept_category_code: Optional[str] = Field(None, alias="deptCategoryCode")
|
||
status: Optional[str] = Field("1")
|
||
leader: Optional[str] = None
|
||
leader_phone: Optional[str] = Field(None, alias="leaderPhone")
|
||
order_num: Optional[int] = Field(0, alias="orderNum")
|
||
remark: Optional[str] = None
|
||
|
||
|
||
class DeptUpdateBody(BaseModel):
|
||
model_config = ConfigDict(populate_by_name=True)
|
||
|
||
dept_name: Optional[str] = Field(None, alias="deptName")
|
||
dept_category_code: Optional[str] = Field(None, alias="deptCategoryCode")
|
||
status: Optional[str] = None
|
||
leader: Optional[str] = None
|
||
leader_phone: Optional[str] = Field(None, alias="leaderPhone")
|
||
order_num: Optional[int] = Field(None, alias="orderNum")
|
||
remark: Optional[str] = None
|
||
|
||
|
||
@router.get("/depts")
|
||
def list_depts(
|
||
keyword: Optional[str] = None,
|
||
user: dict = Depends(require_menu_code("system:dept:list")),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
tenant_no = _get_tenant_no(db, user)
|
||
tree = dept_svc.list_depts_tree(db, keyword=keyword, tenant_no=tenant_no, user_info=user)
|
||
return success_response(tree)
|
||
|
||
|
||
@router.post("/depts")
|
||
def create_dept(
|
||
body: DeptCreateBody,
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
tenant_no = _get_tenant_no(db, user)
|
||
obj = dept_svc.create_dept(
|
||
db,
|
||
parent_id=body.parent_id,
|
||
dept_name=body.dept_name,
|
||
dept_category_code=body.dept_category_code,
|
||
status=body.status,
|
||
leader=body.leader,
|
||
leader_phone=body.leader_phone,
|
||
tenant_no=tenant_no,
|
||
order_num=body.order_num,
|
||
remark=body.remark,
|
||
user_id=user.get("user_id"),
|
||
)
|
||
return success_response(dept_svc._row_to_dict(obj), message="创建成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
@router.put("/depts/{dept_id}")
|
||
def update_dept(
|
||
dept_id: str,
|
||
body: DeptUpdateBody,
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
|
||
obj = dept_svc.update_dept(
|
||
db,
|
||
dept_id,
|
||
dept_name=payload.get("dept_name"),
|
||
dept_category_code=payload.get("dept_category_code"),
|
||
status=payload.get("status"),
|
||
leader=payload.get("leader"),
|
||
leader_phone=payload.get("leader_phone"),
|
||
order_num=payload.get("order_num"),
|
||
remark=payload.get("remark"),
|
||
user_id=user.get("user_id"),
|
||
)
|
||
return success_response(dept_svc._row_to_dict(obj), message="更新成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
@router.delete("/depts/{dept_id}")
|
||
def delete_dept(
|
||
dept_id: str,
|
||
_user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
dept_svc.soft_delete_dept(db, dept_id, user_info=_user)
|
||
return success_response(None, message="删除成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 站点管理
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class SiteCreateBody(BaseModel):
|
||
model_config = ConfigDict(populate_by_name=True)
|
||
|
||
site_name: str = Field(..., alias="siteName")
|
||
site_code: str = Field(..., alias="siteCode")
|
||
site_type: Optional[str] = Field(None, alias="siteType")
|
||
status: Optional[str] = Field("1")
|
||
address: Optional[str] = None
|
||
contact_name: Optional[str] = Field(None, alias="contactName")
|
||
contact_phone: Optional[str] = Field(None, alias="contactPhone")
|
||
business_scope: Optional[str] = Field(None, alias="businessScope")
|
||
order_num: Optional[int] = Field(0, alias="orderNum")
|
||
remark: Optional[str] = None
|
||
|
||
|
||
class SiteUpdateBody(BaseModel):
|
||
model_config = ConfigDict(populate_by_name=True)
|
||
|
||
site_name: Optional[str] = Field(None, alias="siteName")
|
||
site_code: Optional[str] = Field(None, alias="siteCode")
|
||
site_type: Optional[str] = Field(None, alias="siteType")
|
||
status: Optional[str] = None
|
||
address: Optional[str] = None
|
||
contact_name: Optional[str] = Field(None, alias="contactName")
|
||
contact_phone: Optional[str] = Field(None, alias="contactPhone")
|
||
business_scope: Optional[str] = Field(None, alias="businessScope")
|
||
order_num: Optional[int] = Field(None, alias="orderNum")
|
||
remark: Optional[str] = None
|
||
|
||
|
||
@router.get("/sites")
|
||
def list_sites(
|
||
page: int = Query(1, ge=1),
|
||
pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
|
||
keyword: Optional[str] = None,
|
||
user: dict = Depends(require_menu_code("system:site:list")),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
tenant_no = _get_tenant_no(db, user)
|
||
items, total = site_svc.list_sites(db, page=page, page_size=pageSize, keyword=keyword, tenant_no=tenant_no, user_info=user)
|
||
tp = site_svc.total_pages(total, pageSize)
|
||
return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
||
|
||
|
||
@router.post("/sites")
|
||
def create_site(
|
||
body: SiteCreateBody,
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
tenant_no = _get_tenant_no(db, user)
|
||
obj = site_svc.create_site(
|
||
db,
|
||
site_name=body.site_name,
|
||
site_code=body.site_code,
|
||
site_type=body.site_type,
|
||
status=body.status,
|
||
address=body.address,
|
||
contact_name=body.contact_name,
|
||
contact_phone=body.contact_phone,
|
||
business_scope=body.business_scope,
|
||
tenant_no=tenant_no,
|
||
order_num=body.order_num,
|
||
remark=body.remark,
|
||
user_id=user.get("user_id"),
|
||
)
|
||
return success_response(site_svc._row_to_dict(obj), message="创建成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
@router.put("/sites/{site_id}")
|
||
def update_site(
|
||
site_id: str,
|
||
body: SiteUpdateBody,
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
|
||
obj = site_svc.update_site(
|
||
db,
|
||
site_id,
|
||
site_name=payload.get("site_name"),
|
||
site_code=payload.get("site_code"),
|
||
site_type=payload.get("site_type"),
|
||
status=payload.get("status"),
|
||
address=payload.get("address"),
|
||
contact_name=payload.get("contact_name"),
|
||
contact_phone=payload.get("contact_phone"),
|
||
business_scope=payload.get("business_scope"),
|
||
order_num=payload.get("order_num"),
|
||
remark=payload.get("remark"),
|
||
user_id=user.get("user_id"),
|
||
)
|
||
return success_response(site_svc._row_to_dict(obj), message="更新成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
@router.delete("/sites/{site_id}")
|
||
def delete_site(
|
||
site_id: str,
|
||
_user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
site_svc.soft_delete_site(db, site_id, user_info=_user)
|
||
return success_response(None, message="删除成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 职位管理
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class PostCreateBody(BaseModel):
|
||
model_config = ConfigDict(populate_by_name=True)
|
||
|
||
post_name: str = Field(..., alias="postName")
|
||
post_code: str = Field(..., alias="postCode")
|
||
status: Optional[str] = Field("1")
|
||
order_num: Optional[int] = Field(0, alias="orderNum")
|
||
remark: Optional[str] = None
|
||
|
||
|
||
class PostUpdateBody(BaseModel):
|
||
model_config = ConfigDict(populate_by_name=True)
|
||
|
||
post_name: Optional[str] = Field(None, alias="postName")
|
||
post_code: Optional[str] = Field(None, alias="postCode")
|
||
status: Optional[str] = None
|
||
order_num: Optional[int] = Field(None, alias="orderNum")
|
||
remark: Optional[str] = None
|
||
|
||
|
||
@router.get("/posts")
|
||
def list_posts(
|
||
page: int = Query(1, ge=1),
|
||
pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
|
||
keyword: Optional[str] = None,
|
||
user: dict = Depends(require_menu_code("system:post:list")),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
tenant_no = _get_tenant_no(db, user)
|
||
items, total = post_svc.list_posts(db, page=page, page_size=pageSize, keyword=keyword, tenant_no=tenant_no, user_info=user)
|
||
tp = post_svc.total_pages(total, pageSize)
|
||
return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
||
|
||
|
||
@router.post("/posts")
|
||
def create_post(
|
||
body: PostCreateBody,
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
tenant_no = _get_tenant_no(db, user)
|
||
obj = post_svc.create_post(
|
||
db,
|
||
post_name=body.post_name,
|
||
post_code=body.post_code,
|
||
status=body.status,
|
||
tenant_no=tenant_no,
|
||
order_num=body.order_num,
|
||
remark=body.remark,
|
||
user_id=user.get("user_id"),
|
||
)
|
||
return success_response(post_svc._row_to_dict(obj), message="创建成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
@router.put("/posts/{post_id}")
|
||
def update_post(
|
||
post_id: str,
|
||
body: PostUpdateBody,
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
|
||
obj = post_svc.update_post(
|
||
db,
|
||
post_id,
|
||
post_name=payload.get("post_name"),
|
||
post_code=payload.get("post_code"),
|
||
status=payload.get("status"),
|
||
order_num=payload.get("order_num"),
|
||
remark=payload.get("remark"),
|
||
user_id=user.get("user_id"),
|
||
)
|
||
return success_response(post_svc._row_to_dict(obj), message="更新成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
@router.delete("/posts/{post_id}")
|
||
def delete_post(
|
||
post_id: str,
|
||
_user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
post_svc.soft_delete_post(db, post_id, user_info=_user)
|
||
return success_response(None, message="删除成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
class PostUsersUpdateBody(BaseModel):
|
||
model_config = ConfigDict(populate_by_name=True)
|
||
|
||
user_ids: List[str] = Field(default_factory=list, alias="userIds")
|
||
|
||
|
||
@router.get("/posts/{post_id}/users")
|
||
def list_post_bound_users(
|
||
post_id: str,
|
||
user: dict = Depends(require_menu_code("system:post:list")),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
tenant_no = _get_tenant_no(db, user)
|
||
post = post_svc.get_post(db, post_id)
|
||
if not post or post.tenant_no != tenant_no:
|
||
return error_response(404, "职位不存在")
|
||
ids = user_svc.list_user_ids_bound_to_post(db, post_id, tenant_no)
|
||
return success_response({"userIds": ids})
|
||
|
||
|
||
@router.put("/posts/{post_id}/users")
|
||
def update_post_bound_users(
|
||
post_id: str,
|
||
body: PostUsersUpdateBody,
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
tenant_no = _get_tenant_no(db, user)
|
||
post = post_svc.get_post(db, post_id)
|
||
if not post or post.tenant_no != tenant_no:
|
||
return error_response(404, "职位不存在")
|
||
try:
|
||
user_svc.sync_post_user_bindings(
|
||
db, post_id, tenant_no, body.user_ids, user.get("user_id")
|
||
)
|
||
return success_response(None, message="保存成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 菜单管理
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class MenuCreateBody(BaseModel):
|
||
model_config = ConfigDict(populate_by_name=True)
|
||
|
||
menu_name: str = Field(..., alias="menuName")
|
||
parent_id: str = Field("0", alias="parentId")
|
||
menu_type: str = Field("M", alias="menuType")
|
||
path: Optional[str] = Field("", alias="path")
|
||
component: Optional[str] = Field(None, alias="component")
|
||
perms: Optional[str] = Field(None, alias="perms")
|
||
is_frame: Optional[int] = Field(1, alias="isFrame")
|
||
visible: Optional[str] = Field("1", alias="visible")
|
||
status: Optional[str] = Field("1", alias="status")
|
||
order_num: Optional[int] = Field(None, alias="orderNum")
|
||
remark: Optional[str] = None
|
||
|
||
|
||
class MenuUpdateBody(BaseModel):
|
||
model_config = ConfigDict(populate_by_name=True)
|
||
|
||
menu_name: Optional[str] = Field(None, alias="menuName")
|
||
parent_id: Optional[str] = Field(None, alias="parentId")
|
||
menu_type: Optional[str] = Field(None, alias="menuType")
|
||
path: Optional[str] = Field(None, alias="path")
|
||
component: Optional[str] = Field(None, alias="component")
|
||
perms: Optional[str] = Field(None, alias="perms")
|
||
is_frame: Optional[int] = Field(None, alias="isFrame")
|
||
visible: Optional[str] = Field(None, alias="visible")
|
||
status: Optional[str] = Field(None, alias="status")
|
||
order_num: Optional[int] = Field(None, alias="orderNum")
|
||
remark: Optional[str] = None
|
||
|
||
|
||
@router.get("/menus")
|
||
def list_menus(
|
||
keyword: Optional[str] = None,
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
tenant_no = _get_tenant_no(db, user)
|
||
tree = menu_svc.list_menus_tree(db, keyword=keyword, tenant_no=tenant_no, user_info=user)
|
||
return success_response(tree)
|
||
|
||
|
||
@router.post("/menus")
|
||
def create_menu(
|
||
body: MenuCreateBody,
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
tenant_no = _get_tenant_no(db, user)
|
||
obj = menu_svc.create_menu(
|
||
db,
|
||
menu_name=body.menu_name,
|
||
parent_id=body.parent_id,
|
||
menu_type=body.menu_type,
|
||
path=body.path,
|
||
component=body.component,
|
||
perms=body.perms,
|
||
is_frame=body.is_frame,
|
||
visible=body.visible,
|
||
status=body.status,
|
||
tenant_no=tenant_no,
|
||
order_num=body.order_num,
|
||
remark=body.remark,
|
||
user_id=user.get("user_id"),
|
||
)
|
||
return success_response(menu_svc._row_to_dict(obj), message="创建成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
@router.put("/menus/{menu_id}")
|
||
def update_menu(
|
||
menu_id: str,
|
||
body: MenuUpdateBody,
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
|
||
obj = menu_svc.update_menu(
|
||
db,
|
||
menu_id,
|
||
menu_name=payload.get("menu_name"),
|
||
parent_id=payload.get("parent_id"),
|
||
menu_type=payload.get("menu_type"),
|
||
path=payload.get("path"),
|
||
component=payload.get("component"),
|
||
perms=payload.get("perms"),
|
||
is_frame=payload.get("is_frame"),
|
||
visible=payload.get("visible"),
|
||
status=payload.get("status"),
|
||
order_num=payload.get("order_num"),
|
||
remark=payload.get("remark"),
|
||
user_id=user.get("user_id"),
|
||
)
|
||
return success_response(menu_svc._row_to_dict(obj), message="更新成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
@router.delete("/menus/{menu_id}")
|
||
def delete_menu(
|
||
menu_id: str,
|
||
_user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
menu_svc.soft_delete_menu(db, menu_id, user_info=_user)
|
||
return success_response(None, message="删除成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 角色管理
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class RoleCreateBody(BaseModel):
|
||
model_config = ConfigDict(populate_by_name=True)
|
||
|
||
role_name: str = Field(..., alias="roleName")
|
||
data_scope: Optional[str] = Field("4", alias="dataScope")
|
||
status: Optional[str] = Field("1", alias="status")
|
||
order_num: Optional[int] = Field(None, alias="orderNum")
|
||
remark: Optional[str] = None
|
||
|
||
|
||
class RoleUpdateBody(BaseModel):
|
||
model_config = ConfigDict(populate_by_name=True)
|
||
|
||
role_name: Optional[str] = Field(None, alias="roleName")
|
||
data_scope: Optional[str] = Field(None, alias="dataScope")
|
||
status: Optional[str] = Field(None, alias="status")
|
||
order_num: Optional[int] = Field(None, alias="orderNum")
|
||
remark: Optional[str] = None
|
||
|
||
|
||
class RoleMenuBody(BaseModel):
|
||
menu_ids: List[str] = Field(..., alias="menuIds")
|
||
|
||
|
||
class RoleDataScopeBody(BaseModel):
|
||
model_config = ConfigDict(populate_by_name=True)
|
||
|
||
data_scope: str = Field(..., alias="dataScope")
|
||
scope_details: Optional[List[dict]] = Field(None, alias="scopeDetails")
|
||
|
||
|
||
@router.get("/role-management")
|
||
def list_role_management(
|
||
page: int = Query(1, ge=1),
|
||
pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
|
||
keyword: Optional[str] = None,
|
||
user: dict = Depends(require_menu_code("system:role:list")),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
tenant_no = _get_tenant_no(db, user)
|
||
items, total = role_svc.list_roles(db, page=page, page_size=pageSize, keyword=keyword, tenant_no=tenant_no, user_info=user)
|
||
tp = role_svc.total_pages(total, pageSize)
|
||
return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
||
|
||
|
||
@router.post("/role-management")
|
||
def create_role_management(
|
||
body: RoleCreateBody,
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
tenant_no = _get_tenant_no(db, user)
|
||
obj = role_svc.create_role(
|
||
db,
|
||
role_name=body.role_name,
|
||
data_scope=body.data_scope,
|
||
status=body.status,
|
||
tenant_no=tenant_no,
|
||
order_num=body.order_num,
|
||
remark=body.remark,
|
||
user_id=user.get("user_id"),
|
||
)
|
||
return success_response(role_svc._row_to_dict(obj), message="创建成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
@router.put("/role-management/{role_id}")
|
||
def update_role_management(
|
||
role_id: str,
|
||
body: RoleUpdateBody,
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
|
||
obj = role_svc.update_role(
|
||
db,
|
||
role_id,
|
||
role_name=payload.get("role_name"),
|
||
data_scope=payload.get("data_scope"),
|
||
status=payload.get("status"),
|
||
order_num=payload.get("order_num"),
|
||
remark=payload.get("remark"),
|
||
user_id=user.get("user_id"),
|
||
)
|
||
return success_response(role_svc._row_to_dict(obj), message="更新成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
@router.delete("/role-management/{role_id}")
|
||
def delete_role_management(
|
||
role_id: str,
|
||
_user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
role_svc.soft_delete_role(db, role_id, user_info=_user)
|
||
return success_response(None, message="删除成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
@router.get("/role-management/{role_id}/menus")
|
||
def get_role_menus(
|
||
role_id: str,
|
||
_user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
menu_ids = role_svc.get_role_menu_ids(db, role_id)
|
||
return success_response({"menuIds": menu_ids})
|
||
|
||
|
||
@router.put("/role-management/{role_id}/menus")
|
||
def update_role_menus(
|
||
role_id: str,
|
||
body: RoleMenuBody,
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
tenant_no = _get_tenant_no(db, user)
|
||
role_svc.update_role_menus(db, role_id, body.menu_ids, tenant_no, user_id=user.get("user_id"))
|
||
return success_response(None, message="更新成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
@router.get("/role-management/{role_id}/data-scope")
|
||
def get_role_data_scope(
|
||
role_id: str,
|
||
_user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
role = role_svc.get_role(db, role_id)
|
||
if not role:
|
||
return error_response(404, "角色不存在")
|
||
details = role_svc.get_role_data_scope(db, role_id)
|
||
return success_response({"dataScope": role.data_scope or "1", "scopeDetails": details})
|
||
|
||
|
||
@router.put("/role-management/{role_id}/data-scope")
|
||
def update_role_data_scope(
|
||
role_id: str,
|
||
body: RoleDataScopeBody,
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
tenant_no = _get_tenant_no(db, user)
|
||
role_svc.update_role_data_scope(
|
||
db,
|
||
role_id,
|
||
body.data_scope,
|
||
body.scope_details,
|
||
tenant_no=tenant_no,
|
||
user_id=user.get("user_id"),
|
||
)
|
||
return success_response(None, message="更新成功")
|
||
except ValueError as e:
|
||
return error_response(400, str(e))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 附件上传
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _format_file_size(size_bytes: int) -> str:
|
||
if size_bytes < 1024:
|
||
return f"{size_bytes} B"
|
||
elif size_bytes < 1024 * 1024:
|
||
return f"{size_bytes / 1024:.1f} KB"
|
||
else:
|
||
return f"{size_bytes / (1024 * 1024):.1f} MB"
|
||
|
||
|
||
@router.post("/attachment/upload")
|
||
def upload_attachment(
|
||
file: UploadFile = File(...),
|
||
folder: Optional[str] = Form(""),
|
||
biz_type: Optional[str] = Form(None, alias="bizType"),
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
if not file or not file.filename:
|
||
return error_response(400, "请选择文件")
|
||
try:
|
||
effective_folder = resolve_upload_folder(
|
||
folder=folder or "", biz_type=biz_type or ""
|
||
)
|
||
file_content = file.file.read()
|
||
full_url = storage.upload_file(file_content, file.filename, effective_folder)
|
||
base_url = settings.MINIO_BASE_URL.rstrip("/") if settings.MINIO_BASE_URL else ""
|
||
if base_url and full_url.startswith(base_url + "/"):
|
||
file_url = full_url[len(base_url) + 1:]
|
||
else:
|
||
file_url = full_url
|
||
file_size = len(file_content)
|
||
file_ext = os.path.splitext(file.filename)[1].lstrip(".") if file.filename else ""
|
||
file_type = file.content_type or ""
|
||
tenant_no = _get_tenant_no(db, user)
|
||
record = oss_svc.create_oss(
|
||
db,
|
||
tenant_no=tenant_no,
|
||
url=file_url,
|
||
original_name=file.filename,
|
||
file_suffix=file_ext,
|
||
file_size=file_size,
|
||
file_type=file_type,
|
||
biz_type=(biz_type or "").strip(),
|
||
user_id=user.get("user_id"),
|
||
ext1=effective_folder or None,
|
||
)
|
||
return success_response({
|
||
"id": record.id,
|
||
"fileName": record.original_name,
|
||
"fileUrl": record.url,
|
||
"fileSize": record.file_size,
|
||
"fileSizeDisplay": _format_file_size(record.file_size or 0),
|
||
"fileType": record.file_type,
|
||
"bizType": record.biz_type or "",
|
||
})
|
||
except Exception as e:
|
||
db.rollback()
|
||
return error_response(500, f"上传失败: {str(e)}")
|
||
|
||
|
||
class AttachmentDownloadUrlBody(BaseModel):
|
||
model_config = ConfigDict(populate_by_name=True)
|
||
file_url: str = Field(..., alias="fileUrl")
|
||
file_name: Optional[str] = Field(None, alias="fileName")
|
||
|
||
|
||
@router.post("/attachment/download-url")
|
||
def attachment_download_url(
|
||
body: AttachmentDownloadUrlBody,
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
"""返回带 Content-Disposition 的下载 URL，保证浏览器使用原始文件名。"""
|
||
file_url = (body.file_url or "").strip()
|
||
if not file_url:
|
||
return error_response(400, "文件地址不能为空")
|
||
tenant_no = _get_tenant_no(db, user)
|
||
file_url, att = oss_svc.resolve_storage_path(db, tenant_no, file_url)
|
||
if not att:
|
||
att = oss_svc.get_oss_by_url(db, tenant_no, file_url)
|
||
download_name = (body.file_name or "").strip()
|
||
if not download_name and att:
|
||
download_name = (att.original_name or "").strip()
|
||
if not download_name:
|
||
download_name = file_url.rsplit("/", 1)[-1] or "download"
|
||
try:
|
||
url = storage.presigned_download_url(file_url, download_name)
|
||
return success_response({"downloadUrl": url})
|
||
except Exception as e:
|
||
return error_response(500, f"获取下载链接失败: {str(e)}")
|
||
|
||
|
||
@router.post("/attachment/preview-url")
|
||
def attachment_preview_url(
|
||
body: AttachmentDownloadUrlBody,
|
||
user: dict = Depends(require_admin_or_super_admin),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
"""返回用于浏览器内联预览的 URL（PDF iframe 等）。"""
|
||
file_url = (body.file_url or "").strip()
|
||
if not file_url:
|
||
return error_response(400, "文件地址不能为空")
|
||
tenant_no = _get_tenant_no(db, user)
|
||
file_url, _att = oss_svc.resolve_storage_path(db, tenant_no, file_url)
|
||
try:
|
||
url = storage.presigned_preview_url(file_url)
|
||
return success_response({"previewUrl": url})
|
||
except Exception as e:
|
||
return error_response(500, f"获取预览链接失败: {str(e)}")
|