158 lines
4.3 KiB
Python
Executable File
158 lines
4.3 KiB
Python
Executable File
"""
|
|
Business Domain DateTime Logic
|
|
|
|
Business-specific datetime operations including business hours,
|
|
business days, and tenant-specific datetime handling.
|
|
"""
|
|
|
|
from datetime import datetime, time, timedelta
|
|
from typing import Optional
|
|
from zoneinfo import ZoneInfo
|
|
import structlog
|
|
|
|
from .constants import DEFAULT_TIMEZONE, BUSINESS_HOURS_DEFAULT
|
|
from .timezone import get_current_datetime_in_tz, combine_date_time_in_tz, convert_from_utc
|
|
|
|
logger = structlog.get_logger()
|
|
|
|
|
|
def is_business_hours(
|
|
dt: Optional[datetime] = None,
|
|
tz: str = DEFAULT_TIMEZONE,
|
|
start_hour: int = BUSINESS_HOURS_DEFAULT["start_hour"],
|
|
end_hour: int = BUSINESS_HOURS_DEFAULT["end_hour"],
|
|
include_weekends: bool = BUSINESS_HOURS_DEFAULT["include_weekends"]
|
|
) -> bool:
|
|
"""
|
|
Check if datetime is within business hours.
|
|
|
|
Args:
|
|
dt: Datetime to check (defaults to now in timezone)
|
|
tz: IANA timezone string
|
|
start_hour: Business hours start (24h format)
|
|
end_hour: Business hours end (24h format)
|
|
include_weekends: Whether to include weekends as business days
|
|
|
|
Returns:
|
|
True if within business hours, False otherwise
|
|
"""
|
|
if dt is None:
|
|
dt = get_current_datetime_in_tz(tz)
|
|
elif dt.tzinfo is None:
|
|
zone = ZoneInfo(tz)
|
|
dt = dt.replace(tzinfo=zone)
|
|
else:
|
|
dt = convert_from_utc(dt, tz)
|
|
|
|
if not include_weekends and dt.weekday() >= 5:
|
|
return False
|
|
|
|
return start_hour <= dt.hour < end_hour
|
|
|
|
|
|
def next_business_day(
|
|
from_dt: Optional[datetime] = None,
|
|
target_time: time = time(9, 0),
|
|
tz: str = DEFAULT_TIMEZONE
|
|
) -> datetime:
|
|
"""
|
|
Get next business day at specific time in timezone.
|
|
|
|
Args:
|
|
from_dt: Starting datetime (defaults to now in timezone)
|
|
target_time: Time to schedule (e.g., time(9, 0) for 9 AM)
|
|
tz: IANA timezone string
|
|
|
|
Returns:
|
|
Next business day at target_time in specified timezone
|
|
"""
|
|
if from_dt is None:
|
|
current = get_current_datetime_in_tz(tz)
|
|
else:
|
|
current = convert_from_utc(from_dt, tz)
|
|
|
|
next_day = current.date()
|
|
next_datetime = combine_date_time_in_tz(next_day, target_time, tz)
|
|
|
|
if current.time() < target_time:
|
|
next_datetime = combine_date_time_in_tz(current.date(), target_time, tz)
|
|
|
|
while next_datetime.weekday() >= 5:
|
|
next_day = next_datetime.date() + timedelta(days=1)
|
|
next_datetime = combine_date_time_in_tz(next_day, target_time, tz)
|
|
|
|
return next_datetime
|
|
|
|
|
|
def get_business_day_range(
|
|
start: datetime,
|
|
end: datetime,
|
|
tz: str = DEFAULT_TIMEZONE
|
|
) -> list[datetime]:
|
|
"""
|
|
Get list of business days (weekdays) between start and end dates.
|
|
|
|
Args:
|
|
start: Start datetime
|
|
end: End datetime
|
|
tz: IANA timezone string
|
|
|
|
Returns:
|
|
List of datetime objects for each business day
|
|
"""
|
|
if start.tzinfo is None:
|
|
start = start.replace(tzinfo=ZoneInfo(tz))
|
|
if end.tzinfo is None:
|
|
end = end.replace(tzinfo=ZoneInfo(tz))
|
|
|
|
business_days = []
|
|
current = start
|
|
|
|
while current <= end:
|
|
if current.weekday() < 5:
|
|
business_days.append(current)
|
|
current += timedelta(days=1)
|
|
|
|
return business_days
|
|
|
|
|
|
def get_tenant_now(tenant_tz: str = DEFAULT_TIMEZONE) -> datetime:
|
|
"""
|
|
Get current datetime for tenant's timezone.
|
|
|
|
Args:
|
|
tenant_tz: Tenant's IANA timezone string
|
|
|
|
Returns:
|
|
Current datetime in tenant's timezone
|
|
"""
|
|
return get_current_datetime_in_tz(tenant_tz)
|
|
|
|
|
|
def is_tenant_business_hours(
|
|
tenant_tz: str = DEFAULT_TIMEZONE,
|
|
dt: Optional[datetime] = None,
|
|
config: Optional[dict] = None
|
|
) -> bool:
|
|
"""
|
|
Check if it's currently business hours for tenant.
|
|
|
|
Args:
|
|
tenant_tz: Tenant's IANA timezone string
|
|
dt: Optional datetime to check (defaults to now)
|
|
config: Optional business hours configuration override
|
|
|
|
Returns:
|
|
True if within tenant's business hours, False otherwise
|
|
"""
|
|
if config is None:
|
|
config = BUSINESS_HOURS_DEFAULT
|
|
|
|
return is_business_hours(
|
|
dt=dt,
|
|
tz=tenant_tz,
|
|
start_hour=config.get("start_hour", BUSINESS_HOURS_DEFAULT["start_hour"]),
|
|
end_hour=config.get("end_hour", BUSINESS_HOURS_DEFAULT["end_hour"]),
|
|
include_weekends=config.get("include_weekends", BUSINESS_HOURS_DEFAULT["include_weekends"])
|
|
)
|