224 lines
8.1 KiB
Python
224 lines
8.1 KiB
Python
import csv
|
||
import io
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, UploadFile
|
||
from sqlalchemy import delete, or_, select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.db import get_db
|
||
from app.deps import require_admin
|
||
from app.models import Unit, User
|
||
from app.models.chat import Attachment, Feedback, Message, MessageReference
|
||
from app.models.user import user_roles
|
||
from app.schemas.user import BulkImportResult, UserCreate, UserResponse, UserUpdate
|
||
from app.services.audit_service import write_audit
|
||
from app.services.user_service import apply_roles, ensure_roles
|
||
|
||
router = APIRouter()
|
||
|
||
|
||
def serialize_user(user: User) -> UserResponse:
|
||
return UserResponse(
|
||
id=user.id,
|
||
phone=user.phone,
|
||
name=user.name,
|
||
department=user.department,
|
||
unit_id=user.unit_id,
|
||
unit_name=user.unit.name if user.unit else None,
|
||
status=user.status,
|
||
role_codes=[role.code for role in user.roles],
|
||
is_superuser=user.is_superuser,
|
||
created_at=user.created_at,
|
||
)
|
||
|
||
|
||
def _validate_unit_id(db: Session, unit_id: int | None) -> None:
|
||
if unit_id is None:
|
||
return
|
||
if not db.get(Unit, unit_id):
|
||
raise HTTPException(status_code=400, detail="所选单位不存在")
|
||
|
||
|
||
@router.get("", response_model=list[UserResponse])
|
||
def list_users(
|
||
q: str = "",
|
||
status: str = "",
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(require_admin),
|
||
) -> list[UserResponse]:
|
||
stmt = select(User).order_by(User.id.desc())
|
||
q = (q or "").strip()
|
||
if q:
|
||
like = f"%{q}%"
|
||
stmt = stmt.where(or_(
|
||
User.name.ilike(like),
|
||
User.phone.ilike(like),
|
||
User.department.ilike(like),
|
||
))
|
||
status = (status or "").strip()
|
||
if status:
|
||
stmt = stmt.where(User.status == status)
|
||
users = db.scalars(stmt).all()
|
||
return [serialize_user(user) for user in users]
|
||
|
||
|
||
@router.post("", response_model=UserResponse)
|
||
def create_user(payload: UserCreate, db: Session = Depends(get_db), admin: User = Depends(require_admin)) -> UserResponse:
|
||
existing = db.scalar(select(User).where(User.phone == payload.phone))
|
||
if existing:
|
||
raise HTTPException(status_code=409, detail="手机号已存在")
|
||
_validate_unit_id(db, payload.unit_id)
|
||
user = User(
|
||
phone=payload.phone,
|
||
name=payload.name,
|
||
department=payload.department,
|
||
unit_id=payload.unit_id,
|
||
is_superuser=payload.is_superuser,
|
||
)
|
||
db.add(user)
|
||
db.flush()
|
||
apply_roles(db, user, payload.role_codes)
|
||
write_audit(db, "user.create", actor_id=admin.id, target_type="user", target_id=str(user.id))
|
||
db.commit()
|
||
db.refresh(user)
|
||
return serialize_user(user)
|
||
|
||
|
||
@router.patch("/{user_id}", response_model=UserResponse)
|
||
def update_user(
|
||
user_id: int,
|
||
payload: UserUpdate,
|
||
db: Session = Depends(get_db),
|
||
admin: User = Depends(require_admin),
|
||
) -> UserResponse:
|
||
user = db.get(User, user_id)
|
||
if not user:
|
||
raise HTTPException(status_code=404, detail="用户不存在")
|
||
updates = payload.model_dump(exclude_unset=True)
|
||
if "unit_id" in updates:
|
||
_validate_unit_id(db, updates["unit_id"])
|
||
user.unit_id = updates["unit_id"]
|
||
for field in ("name", "department", "status", "is_superuser"):
|
||
if field in updates and updates[field] is not None:
|
||
setattr(user, field, updates[field])
|
||
if payload.role_codes is not None:
|
||
apply_roles(db, user, payload.role_codes)
|
||
write_audit(db, "user.update", actor_id=admin.id, target_type="user", target_id=str(user.id))
|
||
db.commit()
|
||
db.refresh(user)
|
||
return serialize_user(user)
|
||
|
||
|
||
@router.delete("/{user_id}")
|
||
def delete_user(
|
||
user_id: int,
|
||
db: Session = Depends(get_db),
|
||
admin: User = Depends(require_admin),
|
||
) -> dict:
|
||
user = db.get(User, user_id)
|
||
if not user:
|
||
raise HTTPException(status_code=404, detail="用户不存在")
|
||
if user.id == admin.id:
|
||
raise HTTPException(status_code=400, detail="不能删除当前登录账户")
|
||
# 清理关联：user_roles / 该用户的消息/反馈/附件/引用
|
||
db.execute(delete(user_roles).where(user_roles.c.user_id == user_id))
|
||
msg_ids = list(db.scalars(select(Message.id).where(Message.user_id == user_id)))
|
||
if msg_ids:
|
||
db.execute(delete(MessageReference).where(MessageReference.message_id.in_(msg_ids)))
|
||
db.execute(delete(Feedback).where(Feedback.message_id.in_(msg_ids)))
|
||
db.execute(delete(Message).where(Message.id.in_(msg_ids)))
|
||
db.execute(delete(Attachment).where(Attachment.user_id == user_id))
|
||
write_audit(db, "user.delete", actor_id=admin.id, target_type="user", target_id=str(user_id),
|
||
detail={"phone": user.phone, "name": user.name})
|
||
db.delete(user)
|
||
db.commit()
|
||
return {"deleted": True, "id": user_id}
|
||
|
||
|
||
@router.post("/bulk-import", response_model=BulkImportResult)
|
||
def bulk_import(
|
||
file: UploadFile,
|
||
db: Session = Depends(get_db),
|
||
admin: User = Depends(require_admin),
|
||
) -> BulkImportResult:
|
||
ensure_roles(db)
|
||
name = (file.filename or "").lower()
|
||
raw = file.file.read()
|
||
if not raw:
|
||
raise HTTPException(status_code=400, detail="文件为空")
|
||
|
||
try:
|
||
rows = _parse_user_rows(raw, name)
|
||
except RuntimeError as exc:
|
||
raise HTTPException(status_code=400, detail=str(exc))
|
||
|
||
created = 0
|
||
skipped = 0
|
||
errors: list[str] = []
|
||
existing_phones = {p for p, in db.execute(select(User.phone))}
|
||
for idx, row in enumerate(rows, start=2): # 行号 from 2 (header was 1)
|
||
phone = (row.get("phone") or "").strip()
|
||
u_name = (row.get("name") or "").strip()
|
||
dept = (row.get("department") or "").strip() or None
|
||
roles_raw = (row.get("role_codes") or "").strip()
|
||
if not phone or not u_name:
|
||
errors.append(f"第 {idx} 行：手机号或姓名为空")
|
||
continue
|
||
if phone in existing_phones:
|
||
skipped += 1
|
||
continue
|
||
role_codes = [c.strip() for c in roles_raw.replace("，", ",").split(",") if c.strip()]
|
||
user = User(phone=phone, name=u_name, department=dept)
|
||
db.add(user)
|
||
db.flush()
|
||
apply_roles(db, user, role_codes)
|
||
existing_phones.add(phone)
|
||
created += 1
|
||
|
||
write_audit(db, "user.bulk_import", actor_id=admin.id, target_type="user",
|
||
detail={"created": created, "skipped": skipped, "errors": len(errors)})
|
||
db.commit()
|
||
return BulkImportResult(created=created, skipped=skipped, errors=errors)
|
||
|
||
|
||
def _parse_user_rows(raw: bytes, filename: str) -> list[dict]:
|
||
"""支持 CSV / XLSX。列：phone, name, department(可选), role_codes(可选，多个用逗号分隔)。"""
|
||
header_map = {
|
||
"手机号": "phone", "phone": "phone", "电话": "phone",
|
||
"姓名": "name", "name": "name", "用户名": "name", "用户名称": "name",
|
||
"部门": "department", "department": "department",
|
||
"角色": "role_codes", "role_codes": "role_codes", "角色代码": "role_codes",
|
||
}
|
||
|
||
def normalize(rows: list[list[str]]) -> list[dict]:
|
||
if not rows:
|
||
return []
|
||
head = [str(c or "").strip() for c in rows[0]]
|
||
keys = [header_map.get(c, c.lower()) for c in head]
|
||
out: list[dict] = []
|
||
for r in rows[1:]:
|
||
if not any((str(x) or "").strip() for x in r):
|
||
continue
|
||
d: dict = {}
|
||
for k, v in zip(keys, r):
|
||
d[k] = "" if v is None else str(v)
|
||
out.append(d)
|
||
return out
|
||
|
||
if filename.endswith(".csv") or filename.endswith(".txt"):
|
||
text = raw.decode("utf-8-sig", errors="replace")
|
||
reader = csv.reader(io.StringIO(text))
|
||
return normalize(list(reader))
|
||
|
||
if filename.endswith(".xlsx") or filename.endswith(".xls"):
|
||
try:
|
||
from openpyxl import load_workbook
|
||
except ImportError:
|
||
raise RuntimeError("缺少 openpyxl，无法解析 XLSX。请使用 CSV 或运行 pip install openpyxl")
|
||
wb = load_workbook(io.BytesIO(raw), read_only=True, data_only=True)
|
||
ws = wb.active
|
||
rows = [[cell for cell in row] for row in ws.iter_rows(values_only=True)]
|
||
return normalize(rows)
|
||
|
||
raise RuntimeError("仅支持 CSV 或 XLSX")
|