277 lines
8.5 KiB
Python
277 lines
8.5 KiB
Python
# shared/utils/timezone_helper.py
|
|
"""
|
|
Timezone Utility Helper for Bakery Management System
|
|
|
|
Provides timezone-aware date/time utilities for accurate scheduling across
|
|
different geographic locations. All schedulers should use these utilities
|
|
to ensure consistent behavior.
|
|
"""
|
|
|
|
from datetime import datetime, date, time
|
|
from typing import Optional
|
|
from zoneinfo import ZoneInfo
|
|
import structlog
|
|
|
|
logger = structlog.get_logger()
|
|
|
|
|
|
class TimezoneHelper:
|
|
"""Helper class for timezone-aware operations"""
|
|
|
|
DEFAULT_TIMEZONE = "Europe/Madrid"
|
|
VALID_TIMEZONES = {
|
|
"Europe/Madrid", "Europe/London", "Europe/Paris", "Europe/Berlin",
|
|
"America/New_York", "America/Chicago", "America/Los_Angeles",
|
|
"Asia/Tokyo", "Asia/Shanghai", "Australia/Sydney",
|
|
"UTC"
|
|
}
|
|
|
|
@classmethod
|
|
def get_current_date_in_timezone(cls, timezone_str: str) -> date:
|
|
"""
|
|
Get current date in specified timezone
|
|
|
|
Args:
|
|
timezone_str: IANA timezone string (e.g., "Europe/Madrid")
|
|
|
|
Returns:
|
|
Current date in the specified timezone
|
|
"""
|
|
try:
|
|
tz = ZoneInfo(timezone_str)
|
|
return datetime.now(tz).date()
|
|
except Exception as e:
|
|
logger.warning(f"Invalid timezone {timezone_str}, using default",
|
|
error=str(e))
|
|
return datetime.now(ZoneInfo(cls.DEFAULT_TIMEZONE)).date()
|
|
|
|
@classmethod
|
|
def get_current_datetime_in_timezone(cls, timezone_str: str) -> datetime:
|
|
"""
|
|
Get current datetime in specified timezone
|
|
|
|
Args:
|
|
timezone_str: IANA timezone string
|
|
|
|
Returns:
|
|
Current datetime in the specified timezone
|
|
"""
|
|
try:
|
|
tz = ZoneInfo(timezone_str)
|
|
return datetime.now(tz)
|
|
except Exception as e:
|
|
logger.warning(f"Invalid timezone {timezone_str}, using default",
|
|
error=str(e))
|
|
return datetime.now(ZoneInfo(cls.DEFAULT_TIMEZONE))
|
|
|
|
@classmethod
|
|
def combine_date_time_in_timezone(
|
|
cls,
|
|
target_date: date,
|
|
target_time: time,
|
|
timezone_str: str
|
|
) -> datetime:
|
|
"""
|
|
Combine date and time in specified timezone
|
|
|
|
Args:
|
|
target_date: Date component
|
|
target_time: Time component
|
|
timezone_str: IANA timezone string
|
|
|
|
Returns:
|
|
Datetime combining date and time in specified timezone
|
|
"""
|
|
try:
|
|
tz = ZoneInfo(timezone_str)
|
|
return datetime.combine(target_date, target_time, tzinfo=tz)
|
|
except Exception as e:
|
|
logger.warning(f"Invalid timezone {timezone_str}, using default",
|
|
error=str(e))
|
|
tz = ZoneInfo(cls.DEFAULT_TIMEZONE)
|
|
return datetime.combine(target_date, target_time, tzinfo=tz)
|
|
|
|
@classmethod
|
|
def convert_to_utc(cls, dt: datetime) -> datetime:
|
|
"""
|
|
Convert datetime to UTC
|
|
|
|
Args:
|
|
dt: Datetime to convert (must be timezone-aware)
|
|
|
|
Returns:
|
|
Datetime in UTC timezone
|
|
"""
|
|
if dt.tzinfo is None:
|
|
logger.warning("Converting naive datetime to UTC, assuming UTC")
|
|
return dt.replace(tzinfo=ZoneInfo("UTC"))
|
|
|
|
return dt.astimezone(ZoneInfo("UTC"))
|
|
|
|
@classmethod
|
|
def convert_from_utc(cls, dt: datetime, target_timezone: str) -> datetime:
|
|
"""
|
|
Convert UTC datetime to target timezone
|
|
|
|
Args:
|
|
dt: UTC datetime
|
|
target_timezone: Target IANA timezone string
|
|
|
|
Returns:
|
|
Datetime in target timezone
|
|
"""
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=ZoneInfo("UTC"))
|
|
|
|
try:
|
|
tz = ZoneInfo(target_timezone)
|
|
return dt.astimezone(tz)
|
|
except Exception as e:
|
|
logger.warning(f"Invalid timezone {target_timezone}, using default",
|
|
error=str(e))
|
|
tz = ZoneInfo(cls.DEFAULT_TIMEZONE)
|
|
return dt.astimezone(tz)
|
|
|
|
@classmethod
|
|
def validate_timezone(cls, timezone_str: str) -> bool:
|
|
"""
|
|
Validate if timezone string is valid
|
|
|
|
Args:
|
|
timezone_str: IANA timezone string to validate
|
|
|
|
Returns:
|
|
True if valid, False otherwise
|
|
"""
|
|
try:
|
|
ZoneInfo(timezone_str)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
@classmethod
|
|
def get_timezone_offset_hours(cls, timezone_str: str) -> float:
|
|
"""
|
|
Get current UTC offset for timezone in hours
|
|
|
|
Args:
|
|
timezone_str: IANA timezone string
|
|
|
|
Returns:
|
|
UTC offset in hours (e.g., +2.0 for CEST)
|
|
"""
|
|
try:
|
|
tz = ZoneInfo(timezone_str)
|
|
now = datetime.now(tz)
|
|
offset_seconds = now.utcoffset().total_seconds()
|
|
return offset_seconds / 3600
|
|
except Exception as e:
|
|
logger.warning(f"Could not get offset for {timezone_str}",
|
|
error=str(e))
|
|
return 0.0
|
|
|
|
@classmethod
|
|
def is_business_hours(
|
|
cls,
|
|
dt: Optional[datetime] = None,
|
|
timezone_str: str = DEFAULT_TIMEZONE,
|
|
start_hour: int = 8,
|
|
end_hour: int = 20
|
|
) -> bool:
|
|
"""
|
|
Check if datetime is within business hours
|
|
|
|
Args:
|
|
dt: Datetime to check (defaults to now)
|
|
timezone_str: IANA timezone string
|
|
start_hour: Business hours start (24h format)
|
|
end_hour: Business hours end (24h format)
|
|
|
|
Returns:
|
|
True if within business hours, False otherwise
|
|
"""
|
|
if dt is None:
|
|
dt = cls.get_current_datetime_in_timezone(timezone_str)
|
|
elif dt.tzinfo is None:
|
|
# Assume it's in the target timezone
|
|
tz = ZoneInfo(timezone_str)
|
|
dt = dt.replace(tzinfo=tz)
|
|
else:
|
|
# Convert to target timezone
|
|
dt = cls.convert_from_utc(dt, timezone_str)
|
|
|
|
# Check if weekday (Monday=0, Sunday=6)
|
|
if dt.weekday() >= 5: # Saturday or Sunday
|
|
return False
|
|
|
|
# Check if within business hours
|
|
return start_hour <= dt.hour < end_hour
|
|
|
|
@classmethod
|
|
def get_next_business_day_at_time(
|
|
cls,
|
|
target_time: time,
|
|
timezone_str: str = DEFAULT_TIMEZONE,
|
|
from_datetime: Optional[datetime] = None
|
|
) -> datetime:
|
|
"""
|
|
Get next business day at specific time in timezone
|
|
|
|
Args:
|
|
target_time: Time to schedule (e.g., time(6, 0) for 6 AM)
|
|
timezone_str: IANA timezone string
|
|
from_datetime: Starting datetime (defaults to now)
|
|
|
|
Returns:
|
|
Next business day at target_time in specified timezone
|
|
"""
|
|
if from_datetime is None:
|
|
current = cls.get_current_datetime_in_timezone(timezone_str)
|
|
else:
|
|
current = cls.convert_from_utc(from_datetime, timezone_str)
|
|
|
|
# Start with next day
|
|
next_day = current.date()
|
|
next_datetime = cls.combine_date_time_in_timezone(
|
|
next_day, target_time, timezone_str
|
|
)
|
|
|
|
# If we haven't passed target_time today, use today
|
|
if current.time() < target_time:
|
|
next_datetime = cls.combine_date_time_in_timezone(
|
|
current.date(), target_time, timezone_str
|
|
)
|
|
|
|
# Skip weekends
|
|
while next_datetime.weekday() >= 5: # Saturday or Sunday
|
|
next_day = next_datetime.date()
|
|
from datetime import timedelta
|
|
next_day = next_day + timedelta(days=1)
|
|
next_datetime = cls.combine_date_time_in_timezone(
|
|
next_day, target_time, timezone_str
|
|
)
|
|
|
|
return next_datetime
|
|
|
|
|
|
# Convenience functions for common operations
|
|
|
|
def get_tenant_current_date(tenant_timezone: str = "Europe/Madrid") -> date:
|
|
"""Get current date for tenant's timezone"""
|
|
return TimezoneHelper.get_current_date_in_timezone(tenant_timezone)
|
|
|
|
|
|
def get_tenant_current_datetime(tenant_timezone: str = "Europe/Madrid") -> datetime:
|
|
"""Get current datetime for tenant's timezone"""
|
|
return TimezoneHelper.get_current_datetime_in_timezone(tenant_timezone)
|
|
|
|
|
|
def is_tenant_business_hours(tenant_timezone: str = "Europe/Madrid") -> bool:
|
|
"""Check if it's currently business hours for tenant"""
|
|
return TimezoneHelper.is_business_hours(timezone_str=tenant_timezone)
|
|
|
|
|
|
def validate_timezone(timezone_str: str) -> bool:
|
|
"""Validate timezone string"""
|
|
return TimezoneHelper.validate_timezone(timezone_str)
|