393 lines
15 KiB
Python
393 lines
15 KiB
Python
import csv
|
||
import io
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, UploadFile
|
||
from fastapi.responses import StreamingResponse
|
||
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 JobTitle, 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,
|
||
job_title=user.job_title,
|
||
unit_id=user.unit_id,
|
||
unit_name=user.unit.name if user.unit else None,
|
||
department_id=user.department_id,
|
||
department_name=user.department.name if user.department 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(db: Session, unit_id: int | None, *, expected_category: str, field_label: str) -> None:
|
||
if unit_id is None:
|
||
return
|
||
unit = db.get(Unit, unit_id)
|
||
if not unit:
|
||
raise HTTPException(status_code=400, detail=f"所选{field_label}不存在")
|
||
if unit.category != expected_category:
|
||
raise HTTPException(status_code=400, detail=f"所选{field_label}分类不匹配")
|
||
|
||
|
||
@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),
|
||
))
|
||
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(db, payload.unit_id, expected_category="company", field_label="单位")
|
||
_validate_unit(db, payload.department_id, expected_category="department", field_label="部门")
|
||
user = User(
|
||
phone=payload.phone,
|
||
name=payload.name,
|
||
job_title=payload.job_title,
|
||
unit_id=payload.unit_id,
|
||
department_id=payload.department_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(db, updates["unit_id"], expected_category="company", field_label="单位")
|
||
user.unit_id = updates["unit_id"]
|
||
if "department_id" in updates:
|
||
_validate_unit(db, updates["department_id"], expected_category="department", field_label="部门")
|
||
user.department_id = updates["department_id"]
|
||
for field in ("name", "job_title", "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.get("/import-template")
|
||
def download_import_template(
|
||
db: Session = Depends(get_db),
|
||
admin: User = Depends(require_admin),
|
||
) -> StreamingResponse:
|
||
try:
|
||
from openpyxl import Workbook
|
||
from openpyxl.utils import quote_sheetname
|
||
from openpyxl.worksheet.datavalidation import DataValidation
|
||
except ImportError:
|
||
raise HTTPException(status_code=500, detail="缺少 openpyxl，请运行 pip install openpyxl")
|
||
|
||
job_titles = db.scalars(
|
||
select(JobTitle)
|
||
.where(JobTitle.is_active == True) # noqa: E712
|
||
.order_by(JobTitle.sort_order, JobTitle.id)
|
||
).all()
|
||
job_title_names = [jt.name for jt in job_titles]
|
||
|
||
units_all = db.scalars(select(Unit).order_by(Unit.id)).all()
|
||
company_names = [u.name for u in units_all if u.category == "company"]
|
||
department_names = [u.name for u in units_all if u.category == "department"]
|
||
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.title = "用户导入"
|
||
headers = ["姓名", "手机号", "单位", "部门", "职级", "角色", "状态"]
|
||
ws.append(headers)
|
||
sample_jt = job_title_names[0] if job_title_names else ""
|
||
sample_company = company_names[0] if company_names else ""
|
||
sample_dept = department_names[0] if department_names else ""
|
||
ws.append(["张三", "13800000001", sample_company, sample_dept, sample_jt, "普通员工", "启用"])
|
||
ws.append(["李四", "13800000002", sample_company, sample_dept, sample_jt, "业务管理员", "启用"])
|
||
ws.append(["王五", "13800000003", sample_company, sample_dept, sample_jt, "系统管理员", "停用"])
|
||
|
||
# 隐藏字典 sheet：A 列职级、B 列公司、C 列部门 —— 都从表里实时读，
|
||
# Excel 里改后台字典下次下模板就同步。
|
||
dict_ws = wb.create_sheet("_dict")
|
||
dict_ws.sheet_state = "hidden"
|
||
for i, name in enumerate(job_title_names, start=1):
|
||
dict_ws.cell(row=i, column=1, value=name)
|
||
for i, name in enumerate(company_names, start=1):
|
||
dict_ws.cell(row=i, column=2, value=name)
|
||
for i, name in enumerate(department_names, start=1):
|
||
dict_ws.cell(row=i, column=3, value=name)
|
||
|
||
def _range_ref(col_letter: str, count: int) -> str:
|
||
return f"{quote_sheetname(dict_ws.title)}!${col_letter}$1:${col_letter}${count}"
|
||
|
||
if company_names:
|
||
unit_dv = DataValidation(
|
||
type="list",
|
||
formula1=f"={_range_ref('B', len(company_names))}",
|
||
allow_blank=True,
|
||
showErrorMessage=True,
|
||
errorTitle="单位无效",
|
||
error="请从下拉列表中选择单位",
|
||
)
|
||
unit_dv.add("C2:C1048576")
|
||
ws.add_data_validation(unit_dv)
|
||
|
||
if department_names:
|
||
dept_dv = DataValidation(
|
||
type="list",
|
||
formula1=f"={_range_ref('C', len(department_names))}",
|
||
allow_blank=True,
|
||
showErrorMessage=True,
|
||
errorTitle="部门无效",
|
||
error="请从下拉列表中选择部门",
|
||
)
|
||
dept_dv.add("D2:D1048576")
|
||
ws.add_data_validation(dept_dv)
|
||
|
||
if job_title_names:
|
||
job_title_dv = DataValidation(
|
||
type="list",
|
||
formula1=f"={_range_ref('A', len(job_title_names))}",
|
||
allow_blank=True,
|
||
showErrorMessage=True,
|
||
errorTitle="职级无效",
|
||
error="请从下拉列表中选择职级",
|
||
)
|
||
job_title_dv.add("E2:E1048576")
|
||
ws.add_data_validation(job_title_dv)
|
||
|
||
role_dv = DataValidation(
|
||
type="list",
|
||
formula1='"普通员工,业务管理员,系统管理员"',
|
||
allow_blank=True,
|
||
showErrorMessage=True,
|
||
errorTitle="角色无效",
|
||
error="请从下拉列表中选择角色",
|
||
)
|
||
role_dv.add("F2:F1048576")
|
||
ws.add_data_validation(role_dv)
|
||
|
||
status_dv = DataValidation(
|
||
type="list",
|
||
formula1='"启用,停用"',
|
||
allow_blank=True,
|
||
showErrorMessage=True,
|
||
errorTitle="状态无效",
|
||
error="请从下拉列表中选择状态",
|
||
)
|
||
status_dv.add("G2:G1048576")
|
||
ws.add_data_validation(status_dv)
|
||
|
||
for col, width in zip("ABCDEFG", [12, 16, 18, 18, 20, 14, 10]):
|
||
ws.column_dimensions[col].width = width
|
||
|
||
buf = io.BytesIO()
|
||
wb.save(buf)
|
||
buf.seek(0)
|
||
filename = "user_import_template.xlsx"
|
||
return StreamingResponse(
|
||
buf,
|
||
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
||
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
|
||
)
|
||
|
||
|
||
@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))}
|
||
all_units = db.execute(select(Unit)).scalars().all()
|
||
company_by_name = {u.name: u.id for u in all_units if u.category == "company"}
|
||
department_by_name = {u.name: u.id for u in all_units if u.category == "department"}
|
||
status_map = {
|
||
"active": "active", "启用": "active", "正常": "active", "1": "active", "true": "active",
|
||
"inactive": "inactive", "停用": "inactive", "禁用": "inactive", "0": "inactive", "false": "inactive",
|
||
}
|
||
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()
|
||
unit_name = (row.get("unit_name") or "").strip()
|
||
dept_name = (row.get("department_name") or "").strip()
|
||
job_title = (row.get("job_title") or "").strip() or None
|
||
roles_raw = (row.get("role_codes") or "").strip()
|
||
status_raw = (row.get("status") or "").strip().lower()
|
||
if not phone or not u_name:
|
||
errors.append(f"第 {idx} 行：手机号或姓名为空")
|
||
continue
|
||
if phone in existing_phones:
|
||
skipped += 1
|
||
continue
|
||
unit_id = None
|
||
if unit_name:
|
||
unit_id = company_by_name.get(unit_name)
|
||
if unit_id is None:
|
||
errors.append(f"第 {idx} 行：单位「{unit_name}」不存在")
|
||
continue
|
||
department_id = None
|
||
if dept_name:
|
||
department_id = department_by_name.get(dept_name)
|
||
if department_id is None:
|
||
errors.append(f"第 {idx} 行：部门「{dept_name}」不存在")
|
||
continue
|
||
status = status_map.get(status_raw, "active") if status_raw else "active"
|
||
role_alias = {
|
||
"普通员工": "employee", "employee": "employee",
|
||
"业务管理员": "admin", "admin": "admin",
|
||
"系统管理员": "system_admin", "system_admin": "system_admin",
|
||
}
|
||
role_codes = [
|
||
role_alias.get(c.strip(), c.strip())
|
||
for c in roles_raw.replace("，", ",").split(",") if c.strip()
|
||
]
|
||
user = User(
|
||
phone=phone, name=u_name, job_title=job_title,
|
||
unit_id=unit_id, department_id=department_id, status=status,
|
||
)
|
||
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。列：name, phone, unit_name(可选), department_name(可选),
|
||
job_title(可选), role_codes(可选，多个用逗号分隔), status(可选)。"""
|
||
header_map = {
|
||
"手机号": "phone", "phone": "phone", "电话": "phone",
|
||
"姓名": "name", "name": "name", "用户名": "name", "用户名称": "name",
|
||
"单位": "unit_name", "unit": "unit_name", "unit_name": "unit_name",
|
||
"部门": "department_name", "department": "department_name", "department_name": "department_name",
|
||
"职级": "job_title", "岗位": "job_title", "职位": "job_title", "job_title": "job_title",
|
||
"角色": "role_codes", "role_codes": "role_codes", "角色代码": "role_codes",
|
||
"状态": "status", "status": "status",
|
||
}
|
||
|
||
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")
|