Repository URL to install this package:
|
Version:
0.4.198 ▾
|
lib-py-b2b
/
order_dates.py
|
|---|
from datetime import datetime, timedelta, date
from typing import Optional
from dateutil.parser import parse
from dateutil.tz import gettz, UTC
from lib_b2b.calendar import OperatingCalendar
from lib_b2b.change import ChangeRecord
from lib_b2b.customer_service import CustomerServiceCenters
from lib_b2b.datetime import TimePeriod, Time
from lib_b2b.errors import RuleNotFound
from lib_b2b.fulfillment_center import FulfillmentCenters
from lib_b2b.service_center import ServiceCenter
class Offset:
def __init__(self, days: int = 0, hours: int = 0, minutes: int = 0, seconds: int = 0, at: Time = None):
self.at = at
self.seconds = seconds
self.minutes = minutes
self.hours = hours
self.days = days
def add_to(self, date: datetime, calendar: OperatingCalendar, tz: str = None, naive: bool = False) -> datetime:
_tz = gettz(tz) if tz else UTC
_date = date.astimezone(_tz)
_date = _date + timedelta(days=self.days, hours=self.hours, minutes=self.minutes, seconds=self.seconds)
_date = calendar.next_available_day(_date)
if self.at:
_date = _date.replace(hour=self.at.hour, minute=self.at.minute, second=self.at.second)
if naive:
_date = _date.replace(tzinfo=None)
else:
if self.at:
_tz = gettz(self.at.tz)
_date = _date.replace(tzinfo=_tz)
if tz:
_tz = gettz(tz)
_date = _date.astimezone(_tz)
elif tz:
_tz = gettz(tz)
_date = _date.astimezone(_tz)
else:
_date = _date
return _date
def __str__(self):
_str = f"Offset: (days={self.days}, h={self.hours}, m={self.minutes}, s={self.seconds})"
if self.at:
_str += f" at {self.at.isoformat()}"
return _str
class MatchCriteria:
def __init__(self, customer: str, date: datetime):
self.date = date
self.customer = customer
class Rule:
def __init__(self, customer: str, period: TimePeriod, offset: Offset,
service_center: ServiceCenter, naive: bool = False):
self.naive = naive
self.service_center = service_center
self.offset = offset
self.period = period
self.customer = customer
def matches(self, criteria: MatchCriteria):
customer_edi_id = criteria.customer
if customer_edi_id != self.customer:
return False
_date = criteria.date
if isinstance(criteria.date, str):
_date = parse(criteria.date)
return _date in self.period
def at(self, date: datetime, offset: Offset = None, tz: str = None) -> datetime:
_offset = offset or self.offset
_operating_calendar = self.service_center.operating_calendar(self.customer)
_dt = _offset.add_to(date, _operating_calendar, tz=self.service_center.time_zone, naive=self.naive)
if tz:
_tz = gettz(tz)
return _dt.astimezone(_tz)
return _dt
def __str__(self):
return f"{self.customer} : {str(self.period)} : {str(self.offset)}"
def describe(self, tz: str = None):
return f"{self.customer} : {self.period.describe(tz)} : {str(self.offset)}"
class DisplayRule(Rule):
def __init__(self, customer: str, period: TimePeriod, offset: Offset):
_sc = CustomerServiceCenters.for_(customer)
super().__init__(customer=customer, period=period, offset=offset, service_center=_sc, naive=True)
class FulfillmentRule(Rule):
def __init__(self, customer: str, period: TimePeriod, offset: Offset):
_sc = FulfillmentCenters.for_(customer)
super().__init__(customer=customer, period=period, offset=offset, service_center=_sc, naive=False)
class RuleSet:
def __init__(self, rules):
self.rules = rules or []
def append(self, rule: Rule):
self.rules.append(rule)
def find(self, customer, date: datetime, ):
matches = list(filter(lambda r: r.matches(MatchCriteria(customer=customer, date=date)), self.rules))
if len(matches) > 0:
return matches[0]
else:
raise RuleNotFound(f"Unable to find date rule matching customer: {customer} and date: {date.isoformat()}")
def __str__(self):
_str = ""
for rule in self.rules:
_str += f"{str(rule)}\n"
return _str
def describe(self, tz: str = None):
_str = ""
for rule in self.rules:
_str += f"{rule.describe(tz)}\n"
return _str
class DisplayRules(RuleSet):
_rules = [
DisplayRule(
customer='default',
period=TimePeriod(start=Time(0, 0, 0, tz='US/Eastern'), end=Time(23, 59, 59, tz='US/Eastern')),
offset=Offset(days=0, at=Time(0, 0, 0, 'US/Eastern'))
),
DisplayRule(
customer='4021001305',
period=TimePeriod(start=Time(0, 0, 0, tz='US/Eastern'), end=Time(23, 59, 59, tz='US/Eastern')),
offset=Offset(days=0, at=Time(0, 0, 0, 'US/Eastern'))
),
DisplayRule(
customer='402100100561',
period=TimePeriod(start=Time(0, 0, 0, tz='US/Eastern'), end=Time(23, 59, 59, tz='US/Eastern')),
offset=Offset(days=0, at=Time(0, 0, 0, 'US/Eastern'))
),
DisplayRule(
customer='402100200186',
period=TimePeriod(start=Time(0, 0, 0, tz='US/Eastern'), end=Time(23, 59, 59, tz='US/Eastern')),
offset=Offset(days=0, at=Time(0, 0, 0, 'US/Eastern'))
),
DisplayRule(
customer='705700100991',
period=TimePeriod(start=Time(0, 0, 0, tz='US/Eastern'), end=Time(23, 59, 59, tz='US/Eastern')),
offset=Offset(days=0, at=Time(0, 0, 0, 'US/Eastern'))
),
DisplayRule(
customer='402100100554',
period=TimePeriod(start=Time(15, 0, 0, tz='US/Eastern'), end=Time(23, 59, 59, tz='US/Eastern')),
offset=Offset(days=1, at=Time(12, 0, 0, 'US/Eastern'))
),
DisplayRule(
customer='402100100554',
period=TimePeriod(start=Time(0, 0, 0, tz='US/Eastern'), end=Time(14, 59, 59, tz='US/Eastern')),
offset=Offset(days=0, at=Time(12, 0, 0, 'US/Eastern'))
),
DisplayRule(
customer='402100200002',
period=TimePeriod(start=Time(15, 0, 0, tz='US/Eastern'), end=Time(23, 59, 59, tz='US/Eastern')),
offset=Offset(days=1, at=Time(12, 0, 0, 'US/Eastern'))
),
DisplayRule(
customer='402100200002',
period=TimePeriod(start=Time(0, 0, 0, tz='US/Eastern'), end=Time(14, 59, 59, tz='US/Eastern')),
offset=Offset(days=0, at=Time(12, 0, 0, 'US/Eastern'))
)
]
def __init__(self):
super().__init__(DisplayRules._rules)
class ShippingRules(RuleSet):
_rules = [
FulfillmentRule(
customer='default',
period=TimePeriod(start=Time(0, 0, 0, tz='US/Eastern'), end=Time(11, 59, 59, tz='US/Eastern')),
offset=Offset(days=0, at=Time(15, 0, 0, 'US/Eastern'))
),
FulfillmentRule(
customer='default',
period=TimePeriod(start=Time(12, 0, 0, tz='US/Eastern'), end=Time(23, 59, 59, tz='US/Eastern')),
offset=Offset(days=1, at=Time(15, 0, 0, 'US/Eastern'))
),
FulfillmentRule(
customer='4021001305',
period=TimePeriod(start=Time(0, 0, 0, tz='US/Eastern'), end=Time(11, 59, 59, tz='US/Eastern')),
offset=Offset(days=0, at=Time(15, 0, 0, 'US/Eastern'))
),
FulfillmentRule(
customer='4021001305',
period=TimePeriod(start=Time(12, 0, 0, tz='US/Eastern'), end=Time(23, 59, 59, tz='US/Eastern')),
offset=Offset(days=1, at=Time(15, 0, 0, 'US/Eastern'))
),
FulfillmentRule(
customer='402100200186',
period=TimePeriod(start=Time(0, 0, 0, tz='US/Eastern'), end=Time(11, 59, 59, tz='US/Eastern')),
offset=Offset(days=0, at=Time(15, 0, 0, 'US/Eastern'))
),
FulfillmentRule(
customer='402100200186',
period=TimePeriod(start=Time(12, 0, 0, tz='US/Eastern'), end=Time(23, 59, 59, tz='US/Eastern')),
offset=Offset(days=1, at=Time(15, 0, 0, 'US/Eastern'))
),
FulfillmentRule(
customer='705700100991',
period=TimePeriod(start=Time(0, 0, 0, tz='UTC'), end=Time(23, 59, 59, tz='UTC')),
offset=Offset(days=0, at=Time(15, 0, 0, 'US/Eastern'))
),
FulfillmentRule(
customer='402100100554',
period=TimePeriod(start=Time(12, 0, 0, tz='US/Eastern'), end=Time(23, 59, 59, tz='US/Eastern')),
offset=Offset(days=2, at=Time(15, 0, 0, 'US/Eastern'))
),
FulfillmentRule(
customer='402100100554',
period=TimePeriod(start=Time(0, 0, 0, tz='US/Eastern'), end=Time(11, 59, 59, tz='US/Eastern')),
offset=Offset(days=1, at=Time(15, 0, 0, 'US/Eastern'))
),
FulfillmentRule(
customer='402100100561',
period=TimePeriod(start=Time(12, 0, 0, tz='US/Eastern'), end=Time(23, 59, 59, tz='US/Eastern')),
offset=Offset(days=2, at=Time(15, 0, 0, 'US/Eastern'))
),
FulfillmentRule(
customer='402100100561',
period=TimePeriod(start=Time(0, 0, 0, tz='US/Eastern'), end=Time(11, 59, 59, tz='US/Eastern')),
offset=Offset(days=1, at=Time(15, 0, 0, 'US/Eastern'))
),
FulfillmentRule(
customer='402100200002',
period=TimePeriod(start=Time(12, 0, 0, tz='US/Eastern'), end=Time(23, 59, 59, tz='US/Eastern')),
offset=Offset(days=2, at=Time(15, 0, 0, 'US/Eastern'))
),
FulfillmentRule(
customer='402100200002',
period=TimePeriod(start=Time(0, 0, 0, tz='US/Eastern'), end=Time(11, 59, 59, tz='US/Eastern')),
offset=Offset(days=1, at=Time(15, 0, 0, 'US/Eastern'))
)
]
def __init__(self):
super().__init__(ShippingRules._rules)
def print_rules(tz: str = None):
_tz = tz or 'US/Eastern'
_tzinfo = gettz(_tz)
_now = datetime.now(tz=_tzinfo)
_tzname = _now.tzname()
print(f"Display Rules (timezone: {_tzname})")
print("=" * 40)
print(DisplayRules().describe(_tz))
print(f"Fulfillment Rules (timezone: {_tzname})")
print("=" * 40)
print(ShippingRules().describe(_tz))
print(f"Fulfillment Centers")
print("=" * 40)
for fc in FulfillmentCenters.all():
print(fc.describe())
print(f"Fulfillment Center Assignments")
print("-" * 40)
from lib_b2b.customer import Customer
customers = Customer.fetch_all()
for customer in customers:
_fc = FulfillmentCenters.for_(customer=customer['customer_edi_id'])
print(f"{customer['customer_edi_id']} : {_fc.name}")
print(_fc.operating_calendar(customer['customer_edi_id']).describe())
print(f"Customer Service Centers")
print("=" * 40)
for cs in CustomerServiceCenters.all():
print(cs.describe())
print(f"Customer Service Center Assignments")
print("-" * 40)
customers = Customer.fetch_all()
for customer in customers:
_cs = CustomerServiceCenters.for_(customer=customer['customer_edi_id'])
print(f"{customer['customer_edi_id']} : {_cs.name}")
class CalculatedDate:
def __init__(self, date: datetime, rule: Rule, based_on: str):
self.date = date
self.based_on = based_on
self.rule = rule
def isoformat(self, sep='T', timespec='auto') -> str:
return self.date.isoformat(sep=sep, timespec=timespec)
def describe(self):
return f"date: {self.date.isoformat()}, based on: {self.based_on}, using rule: {str(self.rule)}"
class OrderDatesDelegate:
@classmethod
def assign_dates(cls, order):
delegate = cls(customer=order.customer_edi_id,
order_date=order.order_date,
not_before_date=order.not_before_date)
calculated_release_date = delegate.release_date
calculated_ship_date = delegate.ship_date
_before = order.release_date
order.release_date = calculated_release_date.date
order.record(
ChangeRecord(
before=_before.isoformat(),
after=calculated_release_date.date.isoformat(),
description=calculated_release_date.describe(),
when=datetime.now().astimezone(UTC)
)
)
_before = order.ship_date
order.ship_date = calculated_ship_date.date
order.record(
ChangeRecord(
before=_before.isoformat(),
after=calculated_ship_date.date.isoformat(),
description=calculated_ship_date.describe(),
when=datetime.now().astimezone(UTC)
)
)
def __init__(self, customer, order_date: datetime, not_before_date: Optional[datetime] = None):
self.customer = customer
self.order_date = order_date
if not_before_date and not_before_date.tzinfo is None:
not_before_date = not_before_date.replace(tzinfo=self.order_date.tzinfo)
self.not_before_date = not_before_date
@property
def release_date(self) -> CalculatedDate:
"""
Calculates the day and time that an order will become visible to customer service.
The requirement is that the order become visible at the same clock time regardless of timezone of daylight
savings time. So, we returns a timezone naive time date.
:return: A timezone naive date time
:rtype: date
"""
_rule: Rule = DisplayRules().find(self.customer, self.order_date)
if not _rule:
_rule = DisplayRules().find('default', self.order_date)
_date = _rule.at(self.order_date) if _rule else None
return CalculatedDate(date=_date, rule=_rule, based_on='order_date')
@property
def ship_date(self) -> CalculatedDate:
"""
This returns a date time object which is calculated in UTC and returned in UTC.
:return: The date that an order should be shipped.
:rtype: datetime
"""
order_date_rule = ShippingRules().find(self.customer, self.order_date)
if not order_date_rule:
order_date_rule = ShippingRules().find('default', self.order_date)
order_date_date = order_date_rule.at(self.order_date, tz='UTC') if order_date_rule else None
calculated_date = CalculatedDate(rule=order_date_rule,
date=order_date_date,
based_on=f'order_date')
if self.not_before_date and self.not_before_date > calculated_date.date:
# need to make sure that not_before_date is a manufacturing day.
# however, i need to bypass the delay rules
not_before_date_rule = ShippingRules().find(self.customer, self.not_before_date)
if not not_before_date_rule:
not_before_date_rule = ShippingRules().find('default', self.not_before_date)
not_before_date_date = not_before_date_rule.at(self.not_before_date, offset=Offset(days=0), tz='UTC') \
if not_before_date_rule else None
calculated_from_not_before_date = CalculatedDate(rule=not_before_date_rule,
date=not_before_date_date,
based_on=f'not_before_date')
return calculated_from_not_before_date
else:
return calculated_date