164 lines
6.9 KiB
Python
164 lines
6.9 KiB
Python
from fastapi import APIRouter, Depends, File, HTTPException, Query, UploadFile, status
|
|
from fastapi.responses import Response
|
|
from sqlalchemy import func, or_, select
|
|
from sqlalchemy.orm import Session, selectinload
|
|
|
|
from app.database import get_db
|
|
from app.deps import require_roles
|
|
from app.models import PersonAttendancePoint, PersonRole, Personnel, Role
|
|
from app.schemas import ImportResult, PageResponse, PersonnelCreate, PersonnelOut
|
|
from app.services.common import ROLE_NAMES
|
|
from app.services.attendance_points import accessible_point_names, set_person_attendance_points
|
|
from app.services.excel_export import export_people
|
|
from app.services.excel_import import ProductImportValidationError, import_people
|
|
from app.services.serializers import personnel_out, personnel_summary_out
|
|
|
|
router = APIRouter(prefix="/api/people", tags=["people"])
|
|
|
|
|
|
@router.get("", response_model=PageResponse)
|
|
def list_people(
|
|
keyword: str = "",
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(10, ge=1, le=100),
|
|
user: Personnel = Depends(require_roles(Role.admin, Role.manager)),
|
|
db: Session = Depends(get_db),
|
|
) -> PageResponse:
|
|
point_names = accessible_point_names(db, user)
|
|
query = (
|
|
select(Personnel)
|
|
.options(selectinload(Personnel.roles), selectinload(Personnel.attendance_points).selectinload(PersonAttendancePoint.attendance_point))
|
|
.where(Personnel.roles.any())
|
|
.order_by(Personnel.created_at.desc())
|
|
)
|
|
if user.role == Role.admin:
|
|
query = query.where(
|
|
Personnel.roles.any(PersonRole.role == Role.worker),
|
|
Personnel.attendance_points.any(PersonAttendancePoint.attendance_point_name.in_(point_names)),
|
|
)
|
|
if keyword:
|
|
like = f"%{keyword}%"
|
|
matched_roles = [role for role, name in ROLE_NAMES.items() if keyword in name or keyword in role]
|
|
query = query.where(
|
|
or_(
|
|
Personnel.phone.like(like),
|
|
Personnel.name.like(like),
|
|
Personnel.roles.any(PersonRole.role.in_(matched_roles)),
|
|
)
|
|
)
|
|
safe_page = max(1, page)
|
|
safe_page_size = min(100, max(1, page_size))
|
|
total = db.scalar(select(func.count()).select_from(query.order_by(None).subquery())) or 0
|
|
rows = db.scalars(query.limit(safe_page_size).offset((safe_page - 1) * safe_page_size)).all()
|
|
return PageResponse(
|
|
page=safe_page,
|
|
page_size=safe_page_size,
|
|
total=total,
|
|
total_pages=max(1, (total + safe_page_size - 1) // safe_page_size),
|
|
rows=[personnel_summary_out(row) for row in rows],
|
|
)
|
|
|
|
|
|
@router.post("", response_model=PersonnelOut)
|
|
def save_person(
|
|
payload: PersonnelCreate,
|
|
user: Personnel = Depends(require_roles(Role.admin, Role.manager)),
|
|
db: Session = Depends(get_db),
|
|
) -> PersonnelOut:
|
|
if user.role == Role.admin and payload.role != Role.worker:
|
|
raise HTTPException(status_code=403, detail="管理员只能新增或导入冲压工人")
|
|
point_names = payload.attendance_point_names or accessible_point_names(db, user)
|
|
if user.role == Role.admin:
|
|
allowed = set(accessible_point_names(db, user))
|
|
denied = [name for name in point_names if name not in allowed]
|
|
if denied:
|
|
raise HTTPException(status_code=403, detail=f"无该考勤点人员维护权限：{'、'.join(denied)}")
|
|
phone = str(payload.phone or "").strip()
|
|
original_phone = str(payload.original_phone or "").strip()
|
|
if original_phone and original_phone != phone:
|
|
raise HTTPException(status_code=400, detail="编辑人员不能修改手机号")
|
|
|
|
person = db.get(Personnel, phone)
|
|
editing_existing = bool(original_phone)
|
|
active_role_count = db.scalar(select(func.count()).select_from(PersonRole).where(PersonRole.phone == phone)) or 0
|
|
if person is not None and not editing_existing and active_role_count > 0:
|
|
raise HTTPException(status_code=409, detail="该手机号已存在，不能重复添加")
|
|
if person is None:
|
|
person = Personnel(phone=phone, name=payload.name)
|
|
db.add(person)
|
|
db.flush()
|
|
else:
|
|
person.name = payload.name
|
|
person.is_temporary = False
|
|
person.temporary_expires_at = None
|
|
role = db.get(PersonRole, {"phone": phone, "role": payload.role})
|
|
if role is None:
|
|
role = PersonRole(phone=phone, role=payload.role)
|
|
db.add(role)
|
|
try:
|
|
set_person_attendance_points(db, phone, point_names)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
db.commit()
|
|
person = db.scalar(
|
|
select(Personnel)
|
|
.options(selectinload(Personnel.roles), selectinload(Personnel.attendance_points).selectinload(PersonAttendancePoint.attendance_point))
|
|
.where(Personnel.phone == phone)
|
|
)
|
|
if person is None:
|
|
raise HTTPException(status_code=500, detail="人员保存失败")
|
|
return personnel_out(person, payload.role)
|
|
|
|
|
|
@router.delete("/{phone}", status_code=204)
|
|
def delete_person(
|
|
phone: str,
|
|
role: Role | None = None,
|
|
user: Personnel = Depends(require_roles(Role.admin, Role.manager)),
|
|
db: Session = Depends(get_db),
|
|
) -> None:
|
|
person = db.get(Personnel, phone)
|
|
if person is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="人员不存在")
|
|
if user.role == Role.admin:
|
|
point_names = accessible_point_names(db, user)
|
|
if not any(item.attendance_point_name in point_names for item in person.attendance_points):
|
|
raise HTTPException(status_code=403, detail="无该考勤点人员维护权限")
|
|
if role is not None and role != Role.worker:
|
|
raise HTTPException(status_code=403, detail="管理员只能删除冲压工人")
|
|
if role is None:
|
|
db.query(PersonRole).filter(PersonRole.phone == phone).delete()
|
|
else:
|
|
person_role = db.get(PersonRole, {"phone": phone, "role": role})
|
|
if person_role is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="角色不存在")
|
|
db.delete(person_role)
|
|
db.commit()
|
|
|
|
|
|
@router.post("/import", response_model=ImportResult)
|
|
async def import_people_excel(
|
|
file: UploadFile = File(...),
|
|
user: Personnel = Depends(require_roles(Role.admin, Role.manager)),
|
|
db: Session = Depends(get_db),
|
|
) -> ImportResult:
|
|
content = await file.read()
|
|
try:
|
|
result = import_people(db, content, user=user)
|
|
except ProductImportValidationError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
return ImportResult(**result.as_dict())
|
|
|
|
|
|
@router.get("/export")
|
|
def export_people_excel(
|
|
user: Personnel = Depends(require_roles(Role.admin, Role.manager)),
|
|
db: Session = Depends(get_db),
|
|
) -> Response:
|
|
content = export_people(db, user=user)
|
|
return Response(
|
|
content=content,
|
|
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
headers={"Content-Disposition": 'attachment; filename="people.xlsx"'},
|
|
)
|