294 lines
9.9 KiB
Python
294 lines
9.9 KiB
Python
"""人力资源库：人员维护 + 人员负载。
|
||
|
||
项目 / 定制项目的「负责人」统一取这里的姓名，因此：
|
||
· 改名时同步刷新已有项目的负责人字段；
|
||
· 被项目引用的人不允许直接删除，只能归档（归档后不再出现在下拉里）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||
from pydantic import BaseModel
|
||
from sqlalchemy import func
|
||
from sqlalchemy.orm import Session
|
||
|
||
from .. import constants
|
||
from ..db import get_db
|
||
from ..models import (
|
||
CustomProject,
|
||
Period,
|
||
Project,
|
||
Staff,
|
||
Update,
|
||
join_people,
|
||
people_like,
|
||
split_people,
|
||
)
|
||
|
||
router = APIRouter()
|
||
|
||
|
||
def empty_load() -> dict:
|
||
return {
|
||
"project_count": 0,
|
||
"key_project_count": 0,
|
||
"custom_count": 0,
|
||
"custom_amount": 0.0,
|
||
"custom_pending": 0.0,
|
||
"last_update": None,
|
||
}
|
||
|
||
|
||
class StaffIn(BaseModel):
|
||
name: str
|
||
group: str | None = None
|
||
affiliation: str | None = None
|
||
note: str | None = None
|
||
sort_order: int | None = None
|
||
archived: bool = False
|
||
|
||
|
||
ALLOWED_PATCH = ("name", "group", "affiliation", "note", "sort_order", "archived")
|
||
|
||
|
||
def _workload(db: Session) -> dict[str, dict]:
|
||
"""按姓名汇总在管项目 / 定制项目负载。"""
|
||
load: dict[str, dict] = {}
|
||
|
||
def slot(name: str) -> dict:
|
||
return load.setdefault(name, empty_load())
|
||
|
||
projects = db.query(Project).filter(Project.archived.is_(False)).all()
|
||
for p in projects:
|
||
if not p.owner:
|
||
continue
|
||
s = slot(p.owner)
|
||
s["project_count"] += 1
|
||
if p.is_key:
|
||
s["key_project_count"] += 1
|
||
|
||
# 名下项目的最近一次执行记录（判断是否长期没动静）
|
||
latest_sort: dict[str, int] = {}
|
||
ups = (
|
||
db.query(Update, Project, Period)
|
||
.join(Project, Project.id == Update.project_id)
|
||
.join(Period, Period.id == Update.period_id)
|
||
.order_by(Period.sort_key, Update.id)
|
||
.all()
|
||
)
|
||
for u, proj, period in ups:
|
||
for key in {x for x in (u.owner, proj.owner) if x}:
|
||
latest_sort[key] = period.sort_key
|
||
slot(key)["last_update"] = period.to_dict()["full_label"]
|
||
|
||
for cp in db.query(CustomProject).filter(CustomProject.archived.is_(False)).all():
|
||
if not cp.owner:
|
||
continue
|
||
s = slot(cp.owner)
|
||
s["custom_count"] += 1
|
||
s["custom_amount"] += cp.amount or 0
|
||
s["custom_pending"] += max(0.0, (cp.amount or 0) - (cp.revenue_amount or 0))
|
||
if cp.period_id:
|
||
per = db.get(Period, cp.period_id)
|
||
if per and per.sort_key >= latest_sort.get(cp.owner, 0):
|
||
latest_sort[cp.owner] = per.sort_key
|
||
s["last_update"] = per.to_dict()["full_label"]
|
||
|
||
for s in load.values():
|
||
s["custom_amount"] = round(s["custom_amount"], 2)
|
||
s["custom_pending"] = round(s["custom_pending"], 2)
|
||
return load
|
||
|
||
|
||
@router.get("/staff")
|
||
def list_staff(
|
||
archived: str = Query("false", pattern="^(true|false|all)$"),
|
||
group: str | None = None,
|
||
keyword: str | None = None,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
q = db.query(Staff)
|
||
if archived == "false":
|
||
q = q.filter(Staff.archived.is_(False))
|
||
elif archived == "true":
|
||
q = q.filter(Staff.archived.is_(True))
|
||
if group:
|
||
q = q.filter(Staff.group == group)
|
||
if keyword:
|
||
q = q.filter(Staff.name.contains(keyword) | Staff.note.contains(keyword))
|
||
members = q.order_by(Staff.sort_order, Staff.id).all()
|
||
|
||
load = _workload(db)
|
||
items = []
|
||
for m in members:
|
||
d = m.to_dict()
|
||
d.update(load.get(m.name) or empty_load())
|
||
items.append(d)
|
||
|
||
# 花名册之外的人名（历史数据 / 外部负责人），用于数据核对
|
||
known = {m.name for m in db.query(Staff).all()}
|
||
outside: dict[str, dict] = {}
|
||
for name, cnt in (
|
||
db.query(Project.owner, func.count(Project.id))
|
||
.filter(Project.archived.is_(False), Project.owner.isnot(None), Project.owner != "")
|
||
.group_by(Project.owner)
|
||
.all()
|
||
):
|
||
if name not in known:
|
||
outside.setdefault(name, {"name": name, "project_count": 0, "custom_count": 0})[
|
||
"project_count"
|
||
] = cnt
|
||
for name, cnt in (
|
||
db.query(CustomProject.owner, func.count(CustomProject.id))
|
||
.filter(
|
||
CustomProject.archived.is_(False),
|
||
CustomProject.owner.isnot(None),
|
||
CustomProject.owner != "",
|
||
)
|
||
.group_by(CustomProject.owner)
|
||
.all()
|
||
):
|
||
if name not in known:
|
||
outside.setdefault(name, {"name": name, "project_count": 0, "custom_count": 0})[
|
||
"custom_count"
|
||
] = cnt
|
||
|
||
def dist(field: str, names: list[str]) -> list[dict]:
|
||
counter: dict[str, int] = {}
|
||
for m in members:
|
||
key = getattr(m, field) or "未填写"
|
||
counter[key] = counter.get(key, 0) + 1
|
||
ordered = [k for k in names if k in counter] + sorted(
|
||
k for k in counter if k not in names
|
||
)
|
||
return [{"name": k, "count": counter[k]} for k in ordered]
|
||
|
||
return {
|
||
"items": items,
|
||
"summary": {
|
||
"total": len(items),
|
||
"groups": dist("group", constants.STAFF_GROUPS),
|
||
"affiliations": dist("affiliation", constants.STAFF_AFFILIATIONS),
|
||
"with_project": sum(1 for i in items if i["project_count"] or i["custom_count"]),
|
||
"outside": sorted(
|
||
outside.values(),
|
||
key=lambda x: -(x["project_count"] + x["custom_count"]),
|
||
),
|
||
},
|
||
"dict": {
|
||
"groups": constants.STAFF_GROUPS,
|
||
"affiliations": constants.STAFF_AFFILIATIONS,
|
||
},
|
||
}
|
||
|
||
|
||
@router.post("/staff")
|
||
def create_staff(payload: StaffIn, db: Session = Depends(get_db)):
|
||
name = (payload.name or "").strip()
|
||
if not name:
|
||
raise HTTPException(400, "请填写姓名")
|
||
if db.query(Staff).filter(Staff.name == name).one_or_none():
|
||
raise HTTPException(400, f"「{name}」已在花名册中（可能在归档里），请先归档筛选后恢复")
|
||
sort_order = payload.sort_order
|
||
if sort_order is None:
|
||
sort_order = (db.query(func.max(Staff.sort_order)).scalar() or 0) + 1
|
||
m = Staff(
|
||
name=name,
|
||
group=constants.normalize_staff_group(payload.group),
|
||
affiliation=constants.normalize_staff_affiliation(payload.affiliation),
|
||
note=(payload.note or "").strip() or None,
|
||
sort_order=sort_order,
|
||
archived=payload.archived,
|
||
)
|
||
db.add(m)
|
||
db.commit()
|
||
db.refresh(m)
|
||
return m.to_dict()
|
||
|
||
|
||
@router.patch("/staff/{sid}")
|
||
def patch_staff(sid: int, payload: dict, db: Session = Depends(get_db)):
|
||
"""修改人员信息。改名会把名下项目的负责人一并改过来。"""
|
||
m = db.get(Staff, sid)
|
||
if not m:
|
||
raise HTTPException(404, "人员不存在")
|
||
payload = payload or {}
|
||
old_name = m.name
|
||
new_name = old_name
|
||
for k, v in payload.items():
|
||
if k not in ALLOWED_PATCH:
|
||
continue
|
||
if k == "name":
|
||
v = (str(v or "")).strip()
|
||
if not v:
|
||
raise HTTPException(400, "姓名不能为空")
|
||
dup = db.query(Staff).filter(Staff.name == v, Staff.id != sid).one_or_none()
|
||
if dup:
|
||
raise HTTPException(400, f"花名册中已有「{v}」")
|
||
new_name = v
|
||
elif k == "group":
|
||
v = constants.normalize_staff_group(v)
|
||
elif k == "affiliation":
|
||
v = constants.normalize_staff_affiliation(v)
|
||
setattr(m, k, v)
|
||
db.flush()
|
||
|
||
synced = {"projects": 0, "custom_projects": 0}
|
||
if new_name != old_name:
|
||
synced["projects"] = (
|
||
db.query(Project).filter(Project.owner == old_name).update({Project.owner: new_name}, synchronize_session=False)
|
||
)
|
||
synced["custom_projects"] = (
|
||
db.query(CustomProject)
|
||
.filter(CustomProject.owner == old_name)
|
||
.update({CustomProject.owner: new_name}, synchronize_session=False)
|
||
)
|
||
# 其他负责人 / 团队成员里的人名跟着一起改，否则改名后名单里留旧名字
|
||
for model in (Project, CustomProject):
|
||
rows = (
|
||
db.query(model)
|
||
.filter(
|
||
model.other_owners.like(people_like(old_name))
|
||
| model.members.like(people_like(old_name))
|
||
)
|
||
.all()
|
||
)
|
||
for row in rows:
|
||
row.other_owners = join_people(
|
||
[new_name if x == old_name else x for x in split_people(row.other_owners)]
|
||
)
|
||
row.members = join_people(
|
||
[new_name if x == old_name else x for x in split_people(row.members)]
|
||
)
|
||
db.commit()
|
||
db.refresh(m)
|
||
return {"staff": m.to_dict(), "synced": synced}
|
||
|
||
|
||
@router.delete("/staff/{sid}")
|
||
def delete_staff(sid: int, force: bool = False, db: Session = Depends(get_db)):
|
||
"""删除人员。仍名下有在管项目时需要 force=true，否则提示改用归档。"""
|
||
m = db.get(Staff, sid)
|
||
if not m:
|
||
raise HTTPException(404, "人员不存在")
|
||
n1 = (
|
||
db.query(func.count(Project.id))
|
||
.filter(Project.archived.is_(False), Project.owner == m.name)
|
||
.scalar()
|
||
)
|
||
n2 = (
|
||
db.query(func.count(CustomProject.id))
|
||
.filter(CustomProject.archived.is_(False), CustomProject.owner == m.name)
|
||
.scalar()
|
||
)
|
||
if (n1 or n2) and not force:
|
||
raise HTTPException(
|
||
400,
|
||
f"「{m.name}」名下还有 {n1} 个在管项目 / {n2} 个定制项目。"
|
||
f"离职人员建议直接「归档」（归档后不再出现在负责人下拉里，历史记录保留）；"
|
||
f"确需删除请先转交这些项目。",
|
||
)
|
||
name = m.name
|
||
db.delete(m)
|
||
db.commit()
|
||
return {"ok": True, "name": name}
|