1115 lines
40 KiB
Python
1115 lines
40 KiB
Python
from __future__ import annotations
|
||
|
||
import calendar
|
||
from collections.abc import Sequence
|
||
from datetime import date, datetime, timedelta
|
||
|
||
from fastapi import APIRouter, Depends, Form, HTTPException, Request
|
||
from fastapi.responses import HTMLResponse, RedirectResponse
|
||
from sqlalchemy import delete, func, select
|
||
from sqlalchemy.orm import Session, selectinload
|
||
|
||
from .config import settings
|
||
from .database import get_db
|
||
from .models import (
|
||
Assignment,
|
||
MonthlyPattern,
|
||
NotificationLog,
|
||
RotationMember,
|
||
Schedule,
|
||
SwapRequest,
|
||
User,
|
||
)
|
||
from .notifications import send_email_notification, send_many
|
||
from .scheduling import assignments_by_date, ensure_month_assignments, iter_month_dates, month_bounds, rotation_user_for_date
|
||
from .security import hash_password, new_csrf_token, verify_password
|
||
|
||
router = APIRouter()
|
||
|
||
|
||
def _redirect(path: str, status_code: int = 303) -> RedirectResponse:
|
||
return RedirectResponse(path, status_code=status_code)
|
||
|
||
|
||
def _current_user(request: Request, db: Session) -> User | None:
|
||
user_id = request.session.get("user_id")
|
||
if not user_id:
|
||
return None
|
||
return db.get(User, int(user_id))
|
||
|
||
|
||
def _require_user(request: Request, db: Session) -> User:
|
||
user = _current_user(request, db)
|
||
if not user or not user.is_active:
|
||
raise HTTPException(status_code=401, detail="Login required")
|
||
return user
|
||
|
||
|
||
def _require_admin(request: Request, db: Session) -> User:
|
||
user = _require_user(request, db)
|
||
if not user.is_admin:
|
||
raise HTTPException(status_code=403, detail="Administrator access required")
|
||
return user
|
||
|
||
|
||
def _csrf(request: Request) -> str:
|
||
token = request.session.get("csrf")
|
||
if not token:
|
||
token = new_csrf_token()
|
||
request.session["csrf"] = token
|
||
return token
|
||
|
||
|
||
def _check_csrf(request: Request, token: str) -> None:
|
||
expected = request.session.get("csrf")
|
||
if not expected or token != expected:
|
||
raise HTTPException(status_code=400, detail="Invalid CSRF token")
|
||
|
||
|
||
def _flash(request: Request, message: str, level: str = "info") -> None:
|
||
request.session["flash"] = {"message": message, "level": level}
|
||
|
||
|
||
def _context(request: Request, db: Session, **extra):
|
||
return {
|
||
"request": request,
|
||
"settings": settings,
|
||
"current_user": _current_user(request, db),
|
||
"csrf_token": _csrf(request),
|
||
"flash": request.session.pop("flash", None),
|
||
**extra,
|
||
}
|
||
|
||
|
||
def _parse_month(year: int | None, month: int | None) -> tuple[int, int]:
|
||
today = date.today()
|
||
y = year or today.year
|
||
m = month or today.month
|
||
if m < 1 or m > 12 or y < 2000 or y > 2100:
|
||
raise HTTPException(status_code=400, detail="Invalid month")
|
||
return y, m
|
||
|
||
|
||
def _weekday_labels(weekdays: set[int]) -> str:
|
||
return ", ".join(calendar.day_name[d] for d in sorted(weekdays))
|
||
|
||
|
||
ALL_WEEKDAYS = frozenset(range(7))
|
||
|
||
|
||
def _next_dates(weekdays: set[int], count: int = 6) -> list[str]:
|
||
today = date.today()
|
||
labels: list[str] = []
|
||
day = today
|
||
while len(labels) < count:
|
||
if day.weekday() in weekdays:
|
||
labels.append(f"{day.strftime('%a')} {day.day} {day.strftime('%b')}")
|
||
day += timedelta(days=1)
|
||
return labels
|
||
|
||
|
||
def _schedule_card(schedule: Schedule, db: Session | None = None) -> dict:
|
||
weekdays = schedule.weekday_set
|
||
fixed = schedule.schedule_type != "rotation" and weekdays == ALL_WEEKDAYS
|
||
if fixed:
|
||
kind_label = "Every day · repeats every month"
|
||
weekday_label = "Every day"
|
||
dates: list[str] = []
|
||
else:
|
||
weekday_label = _weekday_labels(weekdays)
|
||
kind_label = f"Rotating · {weekday_label}"
|
||
dates = _next_dates(weekdays)
|
||
pattern: list[tuple[int, str]] = []
|
||
if fixed and db is not None:
|
||
rows = db.scalars(
|
||
select(MonthlyPattern)
|
||
.where(MonthlyPattern.schedule_id == schedule.id)
|
||
.options(selectinload(MonthlyPattern.user))
|
||
.order_by(MonthlyPattern.day_of_month)
|
||
).all()
|
||
pattern = [
|
||
(row.day_of_month, row.user.name)
|
||
for row in rows
|
||
if row.user is not None and row.user.is_active
|
||
]
|
||
return {
|
||
"schedule": schedule,
|
||
"kind": "fixed" if fixed else "rotation",
|
||
"kind_label": kind_label,
|
||
"weekday_label": weekday_label,
|
||
"dates": dates,
|
||
"pattern": pattern,
|
||
"rotation_names": [member.user.name for member in schedule.rotation_members],
|
||
}
|
||
|
||
|
||
def _schedule_groups(schedules: Sequence[Schedule], db: Session | None = None) -> list[dict]:
|
||
cards = [_schedule_card(schedule, db) for schedule in schedules]
|
||
# Every rotation is independent: Saturday and Sunday lunch never share an order.
|
||
return sorted(cards, key=lambda card: (card["kind"] != "fixed", card["weekday_label"] != "Every day"))
|
||
|
||
|
||
def _active_schedules(db: Session) -> list[Schedule]:
|
||
return list(
|
||
db.scalars(
|
||
select(Schedule)
|
||
.where(Schedule.is_active.is_(True))
|
||
.options(selectinload(Schedule.rotation_members).selectinload(RotationMember.user))
|
||
.order_by(Schedule.id)
|
||
).all()
|
||
)
|
||
|
||
|
||
def _all_schedules(db: Session) -> list[Schedule]:
|
||
return list(
|
||
db.scalars(
|
||
select(Schedule)
|
||
.options(selectinload(Schedule.rotation_members).selectinload(RotationMember.user))
|
||
.order_by(Schedule.id)
|
||
).all()
|
||
)
|
||
|
||
|
||
def _get_schedule(db: Session, schedule_id: int) -> Schedule:
|
||
schedule = db.get(Schedule, schedule_id)
|
||
if not schedule:
|
||
raise HTTPException(status_code=404, detail="Schedule not found")
|
||
return schedule
|
||
|
||
|
||
def _month_nav(year: int, month: int) -> tuple[tuple[int, int], tuple[int, int]]:
|
||
if month == 1:
|
||
prev = (year - 1, 12)
|
||
else:
|
||
prev = (year, month - 1)
|
||
if month == 12:
|
||
nxt = (year + 1, 1)
|
||
else:
|
||
nxt = (year, month + 1)
|
||
return prev, nxt
|
||
|
||
|
||
@router.get("/health")
|
||
def health():
|
||
return {"status": "ok"}
|
||
|
||
|
||
@router.get("/login", response_class=HTMLResponse)
|
||
def login_page(request: Request, db: Session = Depends(get_db)):
|
||
if _current_user(request, db):
|
||
return _redirect("/")
|
||
return request.app.state.templates.TemplateResponse(
|
||
request=request,
|
||
name="login.html",
|
||
context=_context(request, db),
|
||
)
|
||
|
||
|
||
@router.post("/login")
|
||
def login(
|
||
request: Request,
|
||
username: str = Form(...),
|
||
password: str = Form(...),
|
||
csrf_token: str = Form(...),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
_check_csrf(request, csrf_token)
|
||
user = db.scalar(select(User).where(func.lower(User.username) == username.strip().lower()))
|
||
if not user or not user.is_active or not user.can_login or not verify_password(password, user.password_hash):
|
||
_flash(request, "Invalid username or password.", "error")
|
||
return _redirect("/login")
|
||
request.session["user_id"] = user.id
|
||
_flash(request, f"Welcome, {user.name}.")
|
||
return _redirect("/")
|
||
|
||
|
||
@router.post("/logout")
|
||
def logout(request: Request, csrf_token: str = Form(...)):
|
||
_check_csrf(request, csrf_token)
|
||
request.session.clear()
|
||
return _redirect("/login")
|
||
|
||
|
||
@router.get("/schedules", response_class=HTMLResponse)
|
||
def public_schedules(request: Request, db: Session = Depends(get_db)):
|
||
cards = _schedule_groups(_active_schedules(db), db)
|
||
return request.app.state.templates.TemplateResponse(
|
||
request=request,
|
||
name="schedules.html",
|
||
context=_context(request, db, cards=cards),
|
||
)
|
||
|
||
|
||
@router.get("/schedules/{schedule_id}", response_class=HTMLResponse)
|
||
def public_schedule(schedule_id: int, request: Request, db: Session = Depends(get_db)):
|
||
schedule = _get_schedule(db, schedule_id)
|
||
if not schedule.is_active:
|
||
raise HTTPException(status_code=404, detail="Schedule not found")
|
||
return request.app.state.templates.TemplateResponse(
|
||
request=request,
|
||
name="schedule.html",
|
||
context=_context(request, db, card=_schedule_card(schedule, db), cards=_schedule_groups(_active_schedules(db), db)),
|
||
)
|
||
|
||
|
||
@router.get("/", response_class=HTMLResponse)
|
||
def dashboard(
|
||
request: Request,
|
||
year: int | None = None,
|
||
month: int | None = None,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
user = _require_user(request, db)
|
||
year, month = _parse_month(year, month)
|
||
assignments = ensure_month_assignments(db, year, month)
|
||
grouped = assignments_by_date(assignments)
|
||
prev, nxt = _month_nav(year, month)
|
||
|
||
my_assignment_ids = {a.id for a in assignments if a.user_id == user.id}
|
||
open_swap_assignment_ids = set(
|
||
db.scalars(
|
||
select(SwapRequest.assignment_id).where(
|
||
SwapRequest.requester_id == user.id,
|
||
SwapRequest.status.in_(["open", "pending_approval"]),
|
||
)
|
||
).all()
|
||
)
|
||
|
||
return request.app.state.templates.TemplateResponse(
|
||
request=request,
|
||
name="dashboard.html",
|
||
context=_context(
|
||
request,
|
||
db,
|
||
year=year,
|
||
month=month,
|
||
month_name=calendar.month_name[month],
|
||
month_days=[(date(year, month, d), grouped.get(date(year, month, d), [])) for d in range(1, calendar.monthrange(year, month)[1] + 1)],
|
||
grouped=grouped,
|
||
my_assignment_ids=my_assignment_ids,
|
||
open_swap_assignment_ids=open_swap_assignment_ids,
|
||
prev=prev,
|
||
nxt=nxt,
|
||
active_users=db.scalars(select(User).where(User.is_active.is_(True)).order_by(User.name)).all(),
|
||
today=date.today(),
|
||
),
|
||
)
|
||
|
||
|
||
@router.get("/swaps", response_class=HTMLResponse)
|
||
def swaps_page(request: Request, db: Session = Depends(get_db)):
|
||
user = _require_user(request, db)
|
||
query = (
|
||
select(SwapRequest)
|
||
.options(
|
||
selectinload(SwapRequest.assignment).selectinload(Assignment.schedule),
|
||
selectinload(SwapRequest.requester),
|
||
selectinload(SwapRequest.target_user),
|
||
selectinload(SwapRequest.accepted_by),
|
||
)
|
||
.order_by(SwapRequest.created_at.desc())
|
||
)
|
||
if not user.is_admin:
|
||
query = query.where(
|
||
(SwapRequest.requester_id == user.id)
|
||
| (SwapRequest.target_user_id == user.id)
|
||
| ((SwapRequest.target_user_id.is_(None)) & (SwapRequest.status == "open"))
|
||
| (SwapRequest.accepted_by_id == user.id)
|
||
)
|
||
swaps = db.scalars(query).all()
|
||
return request.app.state.templates.TemplateResponse(
|
||
request=request,
|
||
name="swaps.html",
|
||
context=_context(request, db, swaps=swaps, today=date.today()),
|
||
)
|
||
|
||
|
||
@router.post("/assignments/{assignment_id}/swap")
|
||
def create_swap(
|
||
assignment_id: int,
|
||
request: Request,
|
||
target_user_id: str = Form(""),
|
||
message: str = Form(""),
|
||
csrf_token: str = Form(...),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
_check_csrf(request, csrf_token)
|
||
user = _require_user(request, db)
|
||
assignment = db.scalar(
|
||
select(Assignment)
|
||
.where(Assignment.id == assignment_id)
|
||
.options(selectinload(Assignment.schedule), selectinload(Assignment.user))
|
||
)
|
||
if not assignment or assignment.user_id != user.id:
|
||
raise HTTPException(status_code=403, detail="You can only offer your own assignment")
|
||
if assignment.date < date.today():
|
||
raise HTTPException(status_code=400, detail="Past assignments cannot be changed")
|
||
|
||
existing = db.scalar(
|
||
select(SwapRequest.id).where(
|
||
SwapRequest.assignment_id == assignment.id,
|
||
SwapRequest.status.in_(["open", "pending_approval"]),
|
||
)
|
||
)
|
||
if existing:
|
||
_flash(request, "There is already an active change request for this assignment.", "error")
|
||
return _redirect("/swaps")
|
||
|
||
target = None
|
||
if target_user_id.strip():
|
||
target = db.get(User, int(target_user_id))
|
||
if not target or not target.is_active or target.id == user.id:
|
||
raise HTTPException(status_code=400, detail="Invalid target user")
|
||
|
||
swap = SwapRequest(
|
||
assignment_id=assignment.id,
|
||
requester_id=user.id,
|
||
target_user_id=target.id if target else None,
|
||
status="open",
|
||
message=message.strip() or None,
|
||
)
|
||
db.add(swap)
|
||
db.commit()
|
||
|
||
subject = f"Bar rota change request: {assignment.date:%d %b} – {assignment.schedule.name}"
|
||
body = (
|
||
f"{user.name} is asking for someone to take their {assignment.schedule.name} assignment "
|
||
f"on {assignment.date:%A, %d %B %Y}.\n\n"
|
||
f"{message.strip()}\n\n"
|
||
f"Open the rota: {settings.normalized_base_url}/swaps"
|
||
)
|
||
if target:
|
||
send_email_notification(db, target, "swap_requested", subject, body)
|
||
else:
|
||
candidates = db.scalars(
|
||
select(User).where(User.is_active.is_(True), User.id != user.id).order_by(User.name)
|
||
).all()
|
||
send_many(db, candidates, "swap_requested", subject, body)
|
||
|
||
_flash(request, "Change request created.")
|
||
return _redirect("/swaps")
|
||
|
||
|
||
@router.post("/swaps/{swap_id}/accept")
|
||
def accept_swap(
|
||
swap_id: int,
|
||
request: Request,
|
||
csrf_token: str = Form(...),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
_check_csrf(request, csrf_token)
|
||
user = _require_user(request, db)
|
||
swap = db.scalar(
|
||
select(SwapRequest)
|
||
.where(SwapRequest.id == swap_id)
|
||
.options(selectinload(SwapRequest.assignment).selectinload(Assignment.schedule), selectinload(SwapRequest.requester))
|
||
)
|
||
if not swap or swap.status != "open":
|
||
raise HTTPException(status_code=400, detail="This request is no longer open")
|
||
if swap.requester_id == user.id:
|
||
raise HTTPException(status_code=400, detail="You cannot accept your own request")
|
||
if swap.target_user_id is not None and swap.target_user_id != user.id:
|
||
raise HTTPException(status_code=403, detail="This request is for another user")
|
||
if swap.assignment.user_id != swap.requester_id:
|
||
swap.status = "cancelled"
|
||
db.commit()
|
||
raise HTTPException(status_code=409, detail="The assignment has already changed")
|
||
|
||
swap.accepted_by_id = user.id
|
||
if swap.assignment.schedule.requires_manager_approval:
|
||
swap.status = "pending_approval"
|
||
db.commit()
|
||
admins = db.scalars(select(User).where(User.is_admin.is_(True), User.is_active.is_(True))).all()
|
||
send_many(
|
||
db,
|
||
admins,
|
||
"swap_needs_approval",
|
||
"Bar rota change needs approval",
|
||
f"{user.name} accepted {swap.requester.name}'s {swap.assignment.schedule.name} assignment "
|
||
f"for {swap.assignment.date:%d %B %Y}. Approve it at {settings.normalized_base_url}/swaps",
|
||
)
|
||
send_email_notification(
|
||
db,
|
||
swap.requester,
|
||
"swap_accepted",
|
||
"Your rota change was accepted",
|
||
f"{user.name} accepted your request. It is now waiting for manager approval.",
|
||
)
|
||
_flash(request, "Accepted. Waiting for manager approval.")
|
||
else:
|
||
swap.assignment.user_id = user.id
|
||
swap.assignment.source = "swap"
|
||
swap.status = "approved"
|
||
db.commit()
|
||
send_email_notification(
|
||
db,
|
||
swap.requester,
|
||
"swap_approved",
|
||
"Your rota change is confirmed",
|
||
f"{user.name} will take {swap.assignment.schedule.name} on {swap.assignment.date:%d %B %Y}.",
|
||
)
|
||
send_email_notification(
|
||
db,
|
||
user,
|
||
"swap_approved",
|
||
"Rota assignment confirmed",
|
||
f"You are now assigned to {swap.assignment.schedule.name} on {swap.assignment.date:%d %B %Y}.",
|
||
)
|
||
_flash(request, "Assignment transferred to you.")
|
||
return _redirect("/swaps")
|
||
|
||
|
||
@router.post("/swaps/{swap_id}/reject")
|
||
def reject_swap(
|
||
swap_id: int,
|
||
request: Request,
|
||
csrf_token: str = Form(...),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
_check_csrf(request, csrf_token)
|
||
user = _require_user(request, db)
|
||
swap = db.scalar(
|
||
select(SwapRequest)
|
||
.where(SwapRequest.id == swap_id)
|
||
.options(selectinload(SwapRequest.requester))
|
||
)
|
||
if not swap or swap.status != "open":
|
||
raise HTTPException(status_code=400, detail="This request is no longer open")
|
||
if swap.target_user_id != user.id:
|
||
raise HTTPException(status_code=403, detail="Only the requested user can reject this request")
|
||
swap.status = "rejected"
|
||
db.commit()
|
||
send_email_notification(
|
||
db,
|
||
swap.requester,
|
||
"swap_rejected",
|
||
"Rota change request declined",
|
||
f"{user.name} declined your rota change request.",
|
||
)
|
||
_flash(request, "Request declined.")
|
||
return _redirect("/swaps")
|
||
|
||
|
||
@router.post("/swaps/{swap_id}/cancel")
|
||
def cancel_swap(
|
||
swap_id: int,
|
||
request: Request,
|
||
csrf_token: str = Form(...),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
_check_csrf(request, csrf_token)
|
||
user = _require_user(request, db)
|
||
swap = db.get(SwapRequest, swap_id)
|
||
if not swap or swap.requester_id != user.id or swap.status not in {"open", "pending_approval"}:
|
||
raise HTTPException(status_code=403, detail="Cannot cancel this request")
|
||
swap.status = "cancelled"
|
||
db.commit()
|
||
_flash(request, "Request cancelled.")
|
||
return _redirect("/swaps")
|
||
|
||
|
||
@router.post("/swaps/{swap_id}/approve")
|
||
def approve_swap(
|
||
swap_id: int,
|
||
request: Request,
|
||
csrf_token: str = Form(...),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
_check_csrf(request, csrf_token)
|
||
_require_admin(request, db)
|
||
swap = db.scalar(
|
||
select(SwapRequest)
|
||
.where(SwapRequest.id == swap_id)
|
||
.options(
|
||
selectinload(SwapRequest.assignment).selectinload(Assignment.schedule),
|
||
selectinload(SwapRequest.requester),
|
||
selectinload(SwapRequest.accepted_by),
|
||
)
|
||
)
|
||
if not swap or swap.status != "pending_approval" or not swap.accepted_by:
|
||
raise HTTPException(status_code=400, detail="Request is not ready for approval")
|
||
if swap.assignment.user_id != swap.requester_id:
|
||
swap.status = "cancelled"
|
||
db.commit()
|
||
raise HTTPException(status_code=409, detail="The assignment has already changed")
|
||
|
||
swap.assignment.user_id = swap.accepted_by_id
|
||
swap.assignment.source = "swap"
|
||
swap.status = "approved"
|
||
db.commit()
|
||
subject = "Rota change approved"
|
||
body = (
|
||
f"The change for {swap.assignment.schedule.name} on {swap.assignment.date:%d %B %Y} was approved. "
|
||
f"{swap.accepted_by.name} is now assigned."
|
||
)
|
||
send_many(db, [swap.requester, swap.accepted_by], "swap_approved", subject, body)
|
||
_flash(request, "Change approved and schedule updated.")
|
||
return _redirect("/swaps")
|
||
|
||
|
||
@router.post("/swaps/{swap_id}/admin-reject")
|
||
def admin_reject_swap(
|
||
swap_id: int,
|
||
request: Request,
|
||
csrf_token: str = Form(...),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
_check_csrf(request, csrf_token)
|
||
_require_admin(request, db)
|
||
swap = db.scalar(
|
||
select(SwapRequest)
|
||
.where(SwapRequest.id == swap_id)
|
||
.options(selectinload(SwapRequest.requester), selectinload(SwapRequest.accepted_by))
|
||
)
|
||
if not swap or swap.status != "pending_approval":
|
||
raise HTTPException(status_code=400, detail="Request is not awaiting approval")
|
||
swap.status = "rejected"
|
||
db.commit()
|
||
users = [u for u in [swap.requester, swap.accepted_by] if u]
|
||
send_many(db, users, "swap_rejected", "Rota change not approved", "The manager did not approve the requested rota change.")
|
||
_flash(request, "Change rejected.")
|
||
return _redirect("/swaps")
|
||
|
||
|
||
@router.get("/admin/users", response_class=HTMLResponse)
|
||
def admin_users(request: Request, db: Session = Depends(get_db)):
|
||
_require_admin(request, db)
|
||
users = db.scalars(select(User).order_by(User.name)).all()
|
||
return request.app.state.templates.TemplateResponse(
|
||
request=request,
|
||
name="admin_users.html",
|
||
context=_context(request, db, users=users),
|
||
)
|
||
|
||
|
||
@router.post("/admin/users")
|
||
def admin_create_user(
|
||
request: Request,
|
||
name: str = Form(...),
|
||
username: str = Form(""),
|
||
email: str = Form(""),
|
||
phone: str = Form(""),
|
||
password: str = Form(""),
|
||
is_admin: str = Form(""),
|
||
notify_email: str = Form(""),
|
||
csrf_token: str = Form(...),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
_check_csrf(request, csrf_token)
|
||
_require_admin(request, db)
|
||
username_value = username.strip().lower() or None
|
||
email_value = email.strip().lower() or None
|
||
password_value = password or None
|
||
if bool(username_value) != bool(password_value):
|
||
_flash(request, "Set both a username and a password, or leave both blank.", "error")
|
||
return _redirect("/admin/users")
|
||
if username_value and db.scalar(select(User.id).where(User.username == username_value)):
|
||
_flash(request, "Username already exists.", "error")
|
||
return _redirect("/admin/users")
|
||
if email_value and db.scalar(select(User.id).where(User.email == email_value)):
|
||
_flash(request, "Email address already exists.", "error")
|
||
return _redirect("/admin/users")
|
||
try:
|
||
password_hash = hash_password(password_value) if password_value else None
|
||
except ValueError as exc:
|
||
_flash(request, str(exc), "error")
|
||
return _redirect("/admin/users")
|
||
db.add(
|
||
User(
|
||
name=name.strip(),
|
||
username=username_value,
|
||
email=email_value,
|
||
phone=phone.strip() or None,
|
||
password_hash=password_hash,
|
||
is_admin=bool(is_admin),
|
||
is_active=True,
|
||
notify_email=bool(notify_email),
|
||
)
|
||
)
|
||
db.commit()
|
||
_flash(request, "User created.")
|
||
return _redirect("/admin/users")
|
||
|
||
|
||
@router.post("/admin/users/{user_id}/update")
|
||
def admin_update_user(
|
||
user_id: int,
|
||
request: Request,
|
||
name: str = Form(...),
|
||
username: str = Form(""),
|
||
email: str = Form(""),
|
||
phone: str = Form(""),
|
||
is_admin: str = Form(""),
|
||
is_active: str = Form(""),
|
||
notify_email: str = Form(""),
|
||
new_password: str = Form(""),
|
||
csrf_token: str = Form(...),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
_check_csrf(request, csrf_token)
|
||
admin = _require_admin(request, db)
|
||
user = db.get(User, user_id)
|
||
if not user:
|
||
raise HTTPException(status_code=404, detail="User not found")
|
||
email_value = email.strip().lower() or None
|
||
duplicate = db.scalar(select(User.id).where(User.email == email_value, User.id != user.id)) if email_value else None
|
||
if duplicate:
|
||
_flash(request, "That email address is already in use.", "error")
|
||
return _redirect("/admin/users")
|
||
|
||
username_value = username.strip().lower() or None
|
||
password_value = new_password.strip() or None
|
||
if username_value:
|
||
if user.password_hash is None and password_value is None:
|
||
_flash(request, "Set a new password to give this person a login.", "error")
|
||
return _redirect("/admin/users")
|
||
if db.scalar(select(User.id).where(User.username == username_value, User.id != user.id)):
|
||
_flash(request, "That username is already in use.", "error")
|
||
return _redirect("/admin/users")
|
||
elif password_value and user.id == admin.id:
|
||
_flash(request, "Add a username before setting a password for your own account.", "error")
|
||
return _redirect("/admin/users")
|
||
|
||
try:
|
||
password_hash = hash_password(password_value) if password_value else None
|
||
except ValueError as exc:
|
||
_flash(request, str(exc), "error")
|
||
return _redirect("/admin/users")
|
||
|
||
user.name = name.strip()
|
||
user.username = username_value
|
||
user.password_hash = (password_hash or user.password_hash) if username_value else None
|
||
user.email = email_value
|
||
user.phone = phone.strip() or None
|
||
user.notify_email = bool(notify_email)
|
||
if user.id != admin.id:
|
||
user.is_active = bool(is_active)
|
||
else:
|
||
user.is_active = True
|
||
if user.id != admin.id:
|
||
user.is_admin = bool(is_admin)
|
||
db.commit()
|
||
_flash(request, "User updated.")
|
||
return _redirect("/admin/users")
|
||
|
||
|
||
@router.get("/admin/schedules", response_class=HTMLResponse)
|
||
def admin_schedules(request: Request, db: Session = Depends(get_db)):
|
||
_require_admin(request, db)
|
||
cards = _schedule_groups(_all_schedules(db), db)
|
||
return request.app.state.templates.TemplateResponse(
|
||
request=request,
|
||
name="admin_schedules.html",
|
||
context=_context(request, db, cards=cards),
|
||
)
|
||
|
||
|
||
@router.get("/admin/schedules/{schedule_id}", response_class=HTMLResponse)
|
||
def admin_schedule(
|
||
schedule_id: int,
|
||
request: Request,
|
||
year: int | None = None,
|
||
month: int | None = None,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
_require_admin(request, db)
|
||
schedule = _get_schedule(db, schedule_id)
|
||
users = db.scalars(select(User).where(User.is_active.is_(True)).order_by(User.name)).all()
|
||
year, month = _parse_month(year, month)
|
||
assignments: list[Assignment] = []
|
||
prev = nxt = (year, month)
|
||
if schedule.schedule_type == "rotation":
|
||
assignments = [
|
||
row for row in ensure_month_assignments(db, year, month) if row.schedule_id == schedule.id
|
||
]
|
||
prev, nxt = _month_nav(year, month)
|
||
return request.app.state.templates.TemplateResponse(
|
||
request=request,
|
||
name="admin_schedule.html",
|
||
context=_context(
|
||
request,
|
||
db,
|
||
card=_schedule_card(schedule, db),
|
||
cards=_schedule_groups(_all_schedules(db), db),
|
||
users=users,
|
||
assignments=assignments,
|
||
pattern_days=list(range(1, 32)),
|
||
pattern=_pattern_map(db, schedule),
|
||
year=year,
|
||
month=month,
|
||
month_name=calendar.month_name[month],
|
||
prev=prev,
|
||
nxt=nxt,
|
||
),
|
||
)
|
||
|
||
|
||
def _pattern_map(db: Session, schedule: Schedule) -> dict[int, int]:
|
||
rows = db.scalars(
|
||
select(MonthlyPattern).where(MonthlyPattern.schedule_id == schedule.id)
|
||
).all()
|
||
return {row.day_of_month: row.user_id for row in rows if row.user_id is not None}
|
||
|
||
|
||
@router.post("/admin/schedules/{schedule_id}/pattern")
|
||
async def admin_schedule_pattern(
|
||
schedule_id: int,
|
||
request: Request,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
_require_admin(request, db)
|
||
form = await request.form()
|
||
_check_csrf(request, str(form.get("csrf_token", "")))
|
||
schedule = _get_schedule(db, schedule_id)
|
||
if schedule.schedule_type == "rotation":
|
||
raise HTTPException(status_code=400, detail="Rotations use their own order, not a month pattern")
|
||
users_by_id = {u.id for u in db.scalars(select(User).where(User.is_active.is_(True))).all()}
|
||
|
||
existing = {
|
||
row.day_of_month: row
|
||
for row in db.scalars(
|
||
select(MonthlyPattern).where(MonthlyPattern.schedule_id == schedule.id)
|
||
).all()
|
||
}
|
||
saved = 0
|
||
for day in range(1, 32):
|
||
raw = str(form.get(f"day_{day}", "")).strip()
|
||
user_id = int(raw) if raw else None
|
||
if user_id is not None and user_id not in users_by_id:
|
||
continue
|
||
row = existing.get(day)
|
||
if row is None:
|
||
if user_id is None:
|
||
continue
|
||
db.add(MonthlyPattern(schedule_id=schedule.id, day_of_month=day, user_id=user_id))
|
||
saved += 1
|
||
elif row.user_id != user_id:
|
||
row.user_id = user_id
|
||
saved += 1
|
||
db.commit()
|
||
touched = _apply_pattern_to_open_rows(db, schedule)
|
||
_flash(
|
||
request,
|
||
f"Saved the repeating night pattern ({saved} change(s)). "
|
||
f"Updated {touched} unassigned night(s) in this month and the next; nights already assigned were left alone.",
|
||
)
|
||
return _redirect(f"/admin/schedules/{schedule.id}")
|
||
|
||
|
||
def _apply_pattern_to_open_rows(db: Session, schedule: Schedule) -> int:
|
||
"""Fill empty generated nights from the pattern; never touch existing assignees."""
|
||
patterns = _pattern_map(db, schedule)
|
||
if not patterns:
|
||
return 0
|
||
today = date.today()
|
||
last_month = (today.month % 12) + 1
|
||
last_year = today.year + (1 if today.month == 12 else 0)
|
||
rows = db.scalars(
|
||
select(Assignment).where(
|
||
Assignment.schedule_id == schedule.id,
|
||
Assignment.user_id.is_(None),
|
||
Assignment.source == "generated",
|
||
Assignment.date >= today.replace(day=1),
|
||
Assignment.date <= date(last_year, last_month, calendar.monthrange(last_year, last_month)[1]),
|
||
)
|
||
).all()
|
||
touched = 0
|
||
for row in rows:
|
||
user_id = patterns.get(row.date.day)
|
||
if user_id and row.user_id != user_id:
|
||
row.user_id = user_id
|
||
touched += 1
|
||
if touched:
|
||
db.commit()
|
||
return touched
|
||
|
||
|
||
@router.post("/admin/schedules/{schedule_id}/settings")
|
||
def admin_schedule_settings(
|
||
schedule_id: int,
|
||
request: Request,
|
||
requires_manager_approval: str = Form(""),
|
||
is_active: str = Form(""),
|
||
csrf_token: str = Form(...),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
_check_csrf(request, csrf_token)
|
||
_require_admin(request, db)
|
||
schedule = db.get(Schedule, schedule_id)
|
||
if not schedule:
|
||
raise HTTPException(status_code=404, detail="Schedule not found")
|
||
schedule.requires_manager_approval = bool(requires_manager_approval)
|
||
schedule.is_active = bool(is_active)
|
||
db.commit()
|
||
_flash(request, "Schedule settings saved.")
|
||
return _redirect(f"/admin/schedules/{schedule.id}")
|
||
|
||
|
||
@router.post("/admin/schedules/{schedule_id}/rotation")
|
||
def admin_rotation_add(
|
||
schedule_id: int,
|
||
request: Request,
|
||
user_id: int = Form(...),
|
||
csrf_token: str = Form(...),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
_check_csrf(request, csrf_token)
|
||
_require_admin(request, db)
|
||
schedule = db.get(Schedule, schedule_id)
|
||
user = db.get(User, user_id)
|
||
if not schedule or schedule.schedule_type != "rotation" or not user or not user.is_active:
|
||
raise HTTPException(status_code=400, detail="Invalid rotation member")
|
||
exists = db.scalar(
|
||
select(RotationMember.id).where(
|
||
RotationMember.schedule_id == schedule.id,
|
||
RotationMember.user_id == user.id,
|
||
)
|
||
)
|
||
if exists:
|
||
_flash(request, "That person is already in this rotation.", "error")
|
||
return _redirect(f"/admin/schedules/{schedule.id}")
|
||
max_position = db.scalar(
|
||
select(func.max(RotationMember.position)).where(RotationMember.schedule_id == schedule.id)
|
||
) or 0
|
||
db.add(RotationMember(schedule_id=schedule.id, user_id=user.id, position=max_position + 1))
|
||
db.commit()
|
||
_flash(request, "Rotation member added.")
|
||
return _redirect(f"/admin/schedules/{schedule.id}")
|
||
|
||
|
||
@router.post("/admin/rotation/{member_id}/remove")
|
||
def admin_rotation_remove(
|
||
member_id: int,
|
||
request: Request,
|
||
csrf_token: str = Form(...),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
_check_csrf(request, csrf_token)
|
||
_require_admin(request, db)
|
||
member = db.get(RotationMember, member_id)
|
||
if not member:
|
||
raise HTTPException(status_code=404, detail="Rotation member not found")
|
||
schedule_id = member.schedule_id
|
||
db.delete(member)
|
||
db.commit()
|
||
members = db.scalars(
|
||
select(RotationMember).where(RotationMember.schedule_id == schedule_id).order_by(RotationMember.position)
|
||
).all()
|
||
for idx, row in enumerate(members, start=1):
|
||
row.position = idx
|
||
db.commit()
|
||
_flash(request, "Rotation member removed.")
|
||
return _redirect(f"/admin/schedules/{schedule_id}")
|
||
|
||
|
||
@router.post("/admin/rotation/{member_id}/move")
|
||
def admin_rotation_move(
|
||
member_id: int,
|
||
request: Request,
|
||
direction: str = Form(...),
|
||
csrf_token: str = Form(...),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
_check_csrf(request, csrf_token)
|
||
_require_admin(request, db)
|
||
member = db.get(RotationMember, member_id)
|
||
if not member or direction not in {"up", "down"}:
|
||
raise HTTPException(status_code=400, detail="Invalid move")
|
||
target_position = member.position - 1 if direction == "up" else member.position + 1
|
||
other = db.scalar(
|
||
select(RotationMember).where(
|
||
RotationMember.schedule_id == member.schedule_id,
|
||
RotationMember.position == target_position,
|
||
)
|
||
)
|
||
if other:
|
||
temporary = 1000000 + member.id
|
||
member.position = temporary
|
||
db.flush()
|
||
other.position = target_position + 1 if direction == "up" else target_position - 1
|
||
db.flush()
|
||
member.position = target_position
|
||
db.commit()
|
||
return _redirect(f"/admin/schedules/{member.schedule_id}")
|
||
|
||
|
||
@router.get("/admin/month", response_class=HTMLResponse)
|
||
def admin_month(request: Request, db: Session = Depends(get_db)):
|
||
_require_admin(request, db)
|
||
# Each schedule is now configured on its own page.
|
||
return _redirect("/admin/schedules")
|
||
|
||
|
||
def _save_assignments(db: Session, assignments: Sequence[Assignment], form) -> int:
|
||
users_by_id = {u.id: u for u in db.scalars(select(User).where(User.is_active.is_(True))).all()}
|
||
changed: list[tuple[Assignment, User | None, User | None]] = []
|
||
for assignment in assignments:
|
||
raw = str(form.get(f"assignment_{assignment.id}", "")).strip()
|
||
new_user_id = int(raw) if raw else None
|
||
if new_user_id is not None and new_user_id not in users_by_id:
|
||
continue
|
||
if assignment.user_id != new_user_id:
|
||
old_user = assignment.user
|
||
new_user = users_by_id.get(new_user_id) if new_user_id else None
|
||
assignment.user_id = new_user_id
|
||
assignment.source = "manual"
|
||
changed.append((assignment, old_user, new_user))
|
||
db.commit()
|
||
|
||
_notify_assignment_changes(db, changed)
|
||
return len(changed)
|
||
|
||
|
||
def _notify_assignment_changes(db: Session, changed: Sequence[tuple[Assignment, User | None, User | None]]) -> None:
|
||
for assignment, old_user, new_user in changed:
|
||
subject = f"Bar rota updated: {assignment.date:%d %b} – {assignment.schedule.name}"
|
||
if new_user:
|
||
send_email_notification(
|
||
db,
|
||
new_user,
|
||
"assignment_updated",
|
||
subject,
|
||
f"You are assigned to {assignment.schedule.name} on {assignment.date:%A, %d %B %Y}.\n\n"
|
||
f"View the rota: {settings.normalized_base_url}/",
|
||
)
|
||
if old_user and (not new_user or old_user.id != new_user.id):
|
||
send_email_notification(
|
||
db,
|
||
old_user,
|
||
"assignment_changed",
|
||
subject,
|
||
f"You are no longer assigned to {assignment.schedule.name} on {assignment.date:%A, %d %B %Y}.\n\n"
|
||
f"View the rota: {settings.normalized_base_url}/",
|
||
)
|
||
|
||
|
||
def _month_assignments(db: Session, year: int, month: int) -> list[Assignment]:
|
||
first, last = month_bounds(year, month)
|
||
return list(
|
||
db.scalars(
|
||
select(Assignment)
|
||
.where(Assignment.date >= first, Assignment.date <= last)
|
||
.options(selectinload(Assignment.user), selectinload(Assignment.schedule))
|
||
.order_by(Assignment.date)
|
||
).all()
|
||
)
|
||
|
||
|
||
@router.post("/admin/schedules/{schedule_id}/assignments")
|
||
async def admin_schedule_assignments(
|
||
schedule_id: int,
|
||
request: Request,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
_require_admin(request, db)
|
||
form = await request.form()
|
||
_check_csrf(request, str(form.get("csrf_token", "")))
|
||
schedule = _get_schedule(db, schedule_id)
|
||
year = int(form.get("year", date.today().year))
|
||
month = int(form.get("month", date.today().month))
|
||
assignments = [row for row in _month_assignments(db, year, month) if row.schedule_id == schedule.id]
|
||
changed = _save_assignments(db, assignments, form)
|
||
_flash(request, f"Saved {changed} assignment change(s).")
|
||
return _redirect(f"/admin/schedules/{schedule.id}?year={year}&month={month}")
|
||
|
||
|
||
def _apply_rotation_to_rows(db: Session, schedule: Schedule, assignments: Sequence[Assignment]) -> int:
|
||
"""Explicit admin edit: retain assignment IDs and invalidate requests when ownership changes."""
|
||
changed: list[tuple[Assignment, User | None, User | None]] = []
|
||
touched = 0
|
||
for assignment in assignments:
|
||
new_user = rotation_user_for_date(db, schedule, assignment.date)
|
||
new_user_id = new_user.id if new_user else None
|
||
if assignment.user_id == new_user_id and assignment.source == "generated":
|
||
continue
|
||
if assignment.user_id != new_user_id:
|
||
old_user = db.get(User, assignment.user_id) if assignment.user_id is not None else None
|
||
changed.append((assignment, old_user, new_user))
|
||
if assignment.id is not None:
|
||
requests = db.scalars(select(SwapRequest).where(
|
||
SwapRequest.assignment_id == assignment.id,
|
||
SwapRequest.status.in_(["open", "pending_approval"]),
|
||
)).all()
|
||
for swap in requests:
|
||
swap.status = "cancelled"
|
||
assignment.user_id = new_user_id
|
||
assignment.user = new_user
|
||
assignment.source = "generated"
|
||
touched += 1
|
||
db.commit()
|
||
_notify_assignment_changes(db, changed)
|
||
return touched
|
||
|
||
|
||
@router.post("/admin/schedules/{schedule_id}/apply-rotation")
|
||
def admin_apply_rotation(
|
||
schedule_id: int,
|
||
request: Request,
|
||
year: int = Form(...),
|
||
month: int = Form(...),
|
||
csrf_token: str = Form(...),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
_require_admin(request, db)
|
||
_check_csrf(request, csrf_token)
|
||
schedule = _get_schedule(db, schedule_id)
|
||
if schedule.schedule_type != "rotation":
|
||
raise HTTPException(status_code=400, detail="This schedule does not use a rotation")
|
||
year, month = _parse_month(year, month)
|
||
rows = {row.date: row for row in _month_assignments(db, year, month) if row.schedule_id == schedule.id}
|
||
assignments = []
|
||
created = 0
|
||
for day in iter_month_dates(year, month):
|
||
if day.weekday() not in schedule.weekday_set:
|
||
continue
|
||
row = rows.get(day)
|
||
if row is None:
|
||
row = Assignment(schedule=schedule, date=day, source="generated")
|
||
db.add(row)
|
||
created += 1
|
||
if row.source == "generated":
|
||
assignments.append(row)
|
||
changed = _apply_rotation_to_rows(db, schedule, assignments)
|
||
_flash(request, f"Applied rotation for {calendar.month_name[month]} {year}: "
|
||
f"created {created} shift(s), updated {changed} assignment(s). Manual assignments and swaps were kept.")
|
||
return _redirect(f"/admin/schedules/{schedule.id}?year={year}&month={month}")
|
||
|
||
|
||
@router.post("/admin/schedules/{schedule_id}/assignments/{assignment_id}/use-rotation")
|
||
def admin_assignment_use_rotation(
|
||
schedule_id: int,
|
||
assignment_id: int,
|
||
request: Request,
|
||
csrf_token: str = Form(...),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
_require_admin(request, db)
|
||
_check_csrf(request, csrf_token)
|
||
schedule = _get_schedule(db, schedule_id)
|
||
if schedule.schedule_type != "rotation":
|
||
raise HTTPException(status_code=400, detail="This schedule does not use a rotation")
|
||
assignment = db.get(Assignment, assignment_id)
|
||
if assignment is None or assignment.schedule_id != schedule.id:
|
||
raise HTTPException(status_code=404, detail="Assignment not found")
|
||
if assignment.source != "manual" or assignment.date.weekday() not in schedule.weekday_set:
|
||
raise HTTPException(status_code=400, detail="Only manual rotation assignments can be reset")
|
||
_apply_rotation_to_rows(db, schedule, [assignment])
|
||
_flash(request, "Manual assignment removed; this shift now uses the rotation.")
|
||
return _redirect(f"/admin/schedules/{schedule.id}?year={assignment.date.year}&month={assignment.date.month}")
|
||
|
||
|
||
@router.get("/admin/notifications", response_class=HTMLResponse)
|
||
def admin_notifications(request: Request, db: Session = Depends(get_db)):
|
||
_require_admin(request, db)
|
||
logs = db.scalars(
|
||
select(NotificationLog)
|
||
.options(selectinload(NotificationLog.user))
|
||
.order_by(NotificationLog.created_at.desc())
|
||
.limit(200)
|
||
).all()
|
||
return request.app.state.templates.TemplateResponse(
|
||
request=request,
|
||
name="admin_notifications.html",
|
||
context=_context(request, db, logs=logs),
|
||
)
|