Jul 26 - Update scheduled inspection settings (weekly/monthly)

This commit is contained in:
2026-07-26 13:47:17 -04:00
parent 0c63c45b21
commit 16858108b9
10 changed files with 488 additions and 23 deletions
+145 -10
View File
@@ -19,13 +19,56 @@ The *_notified flags make each of those fire at most once per occurrence and
reset when a recurring schedule rolls forward.
"""
from datetime import timedelta
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: 14 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'
@@ -45,6 +88,18 @@ class ScheduledInspection(db.Model):
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)
@@ -72,22 +127,102 @@ class ScheduledInspection(db.Model):
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 interval of the given 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':
# Add ~1 month by stepping 2831 days to the same day-of-month where possible.
month = d.month + 1
year = d.year + (1 if month > 12 else 0)
month = 1 if month > 12 else month
day = min(d.day, 28) # clamp to avoid invalid dates (e.g. Feb 30)
return d.replace(year=year, month=month, day=day)
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
@@ -102,10 +237,10 @@ class ScheduledInspection(db.Model):
return
# Recurring: advance until the next due date is in the future.
today = now_eastern().date()
nxt = self._add_interval(self.next_due_date, self.frequency)
nxt = self.next_occurrence_after(self.next_due_date)
guard = 0
while nxt <= today and guard < 400:
nxt = self._add_interval(nxt, self.frequency)
nxt = self.next_occurrence_after(nxt)
guard += 1
self.next_due_date = nxt
self.advance_notified = False