""" Demo Date Utilities for Temporal Determinism Adjusts dates from seed data to be relative to demo session creation time """ from datetime import datetime, timezone, timedelta from typing import Optional import pytz # Fixed base reference date for all demo data # This is the "day 0" that all seed data is defined relative to BASE_REFERENCE_DATE = datetime(2025, 1, 15, 6, 0, 0, tzinfo=timezone.utc) def get_base_reference_date(session_created_at: Optional[datetime] = None) -> datetime: """ Get the base reference date for demo data. If session_created_at is provided, calculate relative to it. Otherwise, use current time (for backwards compatibility with seed scripts). Returns: Base reference date at 6 AM UTC """ if session_created_at: if session_created_at.tzinfo is None: session_created_at = session_created_at.replace(tzinfo=timezone.utc) # Reference is session creation time at 6 AM that day return session_created_at.replace( hour=6, minute=0, second=0, microsecond=0 ) # Fallback for seed scripts: use today at 6 AM now = datetime.now(timezone.utc) return now.replace(hour=6, minute=0, second=0, microsecond=0) def adjust_date_for_demo( original_date: Optional[datetime], session_created_at: datetime, base_reference_date: datetime = BASE_REFERENCE_DATE ) -> Optional[datetime]: """ Adjust a date from seed data to be relative to demo session creation time. Example: # Seed data created on 2025-12-13 06:00 # Stock expiration: 2025-12-28 06:00 (15 days from seed date) # Demo session created: 2025-12-16 10:00 # Base reference: 2025-12-16 06:00 # Result: 2025-12-31 10:00 (15 days from session date) """ if original_date is None: return None # Ensure timezone-aware datetimes if original_date.tzinfo is None: original_date = original_date.replace(tzinfo=timezone.utc) if session_created_at.tzinfo is None: session_created_at = session_created_at.replace(tzinfo=timezone.utc) if base_reference_date.tzinfo is None: base_reference_date = base_reference_date.replace(tzinfo=timezone.utc) # Calculate offset from base reference offset = original_date - base_reference_date # Apply offset to session creation date return session_created_at + offset def calculate_edge_case_times(session_created_at: datetime) -> dict: """ Calculate deterministic edge case times for demo sessions. These times are designed to always create specific demo scenarios: - One late delivery (should have arrived hours ago) - One overdue production batch (should have started hours ago) - One in-progress batch (started recently) - One upcoming batch (starts soon) - One arriving-soon delivery (arrives in a few hours) Returns: { 'late_delivery_expected': session - 4h, 'overdue_batch_planned_start': session - 2h, 'in_progress_batch_actual_start': session - 1h45m, 'upcoming_batch_planned_start': session + 1h30m, 'arriving_soon_delivery_expected': session + 2h30m, 'evening_batch_planned_start': today 17:00, 'tomorrow_morning_planned_start': tomorrow 05:00 } """ if session_created_at.tzinfo is None: session_created_at = session_created_at.replace(tzinfo=timezone.utc) # Calculate today at 6 AM (base reference) base_reference = get_base_reference_date(session_created_at) # Calculate tomorrow at 6 AM tomorrow_base = base_reference + timedelta(days=1) return { 'late_delivery_expected': session_created_at - timedelta(hours=4), 'overdue_batch_planned_start': session_created_at - timedelta(hours=2), 'in_progress_batch_actual_start': session_created_at - timedelta(hours=1, minutes=45), 'upcoming_batch_planned_start': session_created_at + timedelta(hours=1, minutes=30), 'arriving_soon_delivery_expected': session_created_at + timedelta(hours=2, minutes=30), 'evening_batch_planned_start': base_reference.replace(hour=17, minute=0, second=0, microsecond=0), 'tomorrow_morning_planned_start': tomorrow_base.replace(hour=5, minute=0, second=0, microsecond=0) } def ensure_future_time( target_time: datetime, reference_time: datetime, min_hours_ahead: float = 1.0 ) -> datetime: """ Ensure a target time is in the future relative to reference time. If target_time is in the past or too close to reference_time, shift it forward by at least min_hours_ahead. """ if target_time.tzinfo is None: target_time = target_time.replace(tzinfo=timezone.utc) if reference_time.tzinfo is None: reference_time = reference_time.replace(tzinfo=timezone.utc) time_diff = (target_time - reference_time).total_seconds() / 3600 if time_diff < min_hours_ahead: # Shift forward to ensure minimum hours ahead return reference_time + timedelta(hours=min_hours_ahead) return target_time def resolve_time_marker( time_marker: str, session_created_at: datetime, base_reference_date: datetime = BASE_REFERENCE_DATE ) -> datetime: """ Resolve time markers like "BASE_TS + 1h30m" to actual datetimes. Supports markers in the format: - "BASE_TS + XhYm" (e.g., "BASE_TS + 1h30m") - "BASE_TS - XhYm" (e.g., "BASE_TS - 2h") - "BASE_TS + Xd" (e.g., "BASE_TS + 2d") - "BASE_TS - Xd" (e.g., "BASE_TS - 1d") Args: time_marker: Time marker string to resolve session_created_at: Demo session creation time base_reference_date: Base reference date for calculation Returns: Resolved datetime adjusted for demo session Raises: ValueError: If time_marker format is invalid Examples: >>> resolve_time_marker("BASE_TS + 1h30m", session_time) >>> # Returns session_created_at + 1h30m >>> resolve_time_marker("BASE_TS - 2h", session_time) >>> # Returns session_created_at - 2h """ if not time_marker or not time_marker.startswith("BASE_TS"): raise ValueError(f"Invalid time marker format: {time_marker}") # Extract the offset part offset_part = time_marker[7:].strip() # Remove "BASE_TS " if not offset_part: # Just "BASE_TS" - return session_created_at return session_created_at # Parse operator and value operator = offset_part[0] value_part = offset_part[1:].strip() if operator not in ['+', '-']: raise ValueError(f"Invalid operator in time marker: {time_marker}") # Parse time components days = 0 hours = 0 minutes = 0 if 'd' in value_part: # Handle days day_part, rest = value_part.split('d', 1) days = int(day_part) value_part = rest if 'h' in value_part: # Handle hours hour_part, rest = value_part.split('h', 1) hours = int(hour_part) value_part = rest if 'm' in value_part: # Handle minutes minute_part = value_part.split('m', 1)[0] minutes = int(minute_part) # Calculate offset offset = timedelta(days=days, hours=hours, minutes=minutes) if operator == '+': return session_created_at + offset else: return session_created_at - offset def shift_to_session_time( original_offset_days: int, original_hour: int, original_minute: int, session_created_at: datetime, base_reference: Optional[datetime] = None ) -> datetime: """ Shift a time from seed data to demo session time with same-day preservation. Ensures that: 1. Items scheduled for "today" (offset_days=0) remain on the same day as session creation 2. Future items stay in the future, past items stay in the past 3. Times don't shift to invalid moments (e.g., past times for pending items) Examples: # Session created at noon, item originally scheduled for morning >>> session = datetime(2025, 12, 12, 12, 0, tzinfo=timezone.utc) >>> result = shift_to_session_time(0, 6, 0, session) # Today at 06:00 >>> # Returns today at 13:00 (shifted forward to stay in future) # Session created at noon, item originally scheduled for evening >>> result = shift_to_session_time(0, 18, 0, session) # Today at 18:00 >>> # Returns today at 18:00 (already in future) """ if session_created_at.tzinfo is None: session_created_at = session_created_at.replace(tzinfo=timezone.utc) if base_reference is None: base_reference = get_base_reference_date(session_created_at) # Calculate original time original_time = base_reference.replace( hour=original_hour, minute=original_minute, second=0, microsecond=0 ) + timedelta(days=original_offset_days) # Calculate offset from base reference offset = original_time - base_reference # Apply offset to session creation date new_time = session_created_at + offset # Ensure the time is in the future for pending items if original_offset_days >= 0: # Future or today new_time = ensure_future_time(new_time, session_created_at, min_hours_ahead=0.5) return new_time def get_working_hours_time( target_date: datetime, hours_from_start: float = 2.0 ) -> datetime: """ Get a time within working hours (8 AM - 6 PM) for a given date. Args: target_date: The date to calculate time for hours_from_start: Hours from working day start (8 AM) Returns: Datetime within working hours """ if target_date.tzinfo is None: target_date = target_date.replace(tzinfo=timezone.utc) # Working hours: 8 AM - 6 PM (10 hours) working_start = target_date.replace(hour=8, minute=0, second=0, microsecond=0) working_end = target_date.replace(hour=18, minute=0, second=0, microsecond=0) # Calculate time within working hours result_time = working_start + timedelta(hours=hours_from_start) # Ensure it's within working hours if result_time > working_end: result_time = working_end return result_time def get_next_workday(date: datetime) -> datetime: """ Get the next workday (Monday-Friday), skipping weekends. If date is Friday, returns Monday. If date is Saturday, returns Monday. Otherwise returns next day. """ if date.tzinfo is None: date = date.replace(tzinfo=timezone.utc) next_day = date + timedelta(days=1) # Skip weekends while next_day.weekday() >= 5: # 5=Saturday, 6=Sunday next_day += timedelta(days=1) return next_day def get_previous_workday(date: datetime) -> datetime: """ Get the previous workday (Monday-Friday), skipping weekends. If date is Monday, returns Friday. If date is Sunday, returns Friday. Otherwise returns previous day. """ if date.tzinfo is None: date = date.replace(tzinfo=timezone.utc) prev_day = date - timedelta(days=1) # Skip weekends while prev_day.weekday() >= 5: # 5=Saturday, 6=Sunday prev_day -= timedelta(days=1) return prev_day def format_iso_with_timezone(dt: datetime) -> str: """ Format datetime as ISO 8601 with timezone, replacing Z with +00:00 for compatibility. """ if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) iso_string = dt.isoformat() return iso_string.replace('+00:00', 'Z') if iso_string.endswith('+00:00') else iso_string