123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602 |
- import datetime
- import ioex
- import ioex.calcex
- import ioex.reex
- import locale
- import pytz
- import re
- import yaml
- class _Object(object):
- def __eq__(self, other):
- return type(self) == type(other) and vars(self) == vars(other)
- def __ne__(self, other):
- return not (self == other)
- def __repr__(self):
- return self.__class__.__name__ + '(%s)' % ', '.join([
- '%s=%r' % (k, v) for k, v in vars(self).items()
- ])
- class _YamlInitConstructor(yaml.YAMLObject):
- @classmethod
- def from_yaml(cls, loader, node):
- return cls(**loader.construct_mapping(node, deep=True))
- # return cls(**{
- # k: unicode(v) if isinstance(v, str) else v
- # for (k, v) in loader.construct_mapping(node, deep = True).items()
- # })
- @classmethod
- def register_yaml_constructor(cls, loader, tag=None):
- loader.add_constructor(
- cls.yaml_tag if tag is None else tag,
- cls.from_yaml,
- )
- class _YamlVarsRepresenter(yaml.YAMLObject):
- @classmethod
- def to_yaml(cls, dumper, obj):
- return dumper.represent_mapping(
- cls.yaml_tag,
- {k: v for k, v in vars(obj).items()
- if v and (not isinstance(v, list) or len(v) > 0)},
- )
- class Sum(ioex.calcex.Figure):
- currency_symbol_map = {
- '€': 'EUR',
- 'US$': 'USD',
- '¥': 'CNY',
- }
- yaml_tag = u'!sum'
- def __init__(self, value=None, currency=None, unit=None):
- if not currency is None and not unit is None:
- raise ValueError('missing currency')
- else:
- unit = currency if currency else unit
- super(Sum, self).__init__(
- value=value,
- unit=currency if currency else unit,
- )
- def get_value(self):
- return super(Sum, self).get_value()
- def set_value(self, value):
- assert type(value) is float
- super(Sum, self).set_value(value)
- """ use property() instead of decorator to enable overriding """
- value = property(get_value, set_value)
- def get_unit(self):
- return super(Sum, self).get_unit()
- def set_unit(self, currency):
- currency = Sum.currency_symbol_map.get(currency, currency)
- assert type(currency) is str
- super(Sum, self).set_unit(currency)
- """ use property() instead of decorator to enable overriding """
- unit = property(get_unit, set_unit)
- currency = property(get_unit, set_unit)
- space_regex = '[\xa0 ]'
- value_regex = r"-?\d+([\.,]\d+)?"
- currency_regex = r"[^\d\s-]+"
- sum_regex_value_first = r'\$?(?P<value>{}){}(?P<currency>{})'.format(
- value_regex, space_regex, currency_regex,
- )
- sum_regex_currency_first = r'(?P<currency>{}){}?(?P<value>{})'.format(
- currency_regex, space_regex, value_regex,
- )
- sum_regex = r'({})'.format('|'.join([
- ioex.reex.rename_groups(
- sum_regex_value_first,
- lambda n: {'value': 'pre_value', 'currency': 'post_currency'}[n]
- ),
- ioex.reex.rename_groups(
- sum_regex_currency_first,
- lambda n: {'currency': 'pre_currency', 'value': 'post_value'}[n]
- ),
- ]))
- @staticmethod
- def parse_text(text):
- match = re.search('^{}$'.format(Sum.sum_regex), text, re.UNICODE)
- assert not match is None, '\n' + '\n'.join([
- 'regex: {}'.format(Sum.sum_regex),
- 'text: {}'.format(text),
- 'text repr: {!r}'.format(text),
- ])
- attr = ioex.dict_collapse(
- match.groupdict(),
- lambda k: k.split('_')[-1],
- )
- return Sum(
- value=locale.atof(attr['value']),
- currency=attr['currency'],
- )
- class _ItemCollection():
- def __init__(self, items=None):
- if items is not None:
- assert isinstance(items, list)
- assert all([isinstance(i, Item) for i in items])
- self.items = items
- else:
- self.items = []
- class Invoice(_Object, _ItemCollection, _YamlInitConstructor, _YamlVarsRepresenter):
- yaml_tag = u'!invoice'
- def __init__(self,
- creditor,
- debitor_id,
- invoice_date,
- invoice_id,
- discounts=None,
- invoice_url=None,
- items=None
- ):
- assert isinstance(creditor, str)
- self.creditor = creditor
- assert isinstance(invoice_id, str)
- self.invoice_id = invoice_id
- assert (isinstance(invoice_date, datetime.date)
- or isinstance(invoice_date, datetime.datetime))
- self.invoice_date = invoice_date
- assert isinstance(debitor_id, str)
- self.debitor_id = debitor_id
- if discounts:
- assert isinstance(discounts, list)
- assert all([isinstance(d, Discount) for d in discounts])
- self.discounts = discounts
- else:
- self.discounts = []
- if invoice_url:
- assert isinstance(invoice_url, str)
- self.invoice_url = invoice_url
- _ItemCollection.__init__(self, items)
- class Order(_Object, _ItemCollection, _YamlInitConstructor, _YamlVarsRepresenter):
- yaml_tag = u'!order'
- def __init__(self, platform, order_id, order_date,
- customer_id=None,
- discounts=None,
- items=None,
- platform_view_url=None,
- ):
- assert type(platform) is str
- self.platform = platform
- if type(order_id) in [int]:
- order_id = str(order_id)
- assert type(order_id) is str
- self.order_id = order_id
- assert type(order_date) in [datetime.date, datetime.datetime]
- if type(order_date) is datetime.datetime and order_date.tzinfo:
- order_date = order_date.astimezone(pytz.utc)
- self.order_date = order_date
- if customer_id is not None:
- assert type(customer_id) is str
- self.customer_id = customer_id
- if discounts is None:
- self.discounts = []
- else:
- assert type(discounts) is list
- assert all([isinstance(d, Discount) for d in discounts])
- self.discounts = discounts
- if platform_view_url:
- assert isinstance(platform_view_url, str)
- self.platform_view_url = platform_view_url
- _ItemCollection.__init__(self, items)
- class Distance(ioex.calcex.Figure):
- yaml_tag = u'!distance'
- def get_value(self):
- return super(Distance, self).get_value()
- def set_value(self, value):
- assert type(value) is float
- super(Distance, self).set_value(value)
- """ use property() instead of decorator to enable overriding """
- value = property(get_value, set_value)
- @property
- def metres(self):
- if self.unit == 'm':
- return Distance(self.value, 'm')
- elif self.unit == 'km':
- return Distance(self.value * 1000, 'm')
- else:
- raise Exception()
- class Discount(_Object, _YamlInitConstructor):
- yaml_tag = u'!discount'
- def __init__(self, name=None, amount=None, code=None):
- assert type(name) is str
- self.name = name
- assert type(amount) is Sum
- assert amount.value >= 0
- self.amount = amount
- if code:
- assert isinstance(code, str)
- self.code = code
- class Item(_Object, _YamlInitConstructor, _YamlVarsRepresenter):
- yaml_tag = u'!item'
- def __init__(self,
- name=None,
- price_brutto=None,
- sub_items=None,
- ):
- if not name is None:
- assert type(name) is str
- self.name = name
- assert type(price_brutto) is Sum
- if sub_items is None:
- self.sub_items = []
- else:
- assert isinstance(sub_items, list)
- assert all([isinstance(i, Item) for i in sub_items])
- self.sub_items = sub_items
- self.price_brutto = price_brutto
- @property
- def total_price_brutto(self):
- return self.price_brutto \
- + sum([s.price_brutto for s in self.sub_items])
- class Campaign(_Object, _YamlInitConstructor):
- yaml_tag = u'!campaign'
- def __init__(self,
- end=None,
- founder=None,
- name=None,
- website_url=None,
- ):
- assert type(name) is str
- self.name = name
- assert type(founder) is str
- self.founder = founder
- if not end is None:
- assert type(end) is datetime.datetime
- self.end = end
- if not website_url is None:
- assert type(website_url) is str
- self.website_url = website_url
- class Pledge(Item):
- yaml_tag = u'!pledge'
- def __init__(self,
- campaign=None,
- reward=None,
- **kwargs
- ):
- super(Pledge, self).__init__(**kwargs)
- assert type(campaign) is Campaign
- self.campaign = campaign
- if not reward is None:
- assert type(reward) is str
- self.reward = reward
- class Contribution(Item):
- yaml_tag = u'!contribution'
- def __init__(self,
- campaign=None,
- reward=None,
- **kwargs
- ):
- super(Contribution, self).__init__(**kwargs)
- assert type(campaign) is Campaign
- self.campaign = campaign
- if not reward is None:
- assert type(reward) is str
- self.reward = reward
- class Article(Item):
- yaml_tag = u'!article'
- def __init__(self,
- access_option=None,
- authors=None,
- color=None,
- delivery_date=None,
- depth=None,
- features=None,
- height=None,
- maximum_load=None,
- option=None,
- product_id=None,
- quantity=None,
- release_date=None,
- reseller=None,
- shipper=None,
- size=None,
- state=None,
- width=None,
- **kwargs
- ):
- super(Article, self).__init__(**kwargs)
- assert not self.name is None
- if quantity is not None:
- assert type(quantity) is int
- self.quantity = quantity
- if access_option is not None:
- assert type(access_option) is str
- self.access_option = access_option
- if authors is not None:
- assert type(authors) is list
- self.authors = authors
- if state is not None:
- assert type(state) is str
- self.state = state
- if release_date is not None:
- assert type(release_date) in [datetime.datetime, datetime.date]
- self.release_date = release_date
- if reseller is not None:
- assert type(reseller) is str
- self.reseller = reseller
- if shipper is not None:
- assert type(shipper) is str
- self.shipper = shipper
- if product_id is not None:
- if type(product_id) in [int]:
- product_id = str(product_id)
- assert type(product_id) is str
- self.product_id = product_id
- if option is not None:
- assert type(option) is str
- self.option = option
- if color is not None:
- assert type(color) is str
- self.color = color
- if size is not None:
- assert type(size) is str
- self.size = size
- if width is not None:
- assert type(width) is Distance
- self.width = width
- if depth is not None:
- assert type(depth) is Distance
- self.depth = depth
- if height is not None:
- assert type(height) is Distance
- self.height = height
- if maximum_load is not None:
- assert type(maximum_load) is ioex.calcex.Figure, type(maximum_load)
- self.maximum_load = maximum_load
- if features is not None:
- assert type(features) is str
- self.features = features
- if delivery_date is not None:
- assert type(delivery_date) is datetime.date
- self.delivery_date = delivery_date
- class Service(Item):
- yaml_tag = u'!service'
- def __init__(self,
- duration=None,
- ip_addresses=None,
- location=None,
- period=None,
- state=None,
- **kwargs
- ):
- super(Service, self).__init__(**kwargs)
- assert not (duration and period)
- if duration:
- assert isinstance(duration, ioex.datetimeex.Duration)
- self.duration = duration
- if ip_addresses:
- assert isinstance(ip_addresses, list)
- assert all([isinstance(a, str) for a in ip_addresses])
- self.ip_addresses = ip_addresses
- if location:
- assert isinstance(location, str)
- self.location = location
- if period:
- assert isinstance(period, ioex.datetimeex.Period)
- self.period = period
- if state:
- assert isinstance(state, str)
- self.state = state
- class HostingService(Service):
- yaml_tag = u'!hosting-service'
- def __init__(self,
- operating_system=None,
- **kwargs
- ):
- super(HostingService, self).__init__(**kwargs)
- if operating_system:
- assert isinstance(operating_system, str)
- self.operating_system = operating_system
- class CloudMining(Service):
- yaml_tag = u'!cloud-mining'
- def __init__(self,
- hashrate=None,
- **kwargs
- ):
- super(CloudMining, self).__init__(**kwargs)
- if hashrate:
- assert isinstance(hashrate, ioex.calcex.Figure)
- self.hashrate = hashrate
- class Transportation(Item):
- yaml_tag = u'!transportation'
- def __init__(self,
- arrival_time=None,
- departure_point=None,
- destination_point=None,
- distance=None,
- estimated_arrival_time=None,
- passenger=None,
- route_map=None,
- ticket_url=None,
- valid_from=None,
- valid_until=None,
- **kwargs
- ):
- super(Transportation, self).__init__(**kwargs)
- if arrival_time is not None:
- assert isinstance(arrival_time, datetime.datetime) \
- or isinstance(arrival_time, ioex.datetimeex.Period)
- self.arrival_time = arrival_time
- if departure_point is not None:
- assert type(departure_point) is str
- self.departure_point = departure_point
- if destination_point is not None:
- assert type(destination_point) is str
- self.destination_point = destination_point
- if distance is not None:
- assert type(distance) is Distance
- self.distance = distance
- if route_map is not None:
- assert type(route_map) is bytes
- self.route_map = route_map
- if passenger is not None:
- assert type(passenger) is Person
- self.passenger = passenger
- if valid_from is not None:
- assert type(valid_from) is datetime.datetime
- assert not valid_from.tzinfo is None
- self.valid_from = valid_from
- if valid_until is not None:
- assert type(valid_until) is datetime.datetime
- assert not valid_until.tzinfo is None
- self.valid_until = valid_until
- if ticket_url is not None:
- assert type(ticket_url) is str
- self.ticket_url = ticket_url
- if estimated_arrival_time is not None:
- assert type(estimated_arrival_time) is ioex.datetimeex.Period
- assert not estimated_arrival_time.start.tzinfo is None
- assert not estimated_arrival_time.end.tzinfo is None
- self.estimated_arrival_time = estimated_arrival_time
- class Shipping(Transportation):
- yaml_tag = u'!shipping'
- def __init__(self,
- tracking_number=None,
- **kwargs
- ):
- super(Shipping, self).__init__(**kwargs)
- if tracking_number:
- assert isinstance(tracking_number, str)
- self.tracking_number = tracking_number
- class TaxiRide(Transportation):
- yaml_tag = u'!taxi-ride'
- def __init__(self,
- departure_time=None,
- driver=None,
- name=None,
- **kwargs
- ):
- if name is None:
- name = u'Taxi Ride'
- super(TaxiRide, self).__init__(name=name, **kwargs)
- assert type(driver) is str
- self.driver = driver
- assert departure_time is None or type(
- departure_time) is datetime.datetime
- self.departure_time = departure_time
- class Person(_Object, _YamlInitConstructor):
- yaml_tag = u'!person'
- def __init__(self, first_name=None, last_name=None):
- self.first_name = first_name
- self.last_name = last_name
- @property
- def first_name(self):
- return self._first_name
- @first_name.setter
- def first_name(self, first_name):
- assert first_name is None or type(first_name) is str
- self._first_name = first_name
- @property
- def last_name(self):
- return self._last_name
- @last_name.setter
- def last_name(self, last_name):
- assert last_name is None or type(last_name) is str
- self._last_name = last_name
- @classmethod
- def to_yaml(cls, dumper, person):
- return dumper.represent_mapping(cls.yaml_tag, {
- 'first_name': person.first_name,
- 'last_name': person.last_name,
- })
- def __repr__(self):
- return self.__class__.__name__ + '(%s)' % ', '.join([
- '%s=%r' % (k, v) for k, v in vars(self).items()
- ])
|