399 lines
16 KiB
Python
399 lines
16 KiB
Python
|
"""Departments and their workflows, users, sites and categories.
|
||
|
|
|
||
|
|
Reading is open to every signed-in user (except the user list). Departments, workflows, users and
|
||
|
|
sites are managed by administrators; a department's categories also by that department's approvers.
|
||
|
|
"""
|
||
|
|
from fastapi import APIRouter, Depends, HTTPException
|
||
|
|
from fastapi.responses import Response
|
||
|
|
from pydantic import BaseModel, Field
|
||
|
|
from sqlalchemy import func, select
|
||
|
|
from sqlalchemy.orm import Session
|
||
|
|
|
||
|
|
from .. import mfa
|
||
|
|
from ..audit import diff, record, snapshot
|
||
|
|
from ..bootstrap import seed_categories
|
||
|
|
from ..db import get_db
|
||
|
|
from ..models import BudgetYear, Category, CostLine, Department, Site, User, WorkflowStep
|
||
|
|
from ..permissions import is_admin, member
|
||
|
|
from ..security import ROLES, current_user, hash_password, require
|
||
|
|
from ..serializers import category_out, department_out, site_out, user_out
|
||
|
|
from ..summary import build_year_view
|
||
|
|
from ..template import build_template
|
||
|
|
|
||
|
|
router = APIRouter(prefix="/api", tags=["admin"])
|
||
|
|
|
||
|
|
|
||
|
|
class UserIn(BaseModel):
|
||
|
|
department_id: int | None = None
|
||
|
|
email: str | None = None
|
||
|
|
name: str | None = None
|
||
|
|
role: str | None = None
|
||
|
|
language: str | None = None
|
||
|
|
active: bool | None = None
|
||
|
|
password: str | None = None
|
||
|
|
|
||
|
|
|
||
|
|
@router.get("/users")
|
||
|
|
def list_users(db: Session = Depends(get_db), _: User = Depends(require("admin"))):
|
||
|
|
return [user_out(u) for u in db.scalars(select(User).order_by(User.name))]
|
||
|
|
|
||
|
|
|
||
|
|
def _apply_user(u: User, body: UserIn, db: Session):
|
||
|
|
if "department_id" in body.model_fields_set:
|
||
|
|
if body.department_id is not None and not db.get(Department, body.department_id):
|
||
|
|
raise HTTPException(400, "invalid_department")
|
||
|
|
u.department_id = body.department_id
|
||
|
|
if body.email is not None:
|
||
|
|
email = body.email.strip().lower()
|
||
|
|
if "@" not in email:
|
||
|
|
raise HTTPException(400, "invalid_email")
|
||
|
|
clash = db.scalar(select(User).where(func.lower(User.email) == email, User.id != (u.id or 0)))
|
||
|
|
if clash:
|
||
|
|
raise HTTPException(400, "email_taken")
|
||
|
|
u.email = email
|
||
|
|
if body.name is not None:
|
||
|
|
u.name = body.name.strip()
|
||
|
|
if body.role is not None:
|
||
|
|
if body.role not in ROLES:
|
||
|
|
raise HTTPException(400, "invalid_role")
|
||
|
|
u.role = body.role
|
||
|
|
if body.language in ("pt", "en"):
|
||
|
|
u.language = body.language
|
||
|
|
if body.active is not None:
|
||
|
|
u.active = body.active
|
||
|
|
if body.password:
|
||
|
|
if len(body.password) < 8:
|
||
|
|
raise HTTPException(400, "password_too_short")
|
||
|
|
u.password_hash = hash_password(body.password)
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/users")
|
||
|
|
def create_user(body: UserIn, db: Session = Depends(get_db), admin: User = Depends(require("admin"))):
|
||
|
|
if not body.email or not body.name or not body.password:
|
||
|
|
raise HTTPException(400, "missing_fields")
|
||
|
|
u = User(role="viewer", language="pt", active=True, password_hash="")
|
||
|
|
_apply_user(u, body, db)
|
||
|
|
db.add(u)
|
||
|
|
db.flush()
|
||
|
|
record(db, admin, "user", u.id, "create", f"{u.name} <{u.email}> as {u.role}")
|
||
|
|
db.commit()
|
||
|
|
return user_out(u)
|
||
|
|
|
||
|
|
|
||
|
|
@router.put("/users/{user_id}")
|
||
|
|
def update_user(user_id: int, body: UserIn, db: Session = Depends(get_db), admin: User = Depends(require("admin"))):
|
||
|
|
u = db.get(User, user_id)
|
||
|
|
if not u:
|
||
|
|
raise HTTPException(404, "not_found")
|
||
|
|
if u.id == admin.id and (body.role not in (None, "admin") or body.active is False):
|
||
|
|
raise HTTPException(400, "cannot_demote_self")
|
||
|
|
fields = ["email", "name", "role", "active", "department_id"]
|
||
|
|
before = snapshot(u, fields)
|
||
|
|
_apply_user(u, body, db)
|
||
|
|
changes = diff(before, snapshot(u, fields))
|
||
|
|
if body.password:
|
||
|
|
changes["password"] = ["", "reset"]
|
||
|
|
record(db, admin, "user", u.id, "update", u.name, changes)
|
||
|
|
db.commit()
|
||
|
|
return user_out(u)
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/users/{user_id}/mfa/reset")
|
||
|
|
def reset_user_mfa(user_id: int, db: Session = Depends(get_db), admin: User = Depends(require("admin"))):
|
||
|
|
"""For a user who lost their phone: MFA goes off and they can set it up again after signing in."""
|
||
|
|
u = db.get(User, user_id)
|
||
|
|
if not u:
|
||
|
|
raise HTTPException(404, "not_found")
|
||
|
|
mfa.disable(u)
|
||
|
|
record(db, admin, "user", u.id, "mfa_reset", u.name)
|
||
|
|
db.commit()
|
||
|
|
return user_out(u)
|
||
|
|
|
||
|
|
|
||
|
|
@router.get("/users/{user_id}/avatar")
|
||
|
|
def user_avatar(user_id: int, db: Session = Depends(get_db), _: User = Depends(current_user)):
|
||
|
|
u = db.get(User, user_id)
|
||
|
|
if not u or not u.avatar:
|
||
|
|
raise HTTPException(404, "not_found")
|
||
|
|
return Response(u.avatar, media_type=u.avatar_type or "image/jpeg",
|
||
|
|
headers={"Cache-Control": "private, max-age=31536000, immutable"})
|
||
|
|
|
||
|
|
|
||
|
|
class SiteIn(BaseModel):
|
||
|
|
name: str | None = Field(default=None, max_length=120)
|
||
|
|
active: bool | None = None
|
||
|
|
|
||
|
|
|
||
|
|
@router.get("/sites")
|
||
|
|
def list_sites(db: Session = Depends(get_db), _: User = Depends(current_user)):
|
||
|
|
return [site_out(s) for s in db.scalars(select(Site).order_by(Site.name))]
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/sites")
|
||
|
|
def create_site(body: SiteIn, db: Session = Depends(get_db), admin: User = Depends(require("admin"))):
|
||
|
|
if not body.name or not body.name.strip():
|
||
|
|
raise HTTPException(400, "missing_fields")
|
||
|
|
if db.scalar(select(Site).where(func.lower(Site.name) == body.name.strip().lower())):
|
||
|
|
raise HTTPException(400, "name_taken")
|
||
|
|
s = Site(name=body.name.strip(), active=True)
|
||
|
|
db.add(s)
|
||
|
|
db.flush()
|
||
|
|
record(db, admin, "site", s.id, "create", s.name)
|
||
|
|
db.commit()
|
||
|
|
return site_out(s)
|
||
|
|
|
||
|
|
|
||
|
|
@router.put("/sites/{site_id}")
|
||
|
|
def update_site(site_id: int, body: SiteIn, db: Session = Depends(get_db), admin: User = Depends(require("admin"))):
|
||
|
|
s = db.get(Site, site_id)
|
||
|
|
if not s:
|
||
|
|
raise HTTPException(404, "not_found")
|
||
|
|
if body.name:
|
||
|
|
s.name = body.name.strip()
|
||
|
|
if body.active is not None:
|
||
|
|
s.active = body.active
|
||
|
|
record(db, admin, "site", s.id, "update", s.name)
|
||
|
|
db.commit()
|
||
|
|
return site_out(s)
|
||
|
|
|
||
|
|
|
||
|
|
class CategoryIn(BaseModel):
|
||
|
|
department_id: int | None = None
|
||
|
|
name_en: str | None = Field(default=None, max_length=200)
|
||
|
|
name_pt: str | None = Field(default=None, max_length=200)
|
||
|
|
section: str | None = None
|
||
|
|
sort_order: int | None = None
|
||
|
|
show_when_empty: bool | None = None
|
||
|
|
active: bool | None = None
|
||
|
|
|
||
|
|
|
||
|
|
def can_manage_categories(user: User, department_id: int) -> bool:
|
||
|
|
return is_admin(user) or (member(user, department_id) and ROLES.index(user.role) >= ROLES.index("approver"))
|
||
|
|
|
||
|
|
|
||
|
|
def _require_categories(user: User, department_id: int) -> None:
|
||
|
|
if not can_manage_categories(user, department_id):
|
||
|
|
raise HTTPException(403, "forbidden")
|
||
|
|
|
||
|
|
|
||
|
|
@router.get("/categories")
|
||
|
|
def list_categories(department_id: int | None = None, db: Session = Depends(get_db), _: User = Depends(current_user)):
|
||
|
|
q = select(Category).order_by(Category.department_id, Category.sort_order, Category.id)
|
||
|
|
if department_id:
|
||
|
|
q = q.where(Category.department_id == department_id)
|
||
|
|
return [category_out(c) for c in db.scalars(q)]
|
||
|
|
|
||
|
|
|
||
|
|
def _apply_category(c: Category, body: CategoryIn, db: Session):
|
||
|
|
if body.name_en is not None:
|
||
|
|
name = body.name_en.strip()
|
||
|
|
if not name:
|
||
|
|
raise HTTPException(400, "missing_fields")
|
||
|
|
if db.scalar(select(Category).where(Category.department_id == c.department_id,
|
||
|
|
func.lower(Category.name_en) == name.lower(), Category.id != (c.id or 0))):
|
||
|
|
raise HTTPException(400, "name_taken")
|
||
|
|
c.name_en = name
|
||
|
|
if body.name_pt is not None:
|
||
|
|
c.name_pt = body.name_pt.strip()
|
||
|
|
if body.section is not None:
|
||
|
|
if body.section not in ("cost", "revenue"):
|
||
|
|
raise HTTPException(400, "invalid_section")
|
||
|
|
c.section = body.section
|
||
|
|
for f in ("sort_order", "show_when_empty", "active"):
|
||
|
|
if getattr(body, f) is not None:
|
||
|
|
setattr(c, f, getattr(body, f))
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/categories")
|
||
|
|
def create_category(body: CategoryIn, db: Session = Depends(get_db), admin: User = Depends(current_user)):
|
||
|
|
if not body.name_en or not body.department_id or not db.get(Department, body.department_id):
|
||
|
|
raise HTTPException(400, "missing_fields")
|
||
|
|
_require_categories(admin, body.department_id)
|
||
|
|
top = db.scalar(select(func.max(Category.sort_order)).where(Category.department_id == body.department_id))
|
||
|
|
c = Category(department_id=body.department_id, section="cost", name_pt="", show_when_empty=True, active=True,
|
||
|
|
sort_order=(top or 0) + 10)
|
||
|
|
_apply_category(c, body, db)
|
||
|
|
db.add(c)
|
||
|
|
db.flush()
|
||
|
|
record(db, admin, "category", c.id, "create", c.name_en)
|
||
|
|
db.commit()
|
||
|
|
return category_out(c)
|
||
|
|
|
||
|
|
|
||
|
|
@router.put("/categories/{cat_id}")
|
||
|
|
def update_category(cat_id: int, body: CategoryIn, db: Session = Depends(get_db), admin: User = Depends(current_user)):
|
||
|
|
c = db.get(Category, cat_id)
|
||
|
|
if not c:
|
||
|
|
raise HTTPException(404, "not_found")
|
||
|
|
_require_categories(admin, c.department_id)
|
||
|
|
_apply_category(c, body, db)
|
||
|
|
record(db, admin, "category", c.id, "update", c.name_en)
|
||
|
|
db.commit()
|
||
|
|
return category_out(c)
|
||
|
|
|
||
|
|
|
||
|
|
@router.delete("/categories/{cat_id}")
|
||
|
|
def delete_category(cat_id: int, db: Session = Depends(get_db), admin: User = Depends(current_user)):
|
||
|
|
c = db.get(Category, cat_id)
|
||
|
|
if not c:
|
||
|
|
raise HTTPException(404, "not_found")
|
||
|
|
_require_categories(admin, c.department_id)
|
||
|
|
if db.scalar(select(func.count(CostLine.id)).where(CostLine.category_id == cat_id)):
|
||
|
|
raise HTTPException(400, "category_in_use")
|
||
|
|
record(db, admin, "category", c.id, "delete", c.name_en)
|
||
|
|
db.delete(c)
|
||
|
|
db.commit()
|
||
|
|
return {"ok": True}
|
||
|
|
|
||
|
|
|
||
|
|
class DepartmentIn(BaseModel):
|
||
|
|
name: str | None = Field(default=None, max_length=120)
|
||
|
|
active: bool | None = None
|
||
|
|
# where the new department's categories come from: "defaults", "none" or another department's id
|
||
|
|
categories_from: str | int | None = "defaults"
|
||
|
|
|
||
|
|
|
||
|
|
@router.get("/departments")
|
||
|
|
def list_departments(db: Session = Depends(get_db), _: User = Depends(current_user)):
|
||
|
|
return [department_out(d) for d in db.scalars(select(Department).order_by(Department.name))]
|
||
|
|
|
||
|
|
|
||
|
|
def _copy_categories(db: Session, target: Department, source) -> None:
|
||
|
|
if source in (None, "none", ""):
|
||
|
|
return
|
||
|
|
if source == "defaults":
|
||
|
|
seed_categories(db, target)
|
||
|
|
return
|
||
|
|
src = db.get(Department, int(source))
|
||
|
|
if not src:
|
||
|
|
raise HTTPException(400, "invalid_department")
|
||
|
|
existing = {n.lower() for n in db.scalars(select(Category.name_en).where(Category.department_id == target.id))}
|
||
|
|
for c in db.scalars(select(Category).where(Category.department_id == src.id).order_by(Category.sort_order)):
|
||
|
|
if c.name_en.lower() not in existing:
|
||
|
|
db.add(Category(department_id=target.id, name_en=c.name_en, name_pt=c.name_pt, section=c.section,
|
||
|
|
sort_order=c.sort_order, show_when_empty=c.show_when_empty, active=c.active))
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/departments")
|
||
|
|
def create_department(body: DepartmentIn, db: Session = Depends(get_db), admin: User = Depends(require("admin"))):
|
||
|
|
name = (body.name or "").strip()
|
||
|
|
if not name:
|
||
|
|
raise HTTPException(400, "missing_fields")
|
||
|
|
if db.scalar(select(Department).where(func.lower(Department.name) == name.lower())):
|
||
|
|
raise HTTPException(400, "name_taken")
|
||
|
|
d = Department(name=name, active=True)
|
||
|
|
db.add(d)
|
||
|
|
db.flush()
|
||
|
|
_copy_categories(db, d, body.categories_from)
|
||
|
|
record(db, admin, "department", d.id, "create", d.name)
|
||
|
|
db.commit()
|
||
|
|
db.refresh(d)
|
||
|
|
return department_out(d)
|
||
|
|
|
||
|
|
|
||
|
|
@router.put("/departments/{dept_id}")
|
||
|
|
def update_department(dept_id: int, body: DepartmentIn, db: Session = Depends(get_db),
|
||
|
|
admin: User = Depends(require("admin"))):
|
||
|
|
d = db.get(Department, dept_id)
|
||
|
|
if not d:
|
||
|
|
raise HTTPException(404, "not_found")
|
||
|
|
if body.name and body.name.strip() != d.name:
|
||
|
|
if db.scalar(select(Department).where(func.lower(Department.name) == body.name.strip().lower())):
|
||
|
|
raise HTTPException(400, "name_taken")
|
||
|
|
d.name = body.name.strip()
|
||
|
|
if body.active is not None:
|
||
|
|
d.active = body.active
|
||
|
|
record(db, admin, "department", d.id, "update", d.name)
|
||
|
|
db.commit()
|
||
|
|
return department_out(d)
|
||
|
|
|
||
|
|
|
||
|
|
class CopyCategoriesIn(BaseModel):
|
||
|
|
source: str | int
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/departments/{dept_id}/categories/copy")
|
||
|
|
def copy_categories(dept_id: int, body: CopyCategoriesIn, db: Session = Depends(get_db),
|
||
|
|
user: User = Depends(current_user)):
|
||
|
|
d = db.get(Department, dept_id)
|
||
|
|
if not d:
|
||
|
|
raise HTTPException(404, "not_found")
|
||
|
|
_require_categories(user, d.id)
|
||
|
|
if body.source == "defaults":
|
||
|
|
existing = {n.lower() for n in db.scalars(select(Category.name_en).where(Category.department_id == d.id))}
|
||
|
|
from ..seed import CATEGORIES
|
||
|
|
top = db.scalar(select(func.max(Category.sort_order)).where(Category.department_id == d.id)) or 0
|
||
|
|
for i, (en, pt, section) in enumerate(CATEGORIES):
|
||
|
|
if en.lower() not in existing:
|
||
|
|
db.add(Category(department_id=d.id, name_en=en, name_pt=pt, section=section, sort_order=top + (i + 1) * 10))
|
||
|
|
else:
|
||
|
|
_copy_categories(db, d, body.source)
|
||
|
|
record(db, user, "department", d.id, "copy_categories", f"from {body.source}")
|
||
|
|
db.commit()
|
||
|
|
return {"ok": True}
|
||
|
|
|
||
|
|
|
||
|
|
class StepIn(BaseModel):
|
||
|
|
name: str = Field(max_length=120)
|
||
|
|
approver_mode: str = "role"
|
||
|
|
approver_ids: list[int] = []
|
||
|
|
|
||
|
|
|
||
|
|
class WorkflowIn(BaseModel):
|
||
|
|
steps: list[StepIn]
|
||
|
|
|
||
|
|
|
||
|
|
@router.put("/departments/{dept_id}/workflow")
|
||
|
|
def set_workflow(dept_id: int, body: WorkflowIn, db: Session = Depends(get_db), admin: User = Depends(require("admin"))):
|
||
|
|
d = db.get(Department, dept_id)
|
||
|
|
if not d:
|
||
|
|
raise HTTPException(404, "not_found")
|
||
|
|
if db.scalar(select(func.count(BudgetYear.id)).where(BudgetYear.department_id == d.id,
|
||
|
|
BudgetYear.status == "in_review")):
|
||
|
|
raise HTTPException(409, "workflow_in_use")
|
||
|
|
if len(body.steps) > 10:
|
||
|
|
raise HTTPException(400, "too_many_steps")
|
||
|
|
valid_users = set(db.scalars(select(User.id).where(User.role != "management"))) # Upper Management never approves
|
||
|
|
before = [s.name for s in d.steps]
|
||
|
|
d.steps.clear()
|
||
|
|
db.flush()
|
||
|
|
for i, step in enumerate(body.steps):
|
||
|
|
if not step.name.strip() or step.approver_mode not in ("role", "users"):
|
||
|
|
raise HTTPException(400, "invalid_step")
|
||
|
|
ids = [u for u in step.approver_ids if u in valid_users] if step.approver_mode == "users" else []
|
||
|
|
if step.approver_mode == "users" and not ids:
|
||
|
|
raise HTTPException(400, "step_needs_approvers")
|
||
|
|
d.steps.append(WorkflowStep(position=i, name=step.name.strip(), approver_mode=step.approver_mode,
|
||
|
|
approver_ids=ids))
|
||
|
|
record(db, admin, "department", d.id, "workflow", d.name, {"steps": [before, [s.name for s in body.steps]]})
|
||
|
|
db.commit()
|
||
|
|
db.refresh(d)
|
||
|
|
return department_out(d)
|
||
|
|
|
||
|
|
|
||
|
|
XLSX = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
|
||
|
|
|
||
|
|
|
||
|
|
@router.get("/departments/{dept_id}/template.xlsx")
|
||
|
|
def download_template(dept_id: int, lang: str | None = None, year: int | None = None, rate: float | None = None,
|
||
|
|
from_year_id: int | None = None, db: Session = Depends(get_db), user: User = Depends(current_user)):
|
||
|
|
from datetime import date
|
||
|
|
from decimal import Decimal
|
||
|
|
|
||
|
|
d = db.get(Department, dept_id)
|
||
|
|
if not d:
|
||
|
|
raise HTTPException(404, "not_found")
|
||
|
|
lang = lang if lang in ("pt", "en") else user.language
|
||
|
|
lines = None
|
||
|
|
source = db.get(BudgetYear, from_year_id) if from_year_id else None
|
||
|
|
if source and source.department_id == d.id:
|
||
|
|
lines = [lv for cv in build_year_view(db, source).categories for lv in cv.lines]
|
||
|
|
year = year or (source.year + 1 if source else date.today().year + 1)
|
||
|
|
rate_value = Decimal(str(rate)) / 100 if rate is not None else (source.default_rate if source else Decimal(0))
|
||
|
|
data = build_template(db, d, lang, year, rate_value, lines)
|
||
|
|
safe = "".join(ch if ch.isalnum() else "_" for ch in d.name)
|
||
|
|
name = ("Modelo_Orcamento" if lang == "pt" else "Budget_Template") + f"_{safe}_{year}.xlsx"
|
||
|
|
record(db, user, "template", d.id, "download", name)
|
||
|
|
db.commit()
|
||
|
|
return Response(data, media_type=XLSX, headers={"Content-Disposition": f'attachment; filename="{name}"'})
|