""" Portal Schedule — business hours awareness for state/provincial portals. Some portals restrict filing hours. Attempting automation outside these windows results in portal-unavailable errors, which are hard to distinguish from real failures. This module provides a consistent interface to check availability and compute the next open window so the job server can defer rather than fail. Known restricted portals: BC Corporate Online: Mon-Sat 06:00-22:00 PT, Sun 13:00-22:00 PT IRS EIN Assistant: Mon-Fri 07:00-22:00 ET All others: 24/7 (None schedule = always available) Usage: schedule = PortalSchedule.from_config(config["portal_schedule"]) available, next_open = schedule.is_available() if not available: job.defer_until(next_open) """ from __future__ import annotations import logging import random from dataclasses import dataclass, field from datetime import datetime, timedelta from typing import Optional from zoneinfo import ZoneInfo LOG = logging.getLogger("formation.portal_schedule") # Day index: 0=Monday ... 6=Sunday (matches datetime.weekday()) DAY_NAMES = ["mon", "tue", "wed", "thu", "fri", "sat", "sun"] @dataclass class DayWindow: """Open/close hours for a single day of the week. None = closed all day.""" open_hour: int # 0-23 inclusive close_hour: int # 0-23 inclusive (exclusive end — portal closes AT this hour) # Pre-built schedule configs for known restricted portals BC_CORPORATE_ONLINE_SCHEDULE = { "timezone": "America/Vancouver", "jurisdiction": "BC", "closed_holidays": True, "hours": { "mon": [6, 22], "tue": [6, 22], "wed": [6, 22], "thu": [6, 22], "fri": [6, 22], "sat": [6, 22], "sun": [13, 22], }, } IRS_EIN_SCHEDULE = { "timezone": "America/New_York", "jurisdiction": "IRS", "closed_holidays": True, "hours": { "mon": [7, 22], "tue": [7, 22], "wed": [7, 22], "thu": [7, 22], "fri": [7, 22], "sat": None, # Closed "sun": None, # Closed }, } # Standard US state SOS portal — Mon-Fri 7am-11pm ET, closed weekends & federal holidays US_STATE_SOS_SCHEDULE = { "timezone": "America/New_York", "jurisdiction": "US", "closed_holidays": True, "hours": { "mon": [7, 23], "tue": [7, 23], "wed": [7, 23], "thu": [7, 23], "fri": [7, 23], "sat": None, "sun": None, }, } @dataclass class PortalSchedule: """ Defines the business hours of a filing portal. Attributes: timezone: IANA timezone string (e.g. 'America/Vancouver') hours: Dict mapping day name to [open, close] hours or None if closed. jurisdiction: Holiday jurisdiction code: 'US', 'CA', 'BC', 'IRS', or None (no holiday check). closed_holidays: If True (default), treat holidays as closed days even if hours are defined. """ timezone: str hours: dict[str, Optional[list[int]]] # day -> [open_hour, close_hour] | None jurisdiction: Optional[str] = None # 'US', 'CA', 'BC', 'IRS' closed_holidays: bool = True @classmethod def always_open(cls) -> "PortalSchedule": """Return a schedule that is always available (24/7 portals, no holidays).""" return cls( timezone="UTC", hours={d: [0, 24] for d in DAY_NAMES}, jurisdiction=None, closed_holidays=False, ) @classmethod def from_config(cls, config: Optional[dict]) -> "PortalSchedule": """ Build a PortalSchedule from a config dict. If config is None, returns an always-open schedule (24/7 portal). Config keys: timezone: IANA timezone (default: 'UTC') hours: day -> [open, close] | None jurisdiction: 'US' | 'CA' | 'BC' | 'IRS' | None closed_holidays: bool (default True) """ if config is None: return cls.always_open() return cls( timezone=config.get("timezone", "UTC"), hours=config.get("hours", {d: [0, 24] for d in DAY_NAMES}), jurisdiction=config.get("jurisdiction"), closed_holidays=config.get("closed_holidays", True), ) def _is_holiday_today(self, local_date) -> bool: """Check if the local date is a holiday in our jurisdiction.""" if not self.closed_holidays or not self.jurisdiction: return False try: from scripts.formation.holidays import is_holiday as _is_holiday return _is_holiday(local_date, jurisdiction=self.jurisdiction) except ImportError: LOG.warning("holidays module not available — skipping holiday check") return False def _holiday_name(self, local_date) -> Optional[str]: """Get the name of the holiday if it is one.""" if not self.closed_holidays or not self.jurisdiction: return None try: from scripts.formation.holidays import holiday_name as _holiday_name return _holiday_name(local_date, jurisdiction=self.jurisdiction) except ImportError: return None def is_available(self, at: Optional[datetime] = None) -> tuple[bool, Optional[datetime]]: """ Check if the portal is currently available. Checks: 1. Holiday calendar (jurisdiction-aware) 2. Day-of-week business hours Args: at: datetime to check (defaults to now). Should be timezone-naive UTC or tz-aware. Returns: (available: bool, next_open: datetime | None) next_open is UTC datetime of the next opening time (None if currently open). """ tz = ZoneInfo(self.timezone) now_utc = at or datetime.utcnow().replace(tzinfo=ZoneInfo("UTC")) if now_utc.tzinfo is None: now_utc = now_utc.replace(tzinfo=ZoneInfo("UTC")) now_local = now_utc.astimezone(tz) # Holiday check — closed all day if self._is_holiday_today(now_local.date()): hname = self._holiday_name(now_local.date()) or "holiday" LOG.info(f"Portal closed for {hname} ({now_local.date()})") next_open = self._next_open_after(now_local, tz) return False, next_open day_name = DAY_NAMES[now_local.weekday()] window = self.hours.get(day_name) if window is not None: open_h, close_h = window[0], window[1] if open_h <= now_local.hour < close_h: return True, None # Currently open # Not currently available — find next open window next_open = self._next_open_after(now_local, tz) return False, next_open def _next_open_after(self, now_local: datetime, tz: ZoneInfo) -> datetime: """Find the next datetime (in UTC) when the portal opens, skipping holidays.""" # Search up to 14 days ahead (handles multi-day holiday stretches like Christmas week) candidate = now_local.replace(minute=0, second=0, microsecond=0) for _ in range(14 * 24): # hourly steps, 14 days max candidate += timedelta(hours=1) # Skip holidays if self._is_holiday_today(candidate.date()): continue day_name = DAY_NAMES[candidate.weekday()] window = self.hours.get(day_name) if window is not None: open_h, close_h = window[0], window[1] if open_h <= candidate.hour < close_h: # Add small random offset (0-5 min) to avoid thundering herd jitter = timedelta(seconds=random.randint(0, 300)) return candidate.astimezone(ZoneInfo("UTC")) + jitter # Fallback: 24 hours from now (should never hit this) LOG.warning("Could not find next open window within 14 days — deferring 24h") return (now_local + timedelta(hours=24)).astimezone(ZoneInfo("UTC")) def minutes_until_open(self, at: Optional[datetime] = None) -> Optional[int]: """Return minutes until next open, or None if currently open.""" available, next_open = self.is_available(at) if available or next_open is None: return None now_utc = (at or datetime.utcnow()).replace(tzinfo=ZoneInfo("UTC")) if now_utc.tzinfo is None: now_utc = now_utc.replace(tzinfo=ZoneInfo("UTC")) delta = next_open - now_utc return max(0, int(delta.total_seconds() / 60))