dodil/erp-brick-timesheets
DODIL ERP brick: timesheets
src/brick_timesheets/application/reports.py
"""What a project lead reads (TS-09) and the chase that keeps month end from being one (TS-15).
get_project_hours approved, billable, billed, unbilled, submitted-not-approved hours and cost of ONE project, per item and per person, from this
module's rows alone. Projects the caller leads (finance and the owner: all).
tick_ts_reminders the worker's pass: one reminder to each person with a week still open after it was due, and one to each approver of a week waiting
approval_reminder_days business days, keyed (kind, ref, business date), sent from pending rows through the mail capability."""
import json
from typing import Literal, Optional
from zoneinfo import ZoneInfo
from erp_platform import auth
from erp_platform.bricks import Refused
from brick_timesheets.application import use_cases as hub
from brick_timesheets.application.inputs import Day, Ref
from brick_timesheets.domain import hours as hr
from brick_timesheets.domain import reminders as rem
TS = hub.TS
PROJECT_HOURS, TICK = "ts:project_hours:read", "ts:reminder:tick"
@TS.use_case(kind="query", route="GET /project-hours", permission=PROJECT_HOURS, aliases={"date_from": "from", "date_to": "to"})
def get_project_hours(caller: dict, project_id: Ref, date_from: Day, date_to: Day) -> dict:
"""One project's hours and cost for the work dates asked (at most 366 days): approved (the standing revision of every week), billable, billed (what
released ar runs took), unbilled, submitted not yet approved, and cost - in total, per item and per person - from this module's own rows, never ar's or
the order's. A project lead sees the projects they lead; finance and the owner see every project. A project you may not see is a 404."""
start, end = hub._date(date_from, "from"), hub._date(date_to, "to")
if end < start or (end - start).days > 366:
raise hub._bounds("from and to are at most 366 days apart")
full = auth.has_permission(caller, hub.READ_ALL)
with hub.repository() as repo, hub._decide():
rows, standing = repo.sheets.project_rows(project_id, start, end)
if not full:
pid = hub._person_of(caller)
if not any(r["lead_person_id"] == pid for r in rows):
raise Refused(404, f"no hours on project {project_id} that you lead")
released = repo.billing.ar_released(sorted({r["sheet_id"] for r in rows}), project_id)
got = hr.project_hours(rows, standing, released)
fmt, money = hub._h, hub._m
return {"project_id": project_id, "from": start.isoformat(), "to": end.isoformat(), "currency": got["currency"], **hr.shown(got["total"], fmt, money),
"submitted_not_approved": fmt(got["waiting"]),
"items": [{"item_id": k or None, **hr.shown(v, fmt, money)} for k, v in sorted(got["items"].items())],
"people": [{"person_id": k, **hr.shown(v, fmt, money)} for k, v in sorted(got["people"].items(), key=lambda kv: int(kv[0]))]}
@TS.use_case(kind="command", route="POST /reminders/tick", permission=TICK, idempotent=True, tool=False)
def tick_ts_reminders(caller: dict) -> dict:
"""The worker's pass (no server-side scheduler; Mondays and daily): one reminder to the person of each week still open after it was due (submit_due,
default MON 10:00 in business_timezone), and one to the approver of each week submitted approval_reminder_days business days ago. Each is keyed (kind, ref,
business date) in the outbox: a second pass the same business date sends nothing, tomorrow's chases again. Written in one transaction, handed to the mail
capability after it; with no mail provider a reminder is a task for a person. A pass names what it could not address (no email on the sheet)."""
who = hub._who(caller)
with hub.repository() as repo, hub._decide():
today = hub._today(repo)
tz = repo.settings.timezone()
now_local = hub._now().astimezone(ZoneInfo(tz))
due = repo.settings.on("submit_due", today)
days = int(repo.settings.on("approval_reminder_days", today))
sender = caller.get("email") or "timesheets@localhost"
wanted, unaddressed = {}, []
for root in repo.sheets.waiting():
ref = f"{root['sheet_id']}:r{root['revision']}"
week = hub.sh.week_id(root["week_start"])
if root["status"] == "open":
if now_local < rem.due_at(root["week_start"], due, tz):
continue
if not root["person_email"]:
unaddressed.append(f"{ref}: no email for the person")
continue
wanted[rem.key("submit", ref, today)] = (root["sheet_id"], [root["person_email"]], f"Your timesheet for {week} is not submitted",
f"Week {week} ({root['week_start']}) is still open. Book what is missing and submit it.")
else:
since = root["submitted_at"].replace(tzinfo=ZoneInfo("UTC")).astimezone(ZoneInfo(tz)).date() if root["submitted_at"] else None
if since is None or rem.business_days_since(since, today) < days:
continue
if not root["approver_email"]:
unaddressed.append(f"{ref}: no email for the approver")
continue
wanted[rem.key("approval", ref, today)] = (root["sheet_id"], [root["approver_email"]], f"A timesheet is waiting for your approval: {root['sheet_id']}",
f"Week {week} of person {root['person_id']} has waited since {since}. Approve it or return it with a question.")
have = repo.outbox.many("notify", list(wanted))
new = {r: v for r, v in wanted.items() if r not in have}
repo.outbox.add_many([("notify", r, sid, {"from_email": sender, "to": to, "subject": subj, "body_text": text_}) for r, (sid, to, subj, text_) in new.items()],
hub._now())
sent = hub._drain(None, ("notify",), who) # every pending reminder: one a provider refused last time goes out now
return {"business_date": today.isoformat(), "reminders": len(new), "skipped_already_sent": len(wanted) - len(new), "unaddressed": unaddressed, "sent": sent}
@TS.use_case(kind="query", route="GET /reminders", permission=hub.READ_ALL)
def list_ts_reminders(caller: dict, status: Optional[Literal["pending", "sent", "manual"]] = None) -> dict:
"""The reminders the pass wrote (finance and the owner): sent through the mail capability, pending a retry, or manual - no mail provider is wired, so a
person sends them."""
with hub.repository() as repo:
return {"reminders": [{"ref": r["ref"], "status": r["status"], "sheet_id": r["sheet_id"], "to": json.loads(r["payload_json"])["to"],
"subject": json.loads(r["payload_json"])["subject"], "attempts": r["attempts"], "note": r["last_error"]}
for r in repo.outbox.listed("notify", status)]}