461 lines
21 KiB
Python
461 lines
21 KiB
Python
|
import os
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
from zoneinfo import ZoneInfo
|
||
|
|
from decimal import Decimal
|
||
|
|
|
||
|
|
from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile
|
||
|
|
from pydantic import BaseModel, Field
|
||
|
|
from sqlalchemy import select
|
||
|
|
from sqlalchemy.orm import Session
|
||
|
|
|
||
|
|
from ..audit import diff, record, snapshot
|
||
|
|
from ..calc import ExpressionError, clean_months, evaluate, line_figures
|
||
|
|
from ..db import get_db
|
||
|
|
from .. import notify
|
||
|
|
from ..importer import ImportErrorWithDetail, archived_name, current_year, import_workbook
|
||
|
|
from ..models import AuditEntry, BudgetYear, Category, CostLine, Department, ReopenRequest, Site, User
|
||
|
|
from ..permissions import (allowed_actions, is_management, can_reopen, current_step, pending_reopen, require_action,
|
||
|
|
require_department_editor, steps_of)
|
||
|
|
from ..security import current_user
|
||
|
|
from ..serializers import reopen_out, year_detail, year_out
|
||
|
|
from ..summary import build_year_view
|
||
|
|
|
||
|
|
router = APIRouter(prefix="/api", tags=["years"])
|
||
|
|
|
||
|
|
LINE_FIELDS = ["category_id", "site_id", "name", "description", "remarks", "prior_expr", "base_expr",
|
||
|
|
"adj_kind", "adj_value", "adj_reason", "monthly_kind", "monthly_values", "monthly_months", "sort_order"]
|
||
|
|
|
||
|
|
|
||
|
|
def get_year(db: Session, year_id: int) -> BudgetYear:
|
||
|
|
y = db.get(BudgetYear, year_id)
|
||
|
|
if not y:
|
||
|
|
raise HTTPException(404, "not_found")
|
||
|
|
return y
|
||
|
|
|
||
|
|
|
||
|
|
def get_department(db: Session, department_id: int | None) -> Department:
|
||
|
|
d = db.get(Department, department_id or 0)
|
||
|
|
if not d:
|
||
|
|
raise HTTPException(400, "invalid_department")
|
||
|
|
return d
|
||
|
|
|
||
|
|
|
||
|
|
@router.get("/years")
|
||
|
|
def list_years(department_id: int | None = None, archived: bool = False, db: Session = Depends(get_db),
|
||
|
|
user: User = Depends(current_user)):
|
||
|
|
"""Current budgets, or with archived=true the versions that were replaced by an upload."""
|
||
|
|
q = (select(BudgetYear).where(BudgetYear.archived.is_(archived))
|
||
|
|
.order_by(BudgetYear.year.desc(), BudgetYear.department_id, BudgetYear.archived_at.desc()))
|
||
|
|
if department_id:
|
||
|
|
q = q.where(BudgetYear.department_id == department_id)
|
||
|
|
return [year_out(y, build_year_view(db, y), user) for y in db.scalars(q)]
|
||
|
|
|
||
|
|
|
||
|
|
class YearIn(BaseModel):
|
||
|
|
department_id: int | None = None
|
||
|
|
year: int | None = Field(default=None, ge=2000, le=2100)
|
||
|
|
title: str | None = None
|
||
|
|
default_rate: float | None = Field(default=None, ge=-1, le=10)
|
||
|
|
notes: str | None = None
|
||
|
|
copy_from_id: int | None = None
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/years")
|
||
|
|
def create_year(body: YearIn, db: Session = Depends(get_db), user: User = Depends(current_user)):
|
||
|
|
dept = get_department(db, body.department_id or user.department_id)
|
||
|
|
require_department_editor(user, dept.id)
|
||
|
|
if not body.year:
|
||
|
|
raise HTTPException(400, "missing_fields")
|
||
|
|
if current_year(db, dept.id, body.year):
|
||
|
|
raise HTTPException(400, "year_exists")
|
||
|
|
src = get_year(db, body.copy_from_id) if body.copy_from_id else None
|
||
|
|
if src and src.department_id != dept.id:
|
||
|
|
raise HTTPException(400, "invalid_department")
|
||
|
|
rate = Decimal(str(body.default_rate)) if body.default_rate is not None else (src.default_rate if src else Decimal(0))
|
||
|
|
y = BudgetYear(department_id=dept.id, year=body.year, title=body.title or f"Budget {body.year}",
|
||
|
|
default_rate=rate, notes=body.notes or "")
|
||
|
|
db.add(y)
|
||
|
|
db.flush()
|
||
|
|
if src:
|
||
|
|
# Last year's adjusted budget becomes this year's starting point for both actual and base;
|
||
|
|
# users then replace the actual with the real figure.
|
||
|
|
for line in src.lines:
|
||
|
|
f = line_figures(line, src.default_rate)
|
||
|
|
start = str(f.adjusted)
|
||
|
|
db.add(CostLine(
|
||
|
|
year_id=y.id, category_id=line.category_id, site_id=line.site_id, name=line.name,
|
||
|
|
description=line.description, remarks="", prior_expr=start, prior_actual=f.adjusted,
|
||
|
|
base_expr=start, base_amount=f.adjusted, adj_kind=line.adj_kind, adj_value=line.adj_value,
|
||
|
|
adj_reason=line.adj_reason, monthly_kind=line.monthly_kind, monthly_values=line.monthly_values,
|
||
|
|
monthly_months=line.monthly_months,
|
||
|
|
sort_order=line.sort_order))
|
||
|
|
record(db, user, "year", y.id, "create", f"{dept.name} {y.year}" + (f" from {src.year}" if src else ""),
|
||
|
|
year_id=y.id)
|
||
|
|
db.commit()
|
||
|
|
return year_out(y)
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/years/import")
|
||
|
|
async def import_year(file: UploadFile = File(...), year: int | None = Form(None), replace: bool = Form(False),
|
||
|
|
department_id: int | None = Form(None),
|
||
|
|
db: Session = Depends(get_db), user: User = Depends(current_user)):
|
||
|
|
dept = get_department(db, department_id or user.department_id)
|
||
|
|
require_department_editor(user, dept.id)
|
||
|
|
data = await file.read()
|
||
|
|
if len(data) > 20 * 1024 * 1024:
|
||
|
|
raise HTTPException(400, "file_too_large")
|
||
|
|
try:
|
||
|
|
result = import_workbook(db, data, user, dept.id, year=year, replace=replace,
|
||
|
|
filename=_clean_filename(file.filename))
|
||
|
|
except ImportErrorWithDetail as exc:
|
||
|
|
db.rollback()
|
||
|
|
raise HTTPException(409 if str(exc) in ("year_exists", "year_not_draft") else 400, str(exc))
|
||
|
|
y = current_year(db, dept.id, result.year)
|
||
|
|
archived = db.scalars(select(BudgetYear).where(BudgetYear.department_id == dept.id, BudgetYear.year == result.year,
|
||
|
|
BudgetYear.archived.is_(True))
|
||
|
|
.order_by(BudgetYear.archived_at.desc())).first() if replace else None
|
||
|
|
return {"archived": year_out(archived) if archived else None,"year": year_out(y), "lines": result.lines, "categories_created": result.categories_created,
|
||
|
|
"warnings": result.warnings}
|
||
|
|
|
||
|
|
|
||
|
|
def _clean_filename(name: str | None) -> str:
|
||
|
|
return (name or "").replace("\\", "/").rsplit("/", 1)[-1].strip()[:255]
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/years/import/check")
|
||
|
|
async def check_import(file: UploadFile = File(...), year: int | None = Form(None),
|
||
|
|
department_id: int | None = Form(None),
|
||
|
|
db: Session = Depends(get_db), user: User = Depends(current_user)):
|
||
|
|
"""Which budget an upload would create, and whether it would replace (and archive) a current one."""
|
||
|
|
dept = get_department(db, department_id or user.department_id)
|
||
|
|
require_department_editor(user, dept.id)
|
||
|
|
data = await file.read()
|
||
|
|
if len(data) > 20 * 1024 * 1024:
|
||
|
|
raise HTTPException(400, "file_too_large")
|
||
|
|
try:
|
||
|
|
result = import_workbook(db, data, user, dept.id, year=year, check_only=True)
|
||
|
|
except ImportErrorWithDetail as exc:
|
||
|
|
raise HTTPException(400, str(exc))
|
||
|
|
existing = current_year(db, dept.id, result.year)
|
||
|
|
name = _clean_filename(file.filename)
|
||
|
|
out = {"year": result.year, "department": dept.name, "filename": name, "existing": None}
|
||
|
|
if existing:
|
||
|
|
now = datetime.now(ZoneInfo(os.getenv("TZ") or "Europe/Lisbon"))
|
||
|
|
out["existing"] = {
|
||
|
|
**year_out(existing, build_year_view(db, existing)), "can_replace": existing.status == "draft",
|
||
|
|
"same_file": bool(existing.source_file) and existing.source_file.lower() == name.lower(),
|
||
|
|
"archived_title": archived_name(existing.title or f"Budget {existing.year}", now),
|
||
|
|
"archived_file": archived_name(existing.source_file, now) if existing.source_file else "",
|
||
|
|
}
|
||
|
|
return out
|
||
|
|
|
||
|
|
|
||
|
|
@router.get("/years/{year_id}")
|
||
|
|
def read_year(year_id: int, site_id: int | None = None, db: Session = Depends(get_db),
|
||
|
|
user: User = Depends(current_user)):
|
||
|
|
return year_detail(build_year_view(db, get_year(db, year_id), site_id), user)
|
||
|
|
|
||
|
|
|
||
|
|
@router.put("/years/{year_id}")
|
||
|
|
def update_year(year_id: int, body: YearIn, db: Session = Depends(get_db), user: User = Depends(current_user)):
|
||
|
|
y = get_year(db, year_id)
|
||
|
|
require_action(user, y, "edit")
|
||
|
|
before = snapshot(y, ["title", "default_rate", "notes"])
|
||
|
|
if body.title is not None:
|
||
|
|
y.title = body.title
|
||
|
|
if body.default_rate is not None:
|
||
|
|
y.default_rate = Decimal(str(body.default_rate))
|
||
|
|
if body.notes is not None:
|
||
|
|
y.notes = body.notes
|
||
|
|
record(db, user, "year", y.id, "update", str(y.year), diff(before, snapshot(y, ["title", "default_rate", "notes"])),
|
||
|
|
year_id=y.id)
|
||
|
|
db.commit()
|
||
|
|
return year_out(y, user=user)
|
||
|
|
|
||
|
|
|
||
|
|
@router.delete("/years/{year_id}")
|
||
|
|
def delete_year(year_id: int, db: Session = Depends(get_db), user: User = Depends(current_user)):
|
||
|
|
y = get_year(db, year_id)
|
||
|
|
require_action(user, y, "delete")
|
||
|
|
record(db, user, "year", y.id, "delete",
|
||
|
|
f"{y.department.name} {y.year} · {y.title} ({y.status}, {len(y.lines)} lines)", year_id=y.id)
|
||
|
|
db.delete(y)
|
||
|
|
db.commit()
|
||
|
|
return {"ok": True}
|
||
|
|
|
||
|
|
|
||
|
|
class StatusIn(BaseModel):
|
||
|
|
action: str # submit | approve | return | reopen
|
||
|
|
comment: str | None = Field(default=None, max_length=500)
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/years/{year_id}/status")
|
||
|
|
def change_status(year_id: int, body: StatusIn, db: Session = Depends(get_db), user: User = Depends(current_user)):
|
||
|
|
y = get_year(db, year_id)
|
||
|
|
if body.action not in ("submit", "approve", "return", "reopen"):
|
||
|
|
raise HTTPException(400, "invalid_action")
|
||
|
|
expected = {"submit": "draft", "approve": "in_review", "return": "in_review", "reopen": "approved"}[body.action]
|
||
|
|
if y.status != expected:
|
||
|
|
raise HTTPException(409, "invalid_transition")
|
||
|
|
if not allowed_actions(user, y)[body.action]:
|
||
|
|
raise HTTPException(403, "forbidden")
|
||
|
|
if body.action == "reopen" and not (body.comment or "").strip():
|
||
|
|
raise HTTPException(400, "reason_required")
|
||
|
|
now = datetime.now(timezone.utc)
|
||
|
|
before = y.status
|
||
|
|
steps = steps_of(y.department)
|
||
|
|
step = current_step(y)
|
||
|
|
summary = body.comment or ""
|
||
|
|
if body.action == "submit":
|
||
|
|
y.status, y.current_step = "in_review", 0
|
||
|
|
y.submitted_by_id, y.submitted_at = user.id, now
|
||
|
|
y.approved_by_id = y.approved_at = None
|
||
|
|
elif body.action == "approve":
|
||
|
|
index = y.current_step or 0
|
||
|
|
summary = (f"{step.name}: " if step else "") + summary
|
||
|
|
if index + 1 < len(steps):
|
||
|
|
y.current_step = index + 1
|
||
|
|
else:
|
||
|
|
y.status, y.current_step = "approved", None
|
||
|
|
y.approved_by_id, y.approved_at = user.id, now
|
||
|
|
else: # return / reopen
|
||
|
|
y.status, y.current_step = "draft", None
|
||
|
|
y.approved_by_id = y.approved_at = None
|
||
|
|
if body.action == "reopen":
|
||
|
|
_reopen(y, user, now)
|
||
|
|
record(db, user, "year", y.id, body.action, summary.strip(" :"),
|
||
|
|
{"status": [before, y.status], **({"step": step.name} if step else {})}, year_id=y.id)
|
||
|
|
db.commit()
|
||
|
|
return year_out(y, user=user)
|
||
|
|
|
||
|
|
|
||
|
|
def _reopen(y: BudgetYear, user: User, now: datetime) -> None:
|
||
|
|
y.status, y.current_step = "draft", None
|
||
|
|
y.approved_by_id = y.approved_at = None
|
||
|
|
pending = pending_reopen(y)
|
||
|
|
if pending: # reopening answers an open request
|
||
|
|
pending.status, pending.decided_by_id, pending.decided_at = "accepted", user.id, now
|
||
|
|
|
||
|
|
|
||
|
|
class ReopenIn(BaseModel):
|
||
|
|
reason: str = Field(default="", max_length=2000)
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/years/{year_id}/reopen-requests")
|
||
|
|
def request_reopen(year_id: int, body: ReopenIn, db: Session = Depends(get_db), user: User = Depends(current_user)):
|
||
|
|
"""Anyone may ask for an approved budget to be reopened, saying why. Approvers reopen it at once."""
|
||
|
|
y = get_year(db, year_id)
|
||
|
|
if is_management(user):
|
||
|
|
raise HTTPException(403, "forbidden")
|
||
|
|
reason = body.reason.strip()
|
||
|
|
if len(reason) < 5:
|
||
|
|
raise HTTPException(400, "reason_required")
|
||
|
|
if y.archived or y.status != "approved":
|
||
|
|
raise HTTPException(409, "invalid_transition")
|
||
|
|
if pending_reopen(y) and not can_reopen(user, y):
|
||
|
|
raise HTTPException(409, "reopen_already_requested")
|
||
|
|
now = datetime.now(timezone.utc)
|
||
|
|
req = ReopenRequest(year_id=y.id, requested_by_id=user.id, requested_at=now, reason=reason, status="pending",
|
||
|
|
decision_note="")
|
||
|
|
label = f"{y.department.name} {y.year}"
|
||
|
|
if can_reopen(user, y):
|
||
|
|
pending = pending_reopen(y)
|
||
|
|
if pending: # an approver reopening answers the open request
|
||
|
|
req = pending
|
||
|
|
req.decision_note = reason
|
||
|
|
else:
|
||
|
|
db.add(req)
|
||
|
|
db.flush()
|
||
|
|
db.refresh(y)
|
||
|
|
_reopen(y, user, now)
|
||
|
|
record(db, user, "year", y.id, "reopen", reason, {"status": ["approved", "draft"]}, year_id=y.id)
|
||
|
|
db.commit()
|
||
|
|
return {"reopened": True, "request": reopen_out(req), "year": year_out(y, user=user)}
|
||
|
|
db.add(req)
|
||
|
|
db.flush()
|
||
|
|
db.refresh(y)
|
||
|
|
record(db, user, "year", y.id, "reopen_requested", reason, year_id=y.id)
|
||
|
|
db.commit()
|
||
|
|
notify.send(db, user, y, notify.reopeners(db, y), f"Pedido de reabertura / Reopen request: {label}",
|
||
|
|
f"{user.name} pediu a reabertura do orçamento {label} ({y.title}).\n"
|
||
|
|
f"{user.name} asked for the {label} budget ({y.title}) to be reopened.\n\n"
|
||
|
|
f"Motivo / Reason:\n{reason}\n")
|
||
|
|
return {"reopened": False, "request": reopen_out(req), "year": year_out(y, user=user)}
|
||
|
|
|
||
|
|
|
||
|
|
class DecisionIn(BaseModel):
|
||
|
|
accept: bool
|
||
|
|
note: str = Field(default="", max_length=2000)
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/reopen-requests/{request_id}/decide")
|
||
|
|
def decide_reopen(request_id: int, body: DecisionIn, db: Session = Depends(get_db), user: User = Depends(current_user)):
|
||
|
|
req = db.get(ReopenRequest, request_id)
|
||
|
|
if not req:
|
||
|
|
raise HTTPException(404, "not_found")
|
||
|
|
y = req.budget_year
|
||
|
|
if req.status != "pending" or y.status != "approved":
|
||
|
|
raise HTTPException(409, "invalid_transition")
|
||
|
|
if not can_reopen(user, y) or y.archived:
|
||
|
|
raise HTTPException(403, "forbidden")
|
||
|
|
if not body.accept and not body.note.strip():
|
||
|
|
raise HTTPException(400, "reason_required")
|
||
|
|
now = datetime.now(timezone.utc)
|
||
|
|
note = body.note.strip()
|
||
|
|
label = f"{y.department.name} {y.year}"
|
||
|
|
if body.accept:
|
||
|
|
_reopen(y, user, now)
|
||
|
|
req.decision_note = note
|
||
|
|
record(db, user, "year", y.id, "reopen", f"{req.reason}" + (f" — {note}" if note else ""),
|
||
|
|
{"status": ["approved", "draft"], "requested_by": req.requested_by.name if req.requested_by else None},
|
||
|
|
year_id=y.id)
|
||
|
|
else:
|
||
|
|
req.status, req.decided_by_id, req.decided_at, req.decision_note = "declined", user.id, now, note
|
||
|
|
record(db, user, "year", y.id, "reopen_declined", note, year_id=y.id)
|
||
|
|
db.commit()
|
||
|
|
if req.requested_by:
|
||
|
|
verdict = ("aceite / accepted: o orçamento voltou a rascunho / the budget is back in draft" if body.accept
|
||
|
|
else "recusado / declined")
|
||
|
|
notify.send(db, user, y, [req.requested_by], f"Pedido de reabertura {label}: {verdict.split(':')[0]}",
|
||
|
|
f"O seu pedido de reabertura do orçamento {label} foi {verdict}.\n"
|
||
|
|
f"Your request to reopen the {label} budget was {verdict}.\n\n"
|
||
|
|
f"{user.name}" + (f":\n{note}\n" if note else "\n"))
|
||
|
|
return {"request": reopen_out(req), "year": year_out(y, user=user)}
|
||
|
|
|
||
|
|
|
||
|
|
@router.get("/years/{year_id}/audit")
|
||
|
|
def year_audit(year_id: int, limit: int = 200, db: Session = Depends(get_db), _: User = Depends(current_user)):
|
||
|
|
get_year(db, year_id)
|
||
|
|
rows = db.scalars(select(AuditEntry).where(AuditEntry.year_id == year_id)
|
||
|
|
.order_by(AuditEntry.at.desc()).limit(min(limit, 1000))).all()
|
||
|
|
return [{"at": a.at.isoformat(), "user": a.user.name if a.user else None, "entity": a.entity,
|
||
|
|
"entity_id": a.entity_id, "action": a.action, "summary": a.summary, "changes": a.changes}
|
||
|
|
for a in rows]
|
||
|
|
|
||
|
|
|
||
|
|
class LineIn(BaseModel):
|
||
|
|
category_id: int | None = None
|
||
|
|
site_id: int | None = None
|
||
|
|
name: str | None = Field(default=None, max_length=255)
|
||
|
|
description: str | None = Field(default=None, max_length=500)
|
||
|
|
remarks: str | None = None
|
||
|
|
prior_expr: str | None = Field(default=None, max_length=400)
|
||
|
|
base_expr: str | None = Field(default=None, max_length=400)
|
||
|
|
adj_kind: str | None = None
|
||
|
|
adj_value: float | None = None
|
||
|
|
adj_reason: str | None = Field(default=None, max_length=255)
|
||
|
|
monthly_kind: str | None = None
|
||
|
|
monthly_values: list[float] | None = None
|
||
|
|
monthly_months: list[int] | None = None
|
||
|
|
sort_order: int | None = None
|
||
|
|
|
||
|
|
|
||
|
|
def _apply_line(line: CostLine, body: LineIn, db: Session, department_id: int) -> None:
|
||
|
|
data = body.model_dump(exclude_unset=True)
|
||
|
|
if "category_id" in data:
|
||
|
|
cat = db.get(Category, data["category_id"] or 0)
|
||
|
|
if not cat or cat.department_id != department_id:
|
||
|
|
raise HTTPException(400, "invalid_category")
|
||
|
|
line.category_id = data["category_id"]
|
||
|
|
if "site_id" in data:
|
||
|
|
if data["site_id"] is not None and not db.get(Site, data["site_id"]):
|
||
|
|
raise HTTPException(400, "invalid_site")
|
||
|
|
line.site_id = data["site_id"]
|
||
|
|
for f in ("name", "description", "remarks", "adj_reason", "sort_order"):
|
||
|
|
if f in data and data[f] is not None:
|
||
|
|
setattr(line, f, data[f].strip() if isinstance(data[f], str) else data[f])
|
||
|
|
for f, target in (("prior_expr", "prior_actual"), ("base_expr", "base_amount")):
|
||
|
|
if f in data and data[f] is not None:
|
||
|
|
try:
|
||
|
|
value = evaluate(data[f])
|
||
|
|
except ExpressionError:
|
||
|
|
raise HTTPException(400, f"invalid_expression:{f}")
|
||
|
|
setattr(line, f, data[f].strip().lstrip("=") or "0")
|
||
|
|
setattr(line, target, value)
|
||
|
|
if "adj_kind" in data:
|
||
|
|
if data["adj_kind"] not in ("year_rate", "percent", "factor", "none"):
|
||
|
|
raise HTTPException(400, "invalid_adjustment")
|
||
|
|
line.adj_kind = data["adj_kind"]
|
||
|
|
if data.get("adj_value") is not None:
|
||
|
|
line.adj_value = Decimal(str(data["adj_value"]))
|
||
|
|
if "monthly_kind" in data:
|
||
|
|
if data["monthly_kind"] not in ("even", "selected", "manual"):
|
||
|
|
raise HTTPException(400, "invalid_monthly")
|
||
|
|
if data["monthly_kind"] == "manual" and line.monthly_kind != "manual" and not data.get("monthly_values"):
|
||
|
|
# Switching to typed months starts from the current even spread.
|
||
|
|
year = db.get(BudgetYear, line.year_id)
|
||
|
|
spread = line_figures(line, year.default_rate if year else Decimal(0)).months
|
||
|
|
line.monthly_values = [float(m.quantize(Decimal("0.01"))) for m in spread]
|
||
|
|
if line.monthly_kind == "manual" and data["monthly_kind"] != "manual" and data.get("base_expr") is None:
|
||
|
|
# Leaving typed months keeps the year total: it becomes the base, without a further adjustment.
|
||
|
|
total = sum((Decimal(str(v or 0)) for v in (line.monthly_values or [])), Decimal(0))
|
||
|
|
line.base_expr, line.base_amount = str(total), total
|
||
|
|
if data.get("adj_kind") is None:
|
||
|
|
line.adj_kind, line.adj_value = "none", Decimal(0)
|
||
|
|
line.monthly_kind = data["monthly_kind"]
|
||
|
|
if "monthly_values" in data:
|
||
|
|
values = data["monthly_values"]
|
||
|
|
if values is not None and len(values) != 12:
|
||
|
|
raise HTTPException(400, "invalid_monthly")
|
||
|
|
line.monthly_values = values
|
||
|
|
if line.monthly_kind == "manual" and not line.monthly_values:
|
||
|
|
line.monthly_values = [0.0] * 12
|
||
|
|
if "monthly_months" in data:
|
||
|
|
values = data["monthly_months"]
|
||
|
|
if values is not None and any(not 1 <= m <= 12 for m in values):
|
||
|
|
raise HTTPException(400, "invalid_monthly")
|
||
|
|
line.monthly_months = clean_months(values) or None
|
||
|
|
if line.monthly_kind == "selected" and not line.monthly_months:
|
||
|
|
raise HTTPException(400, "no_months_selected")
|
||
|
|
if not line.name:
|
||
|
|
raise HTTPException(400, "missing_fields")
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/years/{year_id}/lines")
|
||
|
|
def create_line(year_id: int, body: LineIn, db: Session = Depends(get_db), user: User = Depends(current_user)):
|
||
|
|
y = get_year(db, year_id)
|
||
|
|
require_action(user, y, "edit")
|
||
|
|
if not body.category_id or not body.name:
|
||
|
|
raise HTTPException(400, "missing_fields")
|
||
|
|
line = CostLine(year_id=y.id, name="", description="", remarks="", prior_expr="0", prior_actual=Decimal(0),
|
||
|
|
base_expr="0", base_amount=Decimal(0), adj_kind="year_rate", adj_value=Decimal(0),
|
||
|
|
adj_reason="", monthly_kind="even", sort_order=0)
|
||
|
|
_apply_line(line, body, db, y.department_id)
|
||
|
|
if not line.sort_order:
|
||
|
|
line.sort_order = max([l.sort_order for l in y.lines if l.category_id == line.category_id] or [0]) + 10
|
||
|
|
db.add(line)
|
||
|
|
db.flush()
|
||
|
|
record(db, user, "line", line.id, "create", line.name, snapshot(line, LINE_FIELDS), year_id=y.id)
|
||
|
|
db.commit()
|
||
|
|
return {"id": line.id}
|
||
|
|
|
||
|
|
|
||
|
|
def _line_and_year(db: Session, line_id: int) -> tuple[CostLine, BudgetYear]:
|
||
|
|
line = db.get(CostLine, line_id)
|
||
|
|
if not line:
|
||
|
|
raise HTTPException(404, "not_found")
|
||
|
|
return line, line.budget_year
|
||
|
|
|
||
|
|
|
||
|
|
@router.put("/lines/{line_id}")
|
||
|
|
def update_line(line_id: int, body: LineIn, db: Session = Depends(get_db), user: User = Depends(current_user)):
|
||
|
|
line, y = _line_and_year(db, line_id)
|
||
|
|
require_action(user, y, "edit")
|
||
|
|
before = snapshot(line, LINE_FIELDS)
|
||
|
|
_apply_line(line, body, db, y.department_id)
|
||
|
|
changes = diff(before, snapshot(line, LINE_FIELDS))
|
||
|
|
if changes:
|
||
|
|
record(db, user, "line", line.id, "update", line.name, changes, year_id=y.id)
|
||
|
|
db.commit()
|
||
|
|
return {"id": line.id}
|
||
|
|
|
||
|
|
|
||
|
|
@router.delete("/lines/{line_id}")
|
||
|
|
def delete_line(line_id: int, db: Session = Depends(get_db), user: User = Depends(current_user)):
|
||
|
|
line, y = _line_and_year(db, line_id)
|
||
|
|
require_action(user, y, "edit")
|
||
|
|
record(db, user, "line", line.id, "delete", line.name, snapshot(line, LINE_FIELDS), year_id=y.id)
|
||
|
|
db.delete(line)
|
||
|
|
db.commit()
|
||
|
|
return {"ok": True}
|