""" 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) # ── Follow-up link (phase45) ───────────────────────────────────────────── # Set when this schedule was created as a follow-up of a specific completed # inspection ("Schedule Follow-up" in the iPad's history detail). The # inspection eventually started from this schedule inherits it as its # parent_inspection_id, so it lands as a true linked re-inspection — # pre-filled from the parent and clearing the parent's follow_up_required on # submit. NULL = an ordinary schedule, which is what every pre-phase45 row # is. parent_inspection_id = db.Column( db.Integer, db.ForeignKey('inspections.id', ondelete='SET NULL'), nullable=True, index=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]) # Explicit foreign_keys is required, not optional: inspections and # scheduled_inspections now reference each other (Inspection # .scheduled_inspection_id points here, parent_inspection_id points back), # so SQLAlchemy cannot infer the join for either side. Inspection # .scheduled_inspection is already declared the same way. parent_inspection = db.relationship('Inspection', foreign_keys=[parent_inspection_id]) 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'')