Why Gemfury? Push, build, and install  RubyGems npm packages Python packages Maven artifacts PHP packages Go Modules Debian packages RPM packages NuGet packages

Repository URL to install this package:

Details    
lib-py-b2b / order_dates.py
Size: Mime:
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