"""工号系统 (employee-number allocation) HTTP surface. Mounted under ``/api/v1/hr/employee-numbers``. Reads admit the HR read gate (global_admin + hr + auditor + management); writes (allocate / import) stay on ``require_hr_admin`` (global_admin + hr). """ from __future__ import annotations import csv import io import logging from datetime import UTC, date, datetime, time from uuid import UUID from fastapi import APIRouter, Depends, File, Query, Request, UploadFile from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.auth.models import AuthUser from app.core.database import get_db from app.core.dependencies import ( get_real_ip, require_hr_admin, require_hr_read, ) from app.core.exceptions import AppException from app.core.rbac import GLOBAL_ADMIN_ROLE, has_role from app.core.response import paginated_response, success_response from app.core.upload import read_upload_bounded from app.domains.hr import employee_number_service as svc from app.domains.hr.employee_number_schemas import ( EmployeeNumberCreateRequest, EmployeeNumberReportResponse, EmployeeNumberResponse, EmployeeNumberUpdateRequest, ImportResultResponse, NextNumberOverrideResponse, NextNumberResponse, SetNextNumberRequest, ) from app.identity.person_models import IdentityPerson logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/v1/hr/employee-numbers", tags=["hr-employee-numbers"]) # Reject uploads larger than this (defence against accidental huge files). _MAX_IMPORT_BYTES = 5 * 1024 * 1024 # 5 MiB _MAX_IMPORT_ROWS = 5_000 _MAX_IMPORT_COLUMNS = 10 _MAX_IMPORT_FIELD_CHARS = 1_024 _CSV_FORMULA_PREFIX = ("=", "+", "-", "@", "\t", "\r") def _parse_import_csv(payload: bytes) -> list[dict[str, str]]: """Decode and collect a CSV without allowing tiny-row amplification.""" try: text = payload.decode("utf-8-sig") except UnicodeDecodeError as exc: raise AppException( code="EMPLOYEE_NUMBER_IMPORT_INVALID_CSV", message="CSV 文件必须使用 UTF-8 编码。", status_code=400, ) from exc try: reader = csv.reader(io.StringIO(text)) headers = next(reader, None) if headers is None: return [] if len(headers) > _MAX_IMPORT_COLUMNS: raise AppException( code="EMPLOYEE_NUMBER_IMPORT_LIMIT_EXCEEDED", message=f"CSV 列数过多，最多 {_MAX_IMPORT_COLUMNS} 列。", status_code=413, ) if any(len(value) > _MAX_IMPORT_FIELD_CHARS for value in headers): raise AppException( code="EMPLOYEE_NUMBER_IMPORT_LIMIT_EXCEEDED", message="CSV 表头字段过长。", status_code=413, ) rows: list[dict[str, str]] = [] for values in reader: # Match DictReader: blank physical lines are ignored. if not values: continue if len(rows) >= _MAX_IMPORT_ROWS: raise AppException( code="EMPLOYEE_NUMBER_IMPORT_LIMIT_EXCEEDED", message=f"CSV 数据行过多，最多 {_MAX_IMPORT_ROWS} 行。", status_code=413, ) if len(values) > _MAX_IMPORT_COLUMNS: raise AppException( code="EMPLOYEE_NUMBER_IMPORT_LIMIT_EXCEEDED", message=f"CSV 列数过多，最多 {_MAX_IMPORT_COLUMNS} 列。", status_code=413, ) if any(len(value) > _MAX_IMPORT_FIELD_CHARS for value in values): raise AppException( code="EMPLOYEE_NUMBER_IMPORT_LIMIT_EXCEEDED", message="CSV 单元格内容过长。", status_code=413, ) rows.append(dict(zip(headers, values, strict=False))) return rows except AppException: raise except csv.Error as exc: raise AppException( code="EMPLOYEE_NUMBER_IMPORT_INVALID_CSV", message="CSV 文件格式无效。", status_code=400, ) from exc def _day_bounds(d_from: date | None, d_to: date | None) -> tuple[datetime | None, datetime | None]: """Inclusive [from 00:00:00, to 23:59:59] UTC datetime bounds. Upper bound uses whole-second 23:59:59 (NOT ``time.max``'s .999999): the ``created_at`` column is ``DATETIME(0)`` (no fractional precision), and MySQL/TiDB ROUND a fractional literal up when comparing against a precision-0 column — ``...59.999999`` would round to next-day 00:00:00 and pull in the following day's rows. """ lo = datetime.combine(d_from, time.min, tzinfo=UTC) if d_from else None hi = datetime.combine(d_to, time(23, 59, 59), tzinfo=UTC) if d_to else None return lo, hi @router.get("") async def list_employee_numbers( request: Request, employee_no: str | None = Query(default=None), name: str | None = Query(default=None), hire_from: date | None = Query(default=None), hire_to: date | None = Query(default=None), source: str | None = Query(default=None), created_from: date | None = Query(default=None), created_to: date | None = Query(default=None), page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=200), _user: AuthUser = Depends(require_hr_read()), db: AsyncSession = Depends(get_db), ): created_lo, created_hi = _day_bounds(created_from, created_to) rows, total = await svc.list_numbers( db, no=employee_no, name=name, hire_from=hire_from, hire_to=hire_to, source=source, created_from=created_lo, created_to=created_hi, page=page, page_size=page_size, ) return paginated_response( data=[EmployeeNumberResponse(**r).model_dump(mode="json") for r in rows], total=total, page=page, page_size=page_size, request=request, ) @router.get("/next") async def get_next_number( request: Request, _user: AuthUser = Depends(require_hr_read()), db: AsyncSession = Depends(get_db), ): next_no, next_seq, max_no = await svc.get_next(db) return success_response( data=NextNumberResponse( next_employee_no=next_no, next_seq=next_seq, current_max_no=max_no, ).model_dump(mode="json"), request=request, ) @router.post("/next") async def set_next_number( request: Request, body: SetNextNumberRequest, current_user: AuthUser = Depends(require_hr_admin()), db: AsyncSession = Depends(get_db), ): """Manually set the next 工号 (工号维护). Records the operator + reason.""" row = await svc.set_next( db, next_employee_no=body.next_employee_no, reason=body.reason, created_by=current_user.id, ) await AuditService(db).log( trace_id=request.state.trace_id, actor_type="user", actor_user_id=current_user.id, action="hr.employee_number.set_next", resource_type="hr_employee_number_override", resource_id=str(row.id), status="success", detail={ "next_employee_no": row.next_employee_no, "previous_next_employee_no": row.previous_next_employee_no, "reason": row.reason, }, ip_address=get_real_ip(request), user_agent=request.headers.get("User-Agent"), ) await db.commit() await db.refresh(row) return success_response( data=NextNumberOverrideResponse( id=row.id, next_employee_no=row.next_employee_no, next_seq=row.next_seq, previous_next_employee_no=row.previous_next_employee_no, previous_next_seq=row.previous_next_seq, reason=row.reason, created_by=row.created_by, created_by_name=current_user.display_name, created_at=row.created_at, ).model_dump(mode="json"), request=request, ) @router.get("/next/history") async def list_next_overrides( request: Request, page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=200), _user: AuthUser = Depends(require_hr_read()), db: AsyncSession = Depends(get_db), ): """工号维护记录 — manual next-number changes, newest first.""" rows, total = await svc.list_overrides(db, page=page, page_size=page_size) return paginated_response( data=[NextNumberOverrideResponse(**r).model_dump(mode="json") for r in rows], total=total, page=page, page_size=page_size, request=request, ) @router.get("/report") async def get_report( request: Request, _user: AuthUser = Depends(require_hr_read()), db: AsyncSession = Depends(get_db), ): report = await svc.build_report(db) return success_response( data=EmployeeNumberReportResponse(**report).model_dump(mode="json"), request=request, ) @router.post("") async def create_employee_number( request: Request, body: EmployeeNumberCreateRequest, current_user: AuthUser = Depends(require_hr_admin()), db: AsyncSession = Depends(get_db), ): bu, bu_ok = svc.resolve_bu(body.bu, await svc.known_bu_map(db)) if not bu_ok: raise AppException( code="UNKNOWN_BU", message="未知的一级部门(BU),请从列表中选择。", status_code=400, ) row = await svc.allocate( db, name=body.name, hire_date=body.hire_date, remark=body.remark, bu=bu, created_by=current_user.id, source="manual", ) await AuditService(db).log( trace_id=request.state.trace_id, actor_type="user", actor_user_id=current_user.id, action="hr.employee_number.allocate", resource_type="hr_employee_number", resource_id=str(row.id), status="success", detail={"employee_no": row.employee_no, "name": row.name, "bu": row.bu}, ip_address=get_real_ip(request), user_agent=request.headers.get("User-Agent"), ) await db.commit() await db.refresh(row) return success_response( data=EmployeeNumberResponse( id=row.id, employee_no=row.employee_no, seq=row.seq, name=row.name, hire_date=row.hire_date, remark=row.remark, bu=row.bu, source=row.source, created_by=row.created_by, created_by_name=current_user.display_name, created_at=row.created_at, ).model_dump(mode="json"), request=request, ) @router.patch("/{number_id}") async def update_employee_number( request: Request, number_id: UUID, body: EmployeeNumberUpdateRequest, current_user: AuthUser = Depends(require_hr_admin()), db: AsyncSession = Depends(get_db), ): """维护 a 工号's descriptive fields (the number itself is immutable). Only the row's creator — or a ``global_admin`` — may edit it; a non-owner HR user gets 403 from the service layer.""" bu, bu_ok = svc.resolve_bu(body.bu, await svc.known_bu_map(db)) if not bu_ok: raise AppException( code="UNKNOWN_BU", message="未知的一级部门(BU),请从列表中选择。", status_code=400, ) is_admin = has_role(current_user.role_names, GLOBAL_ADMIN_ROLE) row, before, after = await svc.update_number( db, number_id=number_id, actor_id=current_user.id, is_admin=is_admin, name=body.name, hire_date=body.hire_date, remark=body.remark, bu=bu, ) await AuditService(db).log( trace_id=request.state.trace_id, actor_type="user", actor_user_id=current_user.id, action="hr.employee_number.update", resource_type="hr_employee_number", resource_id=str(row.id), status="success", detail={"employee_no": row.employee_no, "before": before, "after": after}, ip_address=get_real_ip(request), user_agent=request.headers.get("User-Agent"), ) await db.commit() await db.refresh(row) # created_by_name reflects the CREATOR (an admin may edit someone else's # row), so resolve it rather than reusing the editor's display name. creator_name = ( await db.scalar( select(IdentityPerson.display_name).where( IdentityPerson.id == row.created_by ) ) if row.created_by else None ) return success_response( data=EmployeeNumberResponse( id=row.id, employee_no=row.employee_no, seq=row.seq, name=row.name, hire_date=row.hire_date, remark=row.remark, bu=row.bu, source=row.source, created_by=row.created_by, created_by_name=creator_name, created_at=row.created_at, ).model_dump(mode="json"), request=request, ) @router.delete("/{number_id}") async def delete_employee_number( request: Request, number_id: UUID, current_user: AuthUser = Depends(require_hr_admin()), db: AsyncSession = Depends(get_db), ): """删除 a 工号 (hard delete). Frees the number for reclamation: if the deleted number was the current maximum, the next allocation hands it out again; a gap below a still-existing higher number is left as-is. Only the row's creator — or a ``global_admin`` — may delete it.""" is_admin = has_role(current_user.role_names, GLOBAL_ADMIN_ROLE) info = await svc.delete_number( db, number_id=number_id, actor_id=current_user.id, is_admin=is_admin, ) await AuditService(db).log( trace_id=request.state.trace_id, actor_type="user", actor_user_id=current_user.id, action="hr.employee_number.delete", resource_type="hr_employee_number", resource_id=str(number_id), status="success", detail=info, ip_address=get_real_ip(request), user_agent=request.headers.get("User-Agent"), ) await db.commit() return success_response( data={"employee_no": info["employee_no"]}, request=request, ) @router.post("/import") async def import_employee_numbers( request: Request, file: UploadFile = File(...), current_user: AuthUser = Depends(require_hr_admin()), db: AsyncSession = Depends(get_db), ): filename = (file.filename or "").lower() if not filename.endswith(".csv"): await file.close() raise AppException( code="UNSUPPORTED_FILE", message="仅支持 CSV 文件(列:工号,姓名,入职日期,备注,一级部门)。", status_code=400, ) # Bounded read: pull at most cap+1 bytes so an oversized upload is # rejected WITHOUT buffering the whole file (or its decoded/parsed # copies) in memory. payload = await read_upload_bounded( file, _MAX_IMPORT_BYTES, code="FILE_TOO_LARGE", message="文件过大,请控制在 5MB 以内。", close=True, ) rows = _parse_import_csv(payload) result = await svc.import_rows(db, rows, created_by=current_user.id) await AuditService(db).log( trace_id=request.state.trace_id, actor_type="user", actor_user_id=current_user.id, action="hr.employee_number.import", resource_type="hr_employee_number", resource_id=None, status="success", detail={ "filename": file.filename, "total": result["total"], "created": result["created"], "skipped": result["skipped"], }, ip_address=get_real_ip(request), user_agent=request.headers.get("User-Agent"), ) await db.commit() return success_response( data=ImportResultResponse(**result).model_dump(mode="json"), request=request, ) @router.get("/export") async def export_employee_numbers( employee_no: str | None = Query(default=None), name: str | None = Query(default=None), hire_from: date | None = Query(default=None), hire_to: date | None = Query(default=None), source: str | None = Query(default=None), created_from: date | None = Query(default=None), created_to: date | None = Query(default=None), _user: AuthUser = Depends(require_hr_read()), db: AsyncSession = Depends(get_db), ): from fastapi.responses import StreamingResponse created_lo, created_hi = _day_bounds(created_from, created_to) rows, _total = await svc.list_numbers( db, no=employee_no, name=name, hire_from=hire_from, hire_to=hire_to, source=source, created_from=created_lo, created_to=created_hi, page=1, page_size=1_000_000, ) def _cell(v: object) -> str: if v is None: return "" s = v.isoformat() if isinstance(v, (date, datetime)) else str(v) # CSV-injection guard (OWASP): neutralise leading formula chars. if s.startswith(_CSV_FORMULA_PREFIX): s = "'" + s return s buf = io.StringIO() buf.write("") # BOM so Excel reads UTF-8 (Chinese) correctly. writer = csv.writer(buf) writer.writerow(svc.CSV_COLUMNS) for r in rows: writer.writerow([ _cell(r["employee_no"]), _cell(r["name"]), _cell(r["hire_date"]), _cell(r["remark"]), _cell(r["bu"]), ]) buf.seek(0) filename = f"employee_numbers_{datetime.now(UTC).strftime('%Y%m%d_%H%M%S')}.csv" return StreamingResponse( iter([buf.getvalue()]), media_type="text/csv; charset=utf-8", headers={"Content-Disposition": f'attachment; filename="{filename}"'}, )