308 lines
14 KiB
Python
308 lines
14 KiB
Python
"""
|
||
app/models/scheduled_inspection.py
|
||
-----------------------------------
|
||
Planned/recurring inspection assignments (phase36).
|
||
|
||
A ScheduledInspection is a PLAN, not an inspection: it names a facility, a
|
||
template, the responsible inspector, and a due date. The inspector opens it
|
||
("Start"), which creates a normal in_progress Inspection linked back via
|
||
Inspection.scheduled_inspection_id; when that inspection is completed the
|
||
schedule is marked fulfilled — deactivated (one-time) or rolled forward to the
|
||
next occurrence (recurring).
|
||
|
||
Reminders are dispatched by the cron endpoint
|
||
POST /scheduled-inspections/run?token=DIGEST_SECRET:
|
||
- advance reminder to the inspector 1 day before the due date
|
||
- due reminder to the inspector on the due date
|
||
- overdue alert to admin/director once the due date passes uncompleted
|
||
The *_notified flags make each of those fire at most once per occurrence and
|
||
reset when a recurring schedule rolls forward.
|
||
|
||
Two dates, deliberately distinct (phase44):
|
||
next_due_date — mutable state. The next occurrence. Rewritten by fulfill()
|
||
after every completed inspection.
|
||
end_date — fixed boundary. The last date an occurrence may fall on,
|
||
set by the manager and never rewritten. NULL = forever.
|
||
"""
|
||
|
||
import calendar
|
||
from datetime import date, timedelta
|
||
|
||
from app import db
|
||
from app.utils.time_utils import now_eastern
|
||
|
||
|
||
FREQUENCY_CHOICES = ('once', 'daily', 'weekly', 'monthly')
|
||
|
||
# Monthly recurrence styles (phase43). Stored as VARCHAR, not ENUM, so adding a
|
||
# style later needs no 3-step MySQL ENUM dance (CLAUDE.md rule 3).
|
||
MONTH_MODE_DAY = 'day_of_month' # "the 15th of every month"
|
||
MONTH_MODE_NTH = 'nth_weekday' # "the 2nd Tuesday of every month"
|
||
|
||
# Python weekday numbering: Monday=0 … Sunday=6 (matches date.weekday()).
|
||
WEEKDAY_NAMES = ('Monday', 'Tuesday', 'Wednesday', 'Thursday',
|
||
'Friday', 'Saturday', 'Sunday')
|
||
WEEKDAY_ABBREV = ('Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat', 'Sun')
|
||
|
||
# nth_week: 1–4 are literal, 5 means "5th (or last if the month is short)",
|
||
# -1 means "last" explicitly.
|
||
NTH_WEEK_LABELS = {1: '1st', 2: '2nd', 3: '3rd', 4: '4th', 5: '5th', -1: 'Last'}
|
||
|
||
|
||
def _last_day_of(year, month):
|
||
return calendar.monthrange(year, month)[1]
|
||
|
||
|
||
def _shift_month(year, month, n=1):
|
||
"""Return (year, month) shifted by *n* months."""
|
||
idx = year * 12 + (month - 1) + n
|
||
return idx // 12, idx % 12 + 1
|
||
|
||
|
||
def _nth_weekday_of(year, month, weekday, nth):
|
||
"""Date of the *nth* *weekday* in a month.
|
||
|
||
``nth == -1`` means the last one. A requested 5th occurrence that does not
|
||
exist falls back to the 4th, so every month yields a valid date.
|
||
"""
|
||
last = _last_day_of(year, month)
|
||
if nth == -1:
|
||
d = date(year, month, last)
|
||
return d - timedelta(days=(d.weekday() - weekday) % 7)
|
||
first = date(year, month, 1)
|
||
day = 1 + ((weekday - first.weekday()) % 7) + (nth - 1) * 7
|
||
while day > last:
|
||
day -= 7
|
||
return date(year, month, day)
|
||
|
||
|
||
class ScheduledInspection(db.Model):
|
||
__tablename__ = 'scheduled_inspections'
|
||
|
||
id = db.Column(db.Integer, primary_key=True)
|
||
facility_id = db.Column(db.Integer,
|
||
db.ForeignKey('facilities.id', ondelete='CASCADE'),
|
||
nullable=False, index=True)
|
||
template_id = db.Column(db.Integer,
|
||
db.ForeignKey('inspection_templates.id', ondelete='CASCADE'),
|
||
nullable=False)
|
||
inspector_id = db.Column(db.Integer,
|
||
db.ForeignKey('users.id', ondelete='SET NULL'),
|
||
nullable=True, index=True)
|
||
frequency = db.Column(db.Enum(*FREQUENCY_CHOICES), nullable=False, default='once')
|
||
next_due_date = db.Column(db.Date, nullable=False, index=True)
|
||
# Fixed boundary set by the manager, never rewritten by the app — unlike
|
||
# next_due_date, which fulfill() advances after every completed inspection.
|
||
# NULL = repeat indefinitely. Only meaningful for recurring schedules; the
|
||
# create/edit routes force it to NULL when frequency == 'once'.
|
||
end_date = db.Column(db.Date, nullable=True)
|
||
active = db.Column(db.Boolean, nullable=False, default=True)
|
||
notes = db.Column(db.Text, nullable=True)
|
||
|
||
# ── Recurrence detail (phase43) ──────────────────────────────────────────
|
||
# weekly : CSV of Python weekday ints, e.g. '0,2,4' = Mon/Wed/Fri.
|
||
# NULL/empty falls back to the legacy "every 7 days" behaviour.
|
||
# monthly : month_mode picks which pair of columns applies —
|
||
# MONTH_MODE_DAY → day_of_month; MONTH_MODE_NTH → nth_week + nth_weekday.
|
||
# NULL falls back to the legacy "same day next month" behaviour.
|
||
weekdays = db.Column(db.String(20), nullable=True)
|
||
month_mode = db.Column(db.String(20), nullable=True)
|
||
day_of_month = db.Column(db.SmallInteger, nullable=True)
|
||
nth_week = db.Column(db.SmallInteger, nullable=True)
|
||
nth_weekday = db.Column(db.SmallInteger, nullable=True)
|
||
|
||
created_by = db.Column(db.Integer, db.ForeignKey('users.id', ondelete='SET NULL'),
|
||
nullable=True)
|
||
created_at = db.Column(db.DateTime, nullable=False, default=now_eastern)
|
||
last_completed_at = db.Column(db.DateTime, nullable=True)
|
||
|
||
# Per-occurrence reminder de-dup flags (reset when a recurring one rolls forward)
|
||
advance_notified = db.Column(db.Boolean, nullable=False, default=False)
|
||
due_notified = db.Column(db.Boolean, nullable=False, default=False)
|
||
overdue_notified = db.Column(db.Boolean, nullable=False, default=False)
|
||
|
||
# Relationships
|
||
facility = db.relationship('Facility', foreign_keys=[facility_id])
|
||
template = db.relationship('InspectionTemplate', foreign_keys=[template_id])
|
||
inspector = db.relationship('User', foreign_keys=[inspector_id])
|
||
creator = db.relationship('User', foreign_keys=[created_by])
|
||
|
||
FREQUENCY_LABELS = {
|
||
'once': 'One-time',
|
||
'daily': 'Daily',
|
||
'weekly': 'Weekly',
|
||
'monthly': 'Monthly',
|
||
}
|
||
|
||
@property
|
||
def frequency_label(self):
|
||
return self.FREQUENCY_LABELS.get(self.frequency, self.frequency)
|
||
|
||
# ── Recurrence accessors ─────────────────────────────────────────────────
|
||
|
||
@property
|
||
def weekday_list(self):
|
||
"""Selected weekdays as a sorted list of ints (Mon=0). [] if unset."""
|
||
if not self.weekdays:
|
||
return []
|
||
out = set()
|
||
for part in str(self.weekdays).split(','):
|
||
part = part.strip()
|
||
if part.lstrip('-').isdigit() and 0 <= int(part) <= 6:
|
||
out.add(int(part))
|
||
return sorted(out)
|
||
|
||
def set_weekdays(self, values):
|
||
"""Store an iterable of weekday ints as the CSV column (None if empty)."""
|
||
clean = sorted({int(v) for v in (values or []) if 0 <= int(v) <= 6})
|
||
self.weekdays = ','.join(str(v) for v in clean) or None
|
||
|
||
@property
|
||
def recurrence_label(self):
|
||
"""Human summary of the recurrence rule, e.g. 'Weekly · Mon, Wed, Fri'."""
|
||
base = self.frequency_label
|
||
if self.frequency == 'weekly':
|
||
days = self.weekday_list
|
||
if days:
|
||
return f"{base} · {', '.join(WEEKDAY_ABBREV[d] for d in days)}"
|
||
elif self.frequency == 'monthly':
|
||
if self.month_mode == MONTH_MODE_NTH and self.nth_week and self.nth_weekday is not None:
|
||
nth = NTH_WEEK_LABELS.get(self.nth_week, str(self.nth_week))
|
||
return f'{base} · {nth} {WEEKDAY_NAMES[self.nth_weekday]}'
|
||
if self.day_of_month:
|
||
return f'{base} · day {self.day_of_month}'
|
||
return base
|
||
|
||
# ── Date arithmetic ──────────────────────────────────────────────────────
|
||
|
||
@staticmethod
|
||
def _add_interval(d, frequency):
|
||
"""Return d advanced by one plain interval of *frequency*.
|
||
|
||
Fallback used when no day-of-week / day-of-month detail is configured
|
||
(legacy phase36 rows). Prefer :meth:`next_occurrence_after`.
|
||
"""
|
||
if frequency == 'daily':
|
||
return d + timedelta(days=1)
|
||
if frequency == 'weekly':
|
||
return d + timedelta(weeks=1)
|
||
if frequency == 'monthly':
|
||
year, month = _shift_month(d.year, d.month, 1)
|
||
return date(year, month, min(d.day, _last_day_of(year, month)))
|
||
return d # 'once' has no next interval
|
||
|
||
def next_occurrence_after(self, d):
|
||
"""First occurrence strictly after date *d*, honouring the day rules."""
|
||
if self.frequency == 'weekly':
|
||
days = self.weekday_list
|
||
if days:
|
||
for step in range(1, 8):
|
||
cand = d + timedelta(days=step)
|
||
if cand.weekday() in days:
|
||
return cand
|
||
elif self.frequency == 'monthly':
|
||
year, month = _shift_month(d.year, d.month, 1)
|
||
if self.month_mode == MONTH_MODE_NTH and self.nth_week and self.nth_weekday is not None:
|
||
return _nth_weekday_of(year, month, self.nth_weekday, self.nth_week)
|
||
if self.day_of_month:
|
||
return date(year, month, min(self.day_of_month, _last_day_of(year, month)))
|
||
return self._add_interval(d, self.frequency)
|
||
|
||
def align_due_date(self, d):
|
||
"""Snap *d* forward to the first date on/after it that fits the rule.
|
||
|
||
Lets a manager pick any start date and still get, say, Mon/Wed/Fri:
|
||
picking a Tuesday for a Mon/Wed/Fri schedule yields that Wednesday.
|
||
"""
|
||
if self.frequency == 'weekly':
|
||
days = self.weekday_list
|
||
if days:
|
||
for step in range(0, 7):
|
||
cand = d + timedelta(days=step)
|
||
if cand.weekday() in days:
|
||
return cand
|
||
elif self.frequency == 'monthly':
|
||
if self.month_mode == MONTH_MODE_NTH and self.nth_week and self.nth_weekday is not None:
|
||
cand = _nth_weekday_of(d.year, d.month, self.nth_weekday, self.nth_week)
|
||
elif self.day_of_month:
|
||
cand = date(d.year, d.month,
|
||
min(self.day_of_month, _last_day_of(d.year, d.month)))
|
||
else:
|
||
return d
|
||
if cand < d:
|
||
return self.next_occurrence_after(cand)
|
||
return cand
|
||
return d
|
||
|
||
def is_overdue(self, today=None):
|
||
today = today or now_eastern().date()
|
||
return self.active and self.next_due_date < today
|
||
|
||
# ── End-date boundary (phase44) ──────────────────────────────────────────
|
||
|
||
def is_within_end_date(self, d):
|
||
"""True if date *d* is on or before the end date (inclusive).
|
||
|
||
No end date means the schedule repeats indefinitely, so every date
|
||
qualifies.
|
||
"""
|
||
return self.end_date is None or d <= self.end_date
|
||
|
||
@property
|
||
def is_expired(self):
|
||
"""True once the end date has passed.
|
||
|
||
Independent of `active`: a schedule can be inactive because it expired
|
||
or because a manager switched it off, and the list view distinguishes
|
||
the two. Compare against the *end date* rather than `next_due_date`,
|
||
which may have been advanced past the boundary by fulfill().
|
||
"""
|
||
if self.end_date is None:
|
||
return False
|
||
return self.end_date < now_eastern().date()
|
||
|
||
def expire_if_past_end_date(self, today=None):
|
||
"""Deactivate a schedule whose end date has passed. Caller commits.
|
||
|
||
Returns True if this call changed anything. Needed because a schedule
|
||
can reach its end date *without ever being completed* — fulfill() never
|
||
runs, so the boundary would otherwise be checked nowhere and the cron
|
||
would keep firing overdue alerts forever. Called from run_reminders().
|
||
"""
|
||
today = today or now_eastern().date()
|
||
if self.active and self.end_date is not None and self.end_date < today:
|
||
self.active = False
|
||
return True
|
||
return False
|
||
|
||
def fulfill(self):
|
||
"""Mark this occurrence complete. One-time schedules deactivate;
|
||
recurring ones roll their due date forward past today and reset the
|
||
reminder flags. A recurring schedule whose next occurrence would fall
|
||
past its end date deactivates instead. Caller commits."""
|
||
self.last_completed_at = now_eastern()
|
||
if self.frequency == 'once':
|
||
self.active = False
|
||
return
|
||
# Recurring: advance until the next due date is in the future.
|
||
today = now_eastern().date()
|
||
nxt = self.next_occurrence_after(self.next_due_date)
|
||
guard = 0
|
||
while nxt <= today and guard < 400:
|
||
nxt = self.next_occurrence_after(nxt)
|
||
guard += 1
|
||
self.next_due_date = nxt
|
||
# Past the manager's boundary: this was the last occurrence. next_due_date
|
||
# is left at the computed value rather than clamped, so the row still
|
||
# shows which occurrence it stopped before.
|
||
if not self.is_within_end_date(nxt):
|
||
self.active = False
|
||
return
|
||
self.advance_notified = False
|
||
self.due_notified = False
|
||
self.overdue_notified = False
|
||
|
||
def __repr__(self):
|
||
return (f'<ScheduledInspection {self.id} facility={self.facility_id} '
|
||
f'freq={self.frequency} due={self.next_due_date}>')
|