77 lines
3.9 KiB
Python
77 lines
3.9 KiB
Python
"""The activity log: everything users did, newest first (administrators)."""
|
||
import csv
|
||
import io
|
||
|
||
from fastapi import APIRouter, Depends
|
||
from fastapi.responses import Response
|
||
from sqlalchemy import func, or_, select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ..db import get_db
|
||
from ..models import AuditEntry, BudgetYear, Department, User
|
||
from ..security import require
|
||
from .mail import _date
|
||
|
||
router = APIRouter(prefix="/api", tags=["activity"])
|
||
|
||
|
||
def _query(q, user_id, entity, action, year_id, date_from, date_to):
|
||
query = select(AuditEntry)
|
||
if user_id:
|
||
query = query.where(AuditEntry.user_id == user_id)
|
||
if entity:
|
||
query = query.where(AuditEntry.entity == entity)
|
||
if action:
|
||
query = query.where(AuditEntry.action == action)
|
||
if year_id:
|
||
query = query.where(AuditEntry.year_id == year_id)
|
||
if date_from:
|
||
query = query.where(AuditEntry.at >= _date(date_from))
|
||
if date_to:
|
||
query = query.where(AuditEntry.at < _date(date_to, end=True))
|
||
if q:
|
||
like = f"%{q.strip().lower()}%"
|
||
query = query.where(or_(func.lower(AuditEntry.summary).like(like),
|
||
AuditEntry.user_id.in_(select(User.id).where(func.lower(User.name).like(like)))))
|
||
return query
|
||
|
||
|
||
def _rows(db: Session, entries: list[AuditEntry]) -> list[dict]:
|
||
year_ids = {a.year_id for a in entries if a.year_id}
|
||
labels = {y.id: f"{d.name} {y.year}" for y, d in db.execute(
|
||
select(BudgetYear, Department).join(Department).where(BudgetYear.id.in_(year_ids)))} if year_ids else {}
|
||
return [{"id": a.id, "at": a.at.isoformat(), "user": a.user.name if a.user else None, "user_id": a.user_id,
|
||
"entity": a.entity, "entity_id": a.entity_id, "action": a.action, "summary": a.summary,
|
||
"changes": a.changes, "year_id": a.year_id,
|
||
"budget": labels.get(a.year_id) if a.year_id else None,
|
||
"budget_exists": a.year_id in labels} for a in entries]
|
||
|
||
|
||
@router.get("/activity")
|
||
def activity(q: str | None = None, user_id: int | None = None, entity: str | None = None, action: str | None = None,
|
||
year_id: int | None = None, date_from: str | None = None, date_to: str | None = None,
|
||
limit: int = 100, offset: int = 0, db: Session = Depends(get_db), _: User = Depends(require("admin"))):
|
||
query = _query(q, user_id, entity, action, year_id, date_from, date_to)
|
||
total = db.scalar(select(func.count()).select_from(query.subquery()))
|
||
entries = db.scalars(query.order_by(AuditEntry.at.desc(), AuditEntry.id.desc())
|
||
.limit(min(max(limit, 1), 500)).offset(max(offset, 0))).all()
|
||
kinds = db.execute(select(AuditEntry.entity, AuditEntry.action).distinct()).all()
|
||
return {"total": total, "rows": _rows(db, entries),
|
||
"entities": sorted({e for e, _ in kinds}), "actions": sorted({a for _, a in kinds})}
|
||
|
||
|
||
@router.get("/activity.csv")
|
||
def activity_csv(q: str | None = None, user_id: int | None = None, entity: str | None = None,
|
||
action: str | None = None, year_id: int | None = None, date_from: str | None = None,
|
||
date_to: str | None = None, db: Session = Depends(get_db), _: User = Depends(require("admin"))):
|
||
entries = db.scalars(_query(q, user_id, entity, action, year_id, date_from, date_to)
|
||
.order_by(AuditEntry.at.desc(), AuditEntry.id.desc())).all()
|
||
out = io.StringIO()
|
||
w = csv.writer(out, delimiter=";")
|
||
w.writerow(["date", "user", "area", "action", "budget", "summary", "changes"])
|
||
for r in _rows(db, entries):
|
||
w.writerow([r["at"], r["user"] or "", r["entity"], r["action"], r["budget"] or "", r["summary"],
|
||
"" if not r["changes"] else str(r["changes"])])
|
||
return Response("" + out.getvalue(), media_type="text/csv; charset=utf-8",
|
||
headers={"Content-Disposition": 'attachment; filename="activity_log.csv"'})
|