#!/usr/bin/env python3 # A little tool to make REST requests to the plentymarkets API import argparse import collections import copy import csv import hashlib import itertools import logging import re import time import urllib.parse from collections import Counter, defaultdict, namedtuple from datetime import datetime, timedelta from itertools import chain from logging import debug, info, warning, error from requests_oauthlib import OAuth2Session, requests from oauthlib.oauth2 import LegacyApplicationClient COUNTRYIDS = { 'DE': 1, 'AT': 2, 'BE': 3, 'CH': 4, 'CY': 5, 'CZ': 6, 'DK': 7, 'ES': 8, 'EE': 9, 'FR': 10, 'FI': 11, 'GB': 12, 'GR': 13, 'HU': 14, 'IT': 15, 'IE': 16, 'LU': 17, 'LV': 18, 'MT': 19, 'NO': 20, 'NL': 21, 'PT': 22, 'PL': 23, 'SE': 24, 'SG': 25, 'SK': 26, 'SI': 27, 'US': 28, 'AU': 29, 'CA': 30, 'CN': 31, 'JP': 32, 'LT': 33, 'LI': 34, 'MC': 35, 'MX': 36, 'IC': 37, 'IN': 38, 'BR': 39, 'RU': 40, 'RO': 41, 'EA': 42, 'EA': 43, 'BG': 44, 'XZ': 45, 'KG': 46, 'KZ': 47, 'BY': 48, 'UZ': 49, 'MA': 50, 'AM': 51, 'AL': 52, 'EG': 53, 'HR': 54, 'MV': 55, 'MY': 56, 'HK': 57, 'YE': 58, 'IL': 59, 'TW': 60, 'GP': 61, 'TH': 62, 'TR': 63, 'NZ': 66, 'AF': 67, 'AX': 68, 'DZ': 69, 'AS': 70, 'AD': 71, 'AO': 72, 'AI': 73, 'AQ': 74, 'AG': 75, 'AR': 76, 'AW': 77, 'AZ': 78, 'BS': 79, 'BH': 80, 'BD': 81, 'BB': 82, 'BZ': 83, 'BJ': 84, 'BM': 85, 'BT': 86, 'BO': 87, 'BA': 88, 'BW': 89, 'BV': 90, 'IO': 91, 'BN': 92, 'BF': 93, 'BI': 94, 'KH': 95, 'CM': 96, 'CV': 97, 'KY': 98, 'CF': 99, 'TD': 100, 'CL': 101, 'CX': 102, 'CC': 103, 'CO': 104, 'KM': 105, 'CG': 106, 'CD': 107, 'CK': 108, 'CR': 109, 'CI': 110, 'CU': 112, 'DJ': 113, 'DM': 114, 'DO': 115, 'EC': 116, 'SV': 117, 'GQ': 118, 'ER': 119, 'ET': 120, 'FK': 121, 'FO': 122, 'FJ': 123, 'GF': 124, 'PF': 125, 'TF': 126, 'GA': 127, 'GM': 128, 'GE': 129, 'GH': 130, 'GI': 131, 'GL': 132, 'GD': 133, 'GU': 134, 'GT': 135, 'GG': 136, 'GN': 137, 'GW': 138, 'GY': 139, 'HT': 140, 'HM': 141, 'VA': 142, 'HN': 143, 'IS': 144, 'ID': 145, 'IR': 146, 'IQ': 147, 'IM': 148, 'JM': 149, 'JE': 150, 'JO': 151, 'KE': 152, 'KI': 153, 'KP': 154, 'KR': 155, 'KW': 156, 'LA': 158, 'LB': 159, 'LS': 160, 'LR': 161, 'LY': 162, 'MO': 163, 'MK': 164, 'MG': 165, 'MW': 166, 'ML': 168, 'MH': 169, 'MQ': 170, 'MR': 171, 'MU': 172, 'YT': 173, 'FM': 174, 'MD': 175, 'MN': 176, 'ME': 177, 'MS': 178, 'MZ': 179, 'MM': 180, 'NA': 181, 'NR': 182, 'NP': 183, 'AN': 184, 'NC': 185, 'NI': 186, 'NE': 187, 'NG': 188, 'NU': 189, 'NF': 190, 'MP': 191, 'OM': 192, 'PK': 193, 'PW': 194, 'PS': 195, 'PA': 196, 'PG': 197, 'PY': 198, 'PE': 199, 'PH': 200, 'PN': 201, 'PR': 202, 'QA': 203, 'RE': 204, 'RW': 205, 'SH': 206, 'KN': 207, 'LC': 208, 'PM': 209, 'VC': 210, 'WS': 211, 'SM': 212, 'ST': 213, 'SA': 214, 'SN': 215, 'RS': 216, 'SC': 217, 'SL': 218, 'SB': 219, 'SO': 220, 'ZA': 221, 'GS': 222, 'LK': 223, 'SD': 224, 'SR': 225, 'SJ': 226, 'SZ': 227, 'SY': 228, 'TJ': 229, 'TZ': 230, 'TL': 231, 'TG': 232, 'TK': 233, 'TO': 234, 'TT': 235, 'TN': 236, 'TM': 237, 'TC': 238, 'TV': 239, 'UG': 240, 'UA': 241, 'UM': 242, 'UY': 243, 'VU': 244, 'VE': 245, 'VN': 246, 'VG': 247, 'VI': 248, 'WF': 249, 'EH': 250, 'ZM': 252, 'ZW': 253, 'AE': 254, 'CUW': 258, 'SXM': 259, 'BES': 260, 'BL': 261, 'XI': 265, ' — ': 0 } COUNTRYISO = dict((v, k) for k, v in COUNTRYIDS.items()) EUISONAMESDE = { "AT": "Österreich", "BE": "Belgien", "BG": "Bulgarien", "CY": "Zypern", "CZ": "Tschechien", "DE": "Deutschland", "DK": "Dänemark", "ES": "Spanien", "EE": "Estland", "FI": "Finnland", "FR": "Frankreich", "GR": "Griechenland", "HR": "Kroatien", "HU": "Ungarn", "IE": "Irland", "IT": "Italien", "LT": "Litauen", "LU": "Luxemburg", "LV": "Lettland", "MT": "Malta", "NL": "Niederlande", "PL": "Polen", "PT": "Portugal", "RO": "Rumänien", "SE": "Schweden", "SI": "Slowenien", "SK": "Slowakei", "CH": "Schweiz" } SHIPPINGPROFILES = defaultdict(str, { 6: 'Versichertes Paket', 10: 'Abholung', 11: 'GLS', 15: 'DHL Express', 19: 'DHL Paket', 33: 'Deutsche Post Kompaktbrief', 34: 'DHL', 35: 'PARCEL.ONE Eco', 37: 'PARCEL.ONE Basic', 38: 'PARCEL.ONE Plus', 40: 'PARCEL.ONE Letter Eco', 41: 'PARCEL.ONE Letter Basic', 42: 'PARCEL:ONE Letter Basic Einschreiben', 43: 'Deutsche Post Großbrief', 48: 'DHL mit Emailübertragung', 51: 'DHL Paket FSK18', 54: 'DHL Sperrgut', 56: 'Spedition', 57: 'DHL Paket', 59: 'Warenpost 5cm max', 60: 'Warenpost 3cm max', 62: 'Deutsche Post Standardbrief', 65: 'DHL Paket', 67: 'DHL', 68: 'DHL Kleinpaket', 69: 'DHL Ablago', 71: 'DHL Bicycle', 72: 'DHL Bicycle Kleinpaket', 73: 'DHL Shopify gomazing', 74: 'DHL Paket', 75: 'DHL Kleinpaket', 76: 'DHL Prime', 77: 'DHL Pummeleinhorn Paket', 78: 'ParcelOne Eco Pummeleinhorn', 79: 'DHL Pummeleinhorn Kleinpaket', 80: 'ParcelOne Letter Eco Pummeleinhorn', 81: 'DHL Pummeleinhorn Warenpost 3cm', 82: 'DHL 3,90', 83: 'DHL EinhornLiebe', 84: 'DHL EinhornLiebe Kleinpaket', 85: 'DHL an Amazon', 86: 'DHL Kleinpaket' }) # profile specifications hardcoded, should match plenty config # minweight, maxweight - limits inclusive in g # clients - tuple of plenty ID numbers # origins - tuple of origin IDs # countries - tuple of country ISO codes SP = namedtuple('ShippingProfile', ['profileid', 'minweight', 'maxweight', 'clients', 'origins', 'countries']) # ordered by priority SHIPPINGPROFILEMATRIX = [ SP(86, 0, 1000, (13018, 17603, 19997, 52772, 58589), (), ('DE',)), SP(84, 0, 1000, (67020,), (), ('DE',)), # einhornliebe SP(79, 0, 1000, (15868,), (), ('DE',)), # pummeleinhorn SP(75, 0, 1000, (54768,), (), ('DE',)), # poolpirat SP(72, 0, 1000, (58589,), (), ('DE',)), # bicyclecards SP(6, 0, 31500, (13018, 17603), (), ()), # TODO add more profiles ] class StaticShippingCalculator(): def __init__(self): pass def get_shipping_profile_priority(self, profileid, weight, client, origin, country): """get best shipping profile id""" prio = 1 for sp in SHIPPINGPROFILEMATRIX: if ((profileid == sp.profileid) and (sp.minweight <= weight <= sp.maxweight) and (not sp.clients or client in sp.clients) and (not sp.origins or origin in sp.origins) and (not sp.countries or country in sp.countries)): return prio prio += 1 return None ShippingCalculator = StaticShippingCalculator() class PlentyObject(dict): """Base for returned json objects""" def update_from_json(self, json): for k, v in json.items(): if k in self.conversionmap: self[k] = self.conversionmap[k](self, v) else: self[k] = v def w3c_to_datetime(self, w3c): try: return datetime.strptime(w3c, "%Y-%m-%dT%H:%M:%S%z") except ValueError: return datetime.strptime(w3c, "%Y-%m-%d %H:%M:%S") conversionmap = { 'createdAtFrom': w3c_to_datetime, 'createdAtTo': w3c_to_datetime, 'createdAt': w3c_to_datetime, 'deliveredAt': w3c_to_datetime, 'updatedAtFrom': w3c_to_datetime, 'updatedAtTo': w3c_to_datetime, 'updatedAt': w3c_to_datetime } class PlentyAddress(PlentyObject): DEFAULTS = { 'name1': '', # company 'name2': '', # forename 'name3': '', # surname 'address1': '', # street 'address2': '', # housenumber 'address3': '', # additional 'postalCode': '', 'town': '', 'countryId': 1, # plenty country code 'options': [] } def __init__(self, country=None, defaults={}): """defaults: contains a json like dict to initialize this order with""" if not defaults: defaults = self.DEFAULTS c = copy.deepcopy(defaults) self.update(c) if country: self['countryId'] = COUNTRYIDS[country.upper()] def get_option_index(self, typeid): opttypes = dict((p[1]['typeId'], p[0]) for p in enumerate(self['options'])) if typeid in opttypes: return opttypes[typeid] else: return None def get_option_by_id(self, typeid): propindex = self.get_option_index(typeid) if propindex is not None: return self['options'][propindex]['value'] return None def set_or_append_option(self, typeid, value): if 'options' not in self: self['options'] = [] propindex = self.get_option_index(typeid) if propindex is not None: self['options'][propindex]['value'] = str(value) else: d = {'typeId': typeid, 'value': str(value)} self['options'].append(d) def get_email_address(self): return self.get_option_by_id(5) def set_email_address(self, email): if email: self.set_or_append_option(5, email) def correct(self): """Try to correct an address if it seems to be invalid""" if self['countryId'] != 1: # do not try to correct international addresses return self if not self['address2']: # seems no housenumber present line1 = self['address1'].strip() # split into streetname and housenumber m = re.split(r'([0-9]+\s?[a-z]?)', line1, flags=re.I) if len(m) == 3 and not m[2]: addr = copy.deepcopy(self) addr['address1'] = m[0].strip() addr['address2'] = m[1] return addr return self def stripplenty(self): """remove all keys indicating an already exisiting database item""" for k in ('id', 'createdAt', 'pivot', 'updatedAt', 'readOnly', 'options'): if k in self: del self[k] def json(self): return self def export_to_plenty(self, plenty): r = plenty.post_address(self) return r.json() def get_country_iso(self): return COUNTRYISO[self['countryId']] class PlentyOrder(PlentyObject): DEFAULTS = {} def __init__(self, defaults={}, originalwithparam=""): """defaults: contains a json like dict to initialize this order with""" if not defaults: defaults = self.DEFAULTS c = copy.deepcopy(defaults) self.update(c) self.billingaddress = None self.deliveryaddress = None self.parent = None self.children = [] self.originalwithparam = originalwithparam def get_property_index(self, typeid): proptypes = dict((p[1]['typeId'], p[0]) for p in enumerate(self['properties'])) if typeid in proptypes: return proptypes[typeid] else: return None def get_date_index(self, typeid): datetypes = dict((p[1]['typeId'], p[0]) for p in enumerate(self['dates'])) if typeid in datetypes: return datetypes[typeid] else: return None def get_property_by_id(self, typeid): propindex = self.get_property_index(typeid) if propindex is not None: return self['properties'][propindex]['value'] return None def set_or_append_property(self, typeid, value): propindex = self.get_property_index(typeid) if propindex is not None: self['properties'][propindex]['value'] = str(value) else: d = {'typeId': typeid, 'value': str(value)} self['properties'].append(d) def set_or_append_date(self, typeid, value): dateindex = self.get_date_index(typeid) if dateindex is not None: self['dates'][dateindex]['date'] = str(value) else: d = {'typeId': typeid, 'date': str(value)} self['dates'].append(d) def get_date(self, typeid): dateindex = self.get_date_index(typeid) if dateindex is not None: return self['dates'][dateindex]['date'] return None def get_date_shipped_on(self): # 5 represents created on date return self.get_date(5) def get_date_created_on_epoch(self): # 5 represents created on date s = self.get_date(2) if s: return datetime.strptime(s, "%Y-%m-%dT%H:%M:%S%z").timestamp() def get_date_shipped_on_epoch(self): # 5 represents shipped on date s = self.get_date(5) if s: return datetime.strptime(s, "%Y-%m-%dT%H:%M:%S%z").timestamp() def set_date_created_on(self, value): # 2 represents created on date self.set_or_append_date(2, value) def set_external_order_id(self, eoid): # 7 represents external order id type self.set_or_append_property(7, eoid) def set_customer_sign(self, sign): # 8 represents customer sign type self.set_or_append_property(8, sign) def set_shipping_costs(self, value): poi = PlentyOrderItem() poi['typeId'] = 6 poi['quantity'] = 1 poi['amounts'][0]['priceOriginalGross'] = value self.add_order_item(poi) def get_external_order_id(self): eoid = self.get_property_by_id(7) if eoid is None and self.parent: return self.parent.get_external_order_id() return eoid def get_customer_sign(self): return self.get_property_by_id(8) def get_parcel_numbers(self, include_children=False): if 'shippingPackages' in self: for sp in self['shippingPackages']: yield sp['packageNumber'] if include_children: for c in self.children: for pn in c.get_parcel_numbers(): yield pn else: return [] def get_shipping_profile(self): return self.get_property_by_id(2) def set_billingaddress(self, plentyaddress): self.billingaddress = plentyaddress def set_deliveryaddress(self, plentyaddress): self.deliveryaddress = plentyaddress def set_paymentmethod(self, methodid): # 3 represents payment method self.set_or_append_property(3, methodid) def add_order_item(self, oi): self['orderItems'].append(oi) def contains_preorder_items(self): return any((oi.ispreorder for oi in self['orderItems'])) def get_total_item_quantity(self): """returns the number of all items. bundle items are counted individually""" itemcount = 0 if 'orderItems' in self: for oi in self['orderItems']: if oi['typeId'] != 2: # ignore bundle but count items inside bundle itemcount += oi['quantity'] return itemcount def get_return_reasons_codes(self): """If this order is a return give a set of all reasons present in orderItems""" try: codes = [] for oi in self['orderItems']: for p in oi['properties']: if p['typeId'] == 35: codes.append(int(p['value'])) return set(codes) except KeyError as ke: return set() def order_id_already_used(self, plenty): eoid = self.get_external_order_id() if eoid: end = datetime.now() start = end - timedelta(days=30) fp = PlentyFilterParams({'createdAtFrom': start, 'createdAtTo': end, 'orderType': 1, 'externalOrderId': eoid}) ol = plenty.get_orders(**fp.json()) if len(ol) > 0: warning(f'external order id "{eoid}" already used by order id ' f'"{ol[0]["id"]} ignoring"') return True return False def json(self): return self def export_to_plenty(self, plenty): # a litte safety check if self.order_id_already_used(plenty): return ba = self.billingaddress.export_to_plenty(plenty) baid = ba['id'] if self.deliveryaddress: sa = self.deliveryaddress.export_to_plenty(plenty) said = sa['id'] else: said = baid self['addressRelations'] = [ {'typeId': 1, 'addressId': baid}, {'typeId': 2, 'addressId': said} ] r = plenty.post_order(self) return r.json() def set_addresses_from_raw_json(self, addressrelations, addresses): addressids = dict(((a['typeId'], a['addressId']) for a in addressrelations)) addresses = dict(((a['id'], a) for a in addresses)) if 1 in addressids: ba = PlentyAddress(defaults=addresses[addressids[1]]) self.set_billingaddress(ba) if 2 in addressids: da = PlentyAddress(defaults=addresses[addressids[2]]) self.set_deliveryaddress(da) def has_parent(self): """check if order has a aprent order""" for oref in self['orderReferences']: if oref['referenceType'] == 'parent': return True else: return False def get_parent_id(self): """get parent order id""" for oref in self['orderReferences']: if oref['referenceType'] == 'parent': return int(oref['originOrderId']) def save_to_plenty(self, plenty): """Save upated order to plenty""" return plenty.put_order(self['id'], self) @property def is_paid(self): return self.get_property_by_id(4) == 'fullyPaid' @classmethod def get_by_orderid(cls, plenty, orderid, withparam='addresses'): r = plenty.get_order(orderid, withparam) po = cls(defaults=r.json(), originalwithparam=withparam) po.set_addresses_from_raw_json(po['addressRelations'], po['addresses']) po.orderitems = PlentyOrderItem.list_from_json(po['orderItems']) return po @classmethod def get_by_orderid_with_search(cls, plenty, orderid, withparam='addresses'): """searching an order like this will result in more details like the attribute "hasDeliveryOrders" """ pf = PlentyFilterParams({'orderId': orderid, 'lazyLoaded': False, 'with': withparam.split(',')}) r = plenty.search_orders_post(pf) po = cls(defaults=r[0], originalwithparam=withparam) po.set_addresses_from_raw_json(po['addressRelations'], po['addresses']) po.orderitems = PlentyOrderItem.list_from_json(po['orderItems']) return po @classmethod def get_by_complex_search(cls, plenty, jsonfilter): """searching for an order """ if 'with' in jsonfilter: withparam = ','.join(jsonfilter['with']) else: withparam = '' r = plenty.search_orders_post_lazy(jsonfilter) for entry in r: po = cls(defaults=entry, originalwithparam=withparam) if 'addressRelations' in po and 'addresses' in po: po.set_addresses_from_raw_json(po['addressRelations'], po['addresses']) if 'orderItems' in po: po.orderitems = PlentyOrderItem.list_from_json(po[ 'orderItems']) yield po @classmethod def get_by_status(cls, plenty, status): """Get orders with status as iterator PlentyOrder objects""" r = plenty.search_orders(params={'statusId': status}) for details in r.json()['entries']: yield cls.get_by_orderid(plenty, details['id']) @classmethod def get_by_externalorderid(cls, plenty, eoid, typeid=1, withparam=None): """Get order with external order id as PlentyOrder object""" r = plenty.search_orders(params={'orderProperty_7': eoid, 'orderTypeId': typeid}) entries = r.json()['entries'] if len(entries) == 1: if withparam: return cls.get_by_orderid_with_search(plenty, entries[0]['id'], withparam=withparam) else: return cls.get_by_orderid(plenty, entries[0]['id']) else: return None @classmethod def get_by_packagenumber(cls, plenty, packagenumber, startdate=None, onlyfirstmatch=True): """Get order with packagenumber as PlentyOrder object""" enddate = datetime.now().astimezone() if not startdate: startdate = enddate - timedelta(days=365) complexfilter = { "conditionType": "and", "fields": [ { "field": "createdAt", "operator": "gt", "value": startdate.replace(microsecond=0).isoformat() }, { "field": "packageNumber", "operator": "eq", "value": packagenumber } ], "groups": [], "with": [ "addresses", "shippingPackages", "shippingInformation", ], "lazyLoaded": False } pos = list(PlentyOrder.get_by_complex_search(plenty, complexfilter)) if pos and onlyfirstmatch: return pos[0] else: return pos @classmethod def get_unpaid_orderids(cls, plenty, referrers): complexfilter = { "conditionType": "and", "fields": [ { "field": "statusId", "operator": "lte", "value": 7 }, { "field": "orderTypeId", "operator": "eq", "value": 1 }, { "field": "referrerId", "operator": "in", "value": list(referrers) }, { "field": "orderProperty_4", "operator": "eq", "value": "unpaid" } ], "groups": [], "with": [], "lazyLoaded": False } unpaid = PlentyOrder.get_by_complex_search(plenty, complexfilter) return [order['id'] for order in unpaid] class PlentyOrderItem(PlentyObject): DEFAULTS = { "typeId": 1, "referrerId": 0, "amounts": [ { "isSystemCurrency": True, "currency": "EUR", "exchangeRate": 1, "priceOriginalGross": 0 } ], "properties": [], "orderProperties": [] } def __init__(self, *args, defaults={}, **kwargs): """defaults: contains a json like dict to initialize this order with""" super().__init__(*args, **kwargs) if not defaults: defaults = self.DEFAULTS if not self: c = copy.deepcopy(defaults) self.update(c) self.ispreorder = False def copy(self): c = PlentyOrderItem(self) c.ispreorder = self.ispreorder return c def json(self): return self def get_human_readable(self, namemap): """return a dict with nice human readable names for known fixed ids""" hr = self.copy() for fieldname, valuemap in namemap.items(): if fieldname in hr: hr[fieldname] = valuemap.get(hr[fieldname], hr[fieldname]) return hr def get_property_index(self, typeid): proptypes = dict((p[1]['typeId'], p[0]) for p in enumerate(self['properties'])) if typeid in proptypes: return proptypes[typeid] else: return None def get_property_by_id(self, typeid): propindex = self.get_property_index(typeid) if propindex is not None: return self['properties'][propindex]['value'] return None @classmethod def list_from_json(cls, orderitems): return [cls(oi) for oi in orderitems] class PlentyPayment(PlentyObject): defaults = {} def __init__(self, amount=None, receiveddate=None, transactionnumber=None, transactiontext=None): super().__init__() self.update(self.defaults) if amount is not None: self['amount'] = amount if transactionnumber is not None: self.set_property(1, transactionnumber) if transactiontext is not None: self.set_property(3, transactiontext) now = datetime.now().replace(microsecond=0).astimezone().isoformat() self['importedAt'] = now if receiveddate is not None: if isinstance(receiveddate, datetime): recv = receiveddate.replace(microsecond=0).astimezone()\ .isoformat() self['receivedAt'] = recv else: self['receivedAt'] = str(receiveddate) def export_to_plenty(self, plenty): try: r = plenty.post_payment(PlentyData(self)) except Exception: error(f'Export to Plenty failed for: {str(self)}') raise if r.ok: self['id'] = r.json()['id'] return r def assign_to_order(self, plenty, orderid): return plenty.post_payment_order_relation(self['id'], orderid) def set_property(self, typeid, value): newprops = {'typeId': typeid, 'value': value} if 'properties' not in self or not self['properties']: self['properties'] = [newprops] else: propdict = dict(((d['typeId'], d['value']) for d in self['properties'])) propdict[typeid] = value self['properties'] = [{'typeId': t, 'value': v} for t, v in propdict.items()] @classmethod def get_payments_from_order_id(cls, plenty, oid): data = plenty.get_payments_for_order(oid) payments = [] for payment in data: newpayment = cls() newpayment.update(payment) payments.append(newpayment) return payments class PlentyAmazonPayPayment(PlentyPayment): defaults = { 'currency': 'EUR', 'type': 'credit', 'mopId': 6002, 'transactionType': 2, 'status': 3, } class PlentyPayoneCreditCardPayment(PlentyPayment): defaults = { 'currency': 'EUR', 'type': 'credit', 'mopId': 6014, 'transactionType': 2, 'status': 3, } class PlentyPayoneSofortPayment(PlentyPayment): defaults = { 'currency': 'EUR', 'type': 'credit', 'mopId': 6011, 'transactionType': 2, 'status': 3, } class PlentyManualPayment(PlentyPayment): defaults = { 'currency': 'EUR', 'type': 'credit', 'mopId': 5000, 'transactionType': 2, 'status': 3, } class PlentyOrderItemList(list): def get_collapsed(self): """collapse bundles of one into to one position""" orderitems = PlentyOrderItemList() bundle = PlentyOrderItemList() for oi in self[::-1]: if oi['typeId'] == 3: # is component bundle.append(oi) elif oi['typeId'] == 2: # is bundle if len(bundle) == 1: # bundle of 1 newoi = oi.copy() newoi['typeId'] = 1 newoi['itemVariationId'] = bundle[0]['itemVariationId'] newoi['quantity'] = bundle[0]['quantity'] newoi['orderItemName'] = bundle[0]['orderItemName'][4:] orderitems.append(newoi) else: orderitems.extend(bundle) orderitems.append(oi) bundle = PlentyOrderItemList() else: orderitems.append(oi) orderitems.reverse() return orderitems class PlentyOrderList(list): def __init__(self, orders=[]): super().__init__() for o in orders: orderitems = [] for oi in o['orderItems']: orderitems.append(PlentyOrderItem(oi)) neworder = o.copy() neworder['orderItems'] = PlentyOrderItemList(orderitems) self.append(neworder) def get_order_items(self): """get a all order items combined""" orderitems = (o['orderItems'] for o in self) return itertools.chain.from_iterable(orderitems) def get_order_items_collapsed(self): """returns with an order list where bundles with just one position are collapsed into the main position for easier handling""" orderitems = (o['orderItems'].get_collapsed() for o in self) return itertools.chain.from_iterable(orderitems) def link_related_orders(self): """find related orders in this list and link them to each other""" for po in self: parentid = po.get_parent_id() for parent in self: if parent['id'] == parentid: parent.children.append(po) po.parent = parent break class PlentyVariationStock(): """stock locations for one variationid""" def __init__(self, variationid, warehouse): self.variationid = variationid self.warehouse = warehouse self.by_location = defaultdict(int) def sub_stock_movement(self, sm): assert (self.warehouse == sm['warehouseId'] or self.warehouse == 133 and sm['warehouseName'] == 'Halle') self.by_location[sm['storageLocationName']] -= sm['quantity'] @classmethod def get_current_for(cls, plenty, vid, warehouse=133): pvs = cls(vid, warehouse) fp = PlentyFilterParams({'variationId': vid}) r = plenty.get_stock_locations_by_warehouse(vid, warehouse) locations = [] for stock in r.json()['entries']: locationid = stock['storageLocationId'] if locationid != 0: location = plenty.get_storage_location_by_warehouse( locationid, warehouse).json() locations.append((location['name'], location['position'], stock['quantity'])) else: locations.append(('Standard-Lagerort', 999999, stock['quantity'])) # sort by position locations.sort(key=lambda x: x[1]) for name, position, quantity in locations: pvs.by_location[name] = quantity return pvs @property def first_location(self): locationnames = list(self.by_location.keys()) if locationnames: return locationnames[0] else: return '' class PlentyStockMovement(PlentyObject): @classmethod def from_json(cls, json): psm = cls() psm.update_from_json(json) return psm def __hash__(self): return self['id'] def is_incomming(self): """Returns true if this is a regular incomming stock movement""" reason = self['reasonString'] return reason in ['Wareneingang (WE)', 'Wareneingang', 'Einbuchung durch Inventur', 'Wareneingang (Purchase order)', 'Wareneingang (Logistik)', 'Wareneingang durch Einlagerung', 'Artikel eingebucht'] def is_transfer(self): """Returns true if this is a movement from one location to another""" reason = self['reasonString'] return reason in ['Warenumbuchung über plentyWarehouse', 'Warenumbuchung'] def is_correction(self): """Returns true if this is a correction""" reason = self['reasonString'] return reason in ['Einbuchung durch Korrektur', 'Warenkorrektur (interne Verschiebung)', 'Warenkorrektur (WK)', 'Warenkorrektur durch Inventur', 'Warenkorrektur wegen Beschädigung', 'Warenkorrektur wegen Defekt (ohne Rückgabe)', 'Warenkorrektur wegen Eigenbedarf', 'Warenkorrektur wegen Herstellerfehler', 'Warenkorrektur wegen Muster', 'Warenkorrektur wegen Muster an Kunden', 'Warenkorrektur wegen Schwund', 'Warenkorrektur wegen Versand an FBA', 'Warenkorrektur wegen Versand an Fulfillment', 'Warenkorrektur'] def is_real(self): """Returns true if this is a real movement that affects the physical stock""" reason = self['reasonString'] return (reason != "Artikel erfasst" and reason != "Lagerort geändert" and reason != "Menge geändert" and reason != "Menge korrigiert" and reason != "Artikel aus Wareneingang gelöscht" and reason != "Artikel aus Wareneingang entfernt") class PlentyStockMovementList(list): @classmethod def get_all(cls, plenty, warehouseid, variationid=None, untildate=None, startyear=2012): """get the complete history of stock movements filtered by variation if present""" params = PlentyFilterParams() movements = cls() if variationid: params['variationId'] = variationid if not untildate: untildate = datetime.now() for year in range(startyear, untildate.year + 1): debug(f'Getting stock movements for year {year}') params['year'] = year if year == untildate.year: params['createdAtTo'] = untildate sms = plenty.get_stock_movements(warehouseid, params) movements.extend([PlentyStockMovement.from_json(sm) for sm in sms]) # get movements of the last 3 months del params['year'] sms = plenty.get_stock_movements(warehouseid, params) movements.extend([PlentyStockMovement.from_json(sm) for sm in sms]) return filter(lambda x: x.is_real(), movements) @classmethod def get_all_for_timespan(cls, plenty, warehouseid, variationid, startdate, enddate, calculate_balance=False): """Get all stock movements in a sorted list with balance. If calculate_balance is set the balance will be added to each movement by starting with the oldest or newest stock value and adding up movments from there. Not real movments like 'Artikel erfasst' and 'Lagerort geändert' will be ignored The REST call for the stock movements is an oddball. It will get either movements older than 3 months or newer than 3 months. It gets tricky when the timespan overlaps. """ def get_archived(start=None, end=None): """Be sure to call this with dates before the last 3 months""" movements = cls() if start: startyear = start.year else: startyear = 2012 if end: endyear = end.year else: endyear = datetime.now().year for year in range(startyear, endyear + 1): params = PlentyFilterParams() if variationid: params['variationId'] = variationid params['year'] = year if start and startyear == year: params['createdAtFrom'] = start if end and endyear == year: params['createdAtTo'] = end sms = plenty.get_stock_movements(warehouseid, params) movements.extend([PlentyStockMovement.from_json(sm) for sm in sms]) return movements def get_current(start=None, end=None): """Be sure to call this with dates in the last 3 months""" movements = cls() params = PlentyFilterParams() if variationid: params['variationId'] = variationid if start: params['createdAtFrom'] = start if end: params['createdAtTo'] = end sms = plenty.get_stock_movements(warehouseid, params) movements.extend([PlentyStockMovement.from_json(sm) for sm in sms]) return movements now = datetime.now().astimezone() threemonthsago = now - timedelta(days=90) margin = timedelta(days=5) # all movements current if threemonthsago + margin < startdate: movements = get_current(startdate, enddate) # all movements archived if enddate < threemonthsago - margin: movements = get_archived(startdate, enddate) # part archived and part current if threemonthsago < enddate and threemonthsago > startdate: movements = get_current() movements.extend(get_archived(startdate)) movements = filter(lambda x: startdate < x['createdAt'] < enddate, movements) movements = filter(lambda x: x.is_real(), set(movements)) return sorted(movements, key=lambda x: x['createdAt']) class PlentyOrderStatus(PlentyObject): def __init__(self, **kwargs): self.update_from_json(kwargs) def __lt__(self, other): return self['createdAt'] < other['createdAt'] class PlentyOrderStatusHistory(list): @classmethod def get_by_orderid(cls, plenty, orderid): posh = cls() histories = plenty.get_status_history(orderId=orderid) for h in histories: pos = PlentyOrderStatus(**h) posh.append(pos) posh.sort() return posh @classmethod def get_by_timespan(cls, plenty, createdAtFrom, createdAtTo): posh = cls() histories = plenty.get_status_history(createdAtFrom=createdAtFrom, createdAtTo=createdAtTo) for h in histories: pos = PlentyOrderStatus(**h) posh.append(pos) posh.sort() return posh class PlentyOrderStatusHistoryList(list): @classmethod def get_all_with_filter(cls, plenty, statusId=None, createdAtFrom=None, createdAtTo=None, userId=None): """Gets the complete order histories for all orders with the indicated status change""" histories = plenty.get_status_history(statusId=statusId, createdAtFrom=createdAtFrom, createdAtTo=createdAtTo, userId=userId) poshl = cls() for h in histories: posh = PlentyOrderStatusHistory.get_by_orderid(plenty, h['orderId']) poshl.append(posh) return poshl class PlentySession(): """Authenticate and hold details to the current session""" clientid = 'pr.py' def __init__(self, baseurl, username, password): self.baseurl = baseurl self.username = username self.password = password def __enter__(self): self.session = OAuth2Session(client=LegacyApplicationClient( self.clientid)) self.token = self.session.fetch_token(self.baseurl + 'login', username=self.username, password=self.password, client_id=self.clientid) return self def open(self): return self.__enter__() def __exit__(self, objtype, value, traceback): r = self.session.post(self.baseurl + 'logout') return r.ok and not (objtype or value or traceback) def close(self): return __exit__(None, None, None) class PlentyData(dict): def json(self): """convert to json in place and return self for convenience""" for k, v in self.items(): if k in self.conversionmap: self.conversionmap[k](self, k) elif isinstance(v, list): # consider one level of recursion as well for subdict in v: if isinstance(subdict, dict): for ksub, vsub in subdict.items(): if ksub in self.conversionmap: self.conversionmap[ksub](subdict, ksub) return self def datetime_to_w3c(self, key): if not isinstance(self[key], str): # value is expected to be datetime dt = self[key].replace(microsecond=0).astimezone() self[key] = dt.isoformat() conversionmap = { 'createdAtFrom': datetime_to_w3c, 'createdAtTo': datetime_to_w3c, 'displayDateFrom': datetime_to_w3c, 'displayDateTo': datetime_to_w3c, 'deliveredAt': datetime_to_w3c, 'updatedAtFrom': datetime_to_w3c, 'updatedAtTo': datetime_to_w3c, 'updatedAt': datetime_to_w3c, } class PlentyImage(PlentyObject): def __init__(self, json={}): self.update_from_json(json) def is_available_for_market(self, market): for a in self['availabilities']: if (a['value'] == market and a['type'] == 'marketplace' and a['imageId'] == self.id): return True return False @property def md5checksum(self): if not self['md5Checksum']: # calculate helper md5 from a few unique details for old entries # if a new image gets uploaded it will have a real md5 fields = ['width', 'height', 'size', 'url', 'insert', 'fileType'] s = "".join((str(self[f]) for f in fields)) return hashlib.md5(s.encode('utf8')).hexdigest() else: return self['md5Checksum'] @property def url(self): return self['url'] @property def id(self): return self['id'] @property def position(self): return self['position'] class PlentyVariationProperty(PlentyObject): selectioncache = {} def __init__(self, json={}, session=None): self.session = session self.update_from_json(json) def get_selection_details(self, svalue): selection = self.selectioncache.get(svalue, None) if not selection: r = self.session.get_property_selection_names(svalue) selection = r.json()[0] self.selectioncache[svalue] = selection return selection @property def descriptions(self): d = set() for rv in self['relationValues']: selection = self.get_selection_details(rv['value']) d.add(selection['description']) return d @property def names(self): d = set() for rv in self['relationValues']: selection = self.get_selection_details(rv['value']) d.add(selection['name']) return d @property def name(self): if self.names: return self.names.pop() @property def value(self): if not self['relationValues']: return None else: return self['relationValues'][0]['value'] class PlentyMarketListing(PlentyObject): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) def get_property_by_typeid(self, typeid): for p in self['properties']: if p['typeId'] == typeid: return p['value'] @property def id(self): return self['id'] @property def itemid(self): return int(self['listing']['itemId']) @property def shippingprofile(self): return self.get_property_by_typeid(1) @classmethod def get_by_id(cls, plenty, marketlistingid): r = plenty.get_listings_markets_mlid(marketlistingid) m = cls() m.update_from_json(r.json()) return m @classmethod def get_for_item(cls, plenty, itemid, withparam='properties,listing,texts'): r = plenty.get_listings_markets(itemId=itemid, withparam=withparam) if r: m = cls() m.update_from_json(r[0]) return m @classmethod def get_with_shippingprofiles_in(cls, plenty, profileids): try: profileids = set(profileids) except TypeError: profileids = set([profileids]) listings = [] for pid in profileids: r = plenty.get_listings_markets( shippingProfileId=pid, withparam='properties,listing,texts') if r: for data in r: m = cls() m.update_from_json(data) listings.append(m) return listings class PlentyVariation(PlentyObject): def __init__(self, *args, **kwargs): self.session = None super().__init__(*args, **kwargs) def get_sales_price_by_id(self, priceid): for vsp in self['variationSalesPrices']: if vsp['salesPriceId'] == priceid: return vsp['price'] def get_market_active_by_id(self, marketid): for vm in self['variationMarkets']: if vm['marketId'] == marketid: return True return False def stock_for_warehouse(self, warehouseid): for stock in self['stock']: # we need to check for variation id too because stock of # bundles may also be included if (stock['warehouseId'] == warehouseid and stock['variationId'] == self['id']): return stock['physicalStock'] return 0 def get_property(self, propertyid): properties = list(filter(lambda x: x['propertyId'] == propertyid, self['properties'])) if properties: return PlentyVariationProperty(properties[0], self.session) def get_present_property_values(self, *propertyids): values = [] for pid in propertyids: p = self.get_property(pid) if p and p.value is not None: values.append(p.value) return values def get_images_for_market(self, market): images = [PlentyImage(i) for i in self['images']] images = filter(lambda x: x.is_available_for_market(market), images) return sorted(images, key=lambda x: x.position) @property def id(self): return self['id'] @property def salesprices(self): # ignore reseller price id 52 s = set() for vsp in self['variationSalesPrices']: if vsp['salesPriceId'] != 52: s.add(vsp['price']) return s @property def availability(self): return self['availability'] @property def dimensions(self): weightnetg = self['weightNetG'] if weightnetg is not None: weightnetg = self['weightNetG'] / 1000 return (self['lengthMM'], self['widthMM'], self['heightMM'], weightnetg) @property def ean(self): for vb in self['variationBarcodes']: if vb['barcodeId'] == 1: return vb['code'] @property def images(self): return [PlentyImage(i) for i in self['images']] @property def releasedate(self): return self['releasedAt'] @property def releasedate_datetime(self): if self.releasedate: return datetime.strptime(self.releasedate, "%Y-%m-%dT%H:%M:%S%z") @property def days_till_release(self): if self.releasedate_datetime: now = datetime.now().astimezone() return (self.releasedate_datetime - now).days @classmethod def get_all_for_item_id(cls, plenty, itemid): r = plenty.get_all_variations_for_item(itemid) variations = [cls(e) for e in r] for v in variations: v.tags = plenty.get_tag_relationships(v['tagVariationId']) v.session = plenty return variations @classmethod def get_all_with_tag_id(cls, plenty, tagid, withparam='all'): if withparam == 'all': withparam = plenty.VARIATIONFILTERS else: withparam = ','.join(withparam) fp = PlentyFilterParams({ 'variationTagId': tagid, 'with': withparam, }) return [PlentyVariation(v) for v in plenty.search_variations(fp)] @classmethod def get_by_id(cls, plenty, variationid, withparam='all'): if variationid: if withparam == 'all': withparam = plenty.VARIATIONFILTERS else: withparam = ','.join(withparam) pf = PlentyFilterParams({ 'with': withparam, 'id': variationid }) r = list(plenty.search_variations_lazy(pf)) if r: return r[0][0] @classmethod def get_all_changed_since(cls, plenty, start, withparam='all'): if withparam == 'all': withparam = plenty.VARIATIONFILTERS pf = PlentyFilterParams({ 'with': withparam, 'updatedBetween': start }) processed = set() for v, qty in plenty.search_variations_lazy(pf): processed.add(v['id']) yield v pf = PlentyFilterParams({ 'with': withparam, 'relatedUpdatedBetween': start }) for v, qty in plenty.search_variations_lazy(pf): if v['id'] not in processed: yield v @classmethod def get_minimal_changed_since(cls, plenty, start, withparams=''): """Gets all variations with changed to themselves or related data. Intended for use with minimal withparams since it gets all at once """ fp = PlentyFilterParams({ 'updatedBetween': start, 'with': withparams }) variations = dict((v['id'], v) for v in plenty.search_variations(fp)) fp = PlentyFilterParams({ 'relatedUpdatedBetween': start, 'with': withparams }) variations.update(dict((v['id'], v) for v in plenty.search_variations(fp))) return [PlentyVariation(v) for v in variations.values()] @classmethod def get_all_stock(cls, plenty): pf = PlentyFilterParams({ 'with': 'stock', }) for v, qty in plenty.search_variations_lazy(pf): yield v class PlentyManufacturer(PlentyObject): """All manufacturer details""" @property def name(self): return self['name'] @property def eu_responsible_person(self): countryname = EUISONAMESDE[COUNTRYISO[self['responsibleCountry']]] r = [ self['responsibleName'], f"{self['responsibleStreet']} {self['responsibleHouseNo']}", f"{self['responsiblePostCode']} {self['responsibleTown']}", countryname, self['responsibleEmail'] ] return ' | '.join(map(lambda x: x.strip(), r)) @classmethod def from_id(cls, plenty, manufacturerid): r = plenty.get_manufacturer(manufacturerid) man = cls() man.update_from_json(r.json()) return man class PlentyItem(PlentyObject): @property def images(self): return [PlentyImage(i) for i in self['itemImages']] @property def manufacturer(self): return PlentyManufacturer.from_id(self.session, self['manufacturerId']) @property def detext(self): return next(filter(lambda x: x['lang'] == 'de', self['texts'])) @property def name(self): return self.detext['name1'] @property def id(self): return self['id'] @property def description(self): return self.detext['description'] @property def has_variations(self): return len(self.variations) > 1 @property def main_variationid(self): return self['mainVariationId'] @property def main_variation(self): return self.variations[self.main_variationid] @property def latest_release_date(self): dates = [] for v in self.variations: if v.releasedate_datetime: dates.append(v.releasedate_datetime) if dates: return max(dates) def all_variations_have_stock_in_warehouse(self, warehouseid): stocks = [v.stock_for_warehouse(warehouseid) for v in self.variations] return min(stocks) > 0 @property def days_till_release(self): releasedate = self.latest_release_date if releasedate: now = datetime.now().astimezone() return (releasedate - now).days def get_images_for_market(self, market): images = [PlentyImage(i) for i in self['itemImages']] images = filter(lambda x: x.is_available_for_market(market), images) return sorted(images, key=lambda x: x.position) @classmethod def get_from_id(cls, plenty, itemid): r = plenty.get_item(itemid) item = cls(r.json()) item.variations = PlentyVariation.get_all_for_item_id(plenty, itemid) # keep session for later item.session = plenty return item @classmethod def get_by_id(cls, plenty, itemid, withparam='all'): if itemid: if withparam == 'all': withparam = plenty.ITEMFILTERS else: withparam = ','.join(withparam) pf = PlentyFilterParams({ 'with': withparam, 'id': itemid }) r = list(plenty.search_items_lazy(pf)) if r: item = r[0][0] variations = PlentyVariation.get_all_for_item_id(plenty, itemid) item.variations = dict(((v['id'], v) for v in variations)) # keep session for later item.session = plenty return item @classmethod def get_all_with_variation_tag(cls, plenty, tagid, withparam='all'): """Get all items with at least one variation with the specified tag""" variations = PlentyVariation.get_all_with_tag_id(plenty, tagid, withparam) if not variations: return [] # save variations for later so that we do not need to get them again variationmap = defaultdict(list) for v in variations: variationmap[v['itemId']].append(v) itemids = set([v['itemId'] for v in variations if v['isActive']]) items = [] for itemid in itemids: pf = PlentyFilterParams({ 'id': itemid, 'with': 'itemShippingProfiles' }) data = plenty.search_items(pf) if not data: continue for d in data: item = cls() item.update_from_json(d) item.variations = variationmap[itemid] items.append(item) return items @classmethod def get_all_changed_since(cls, plenty, start, withparam='all'): if withparam == 'all': withparam = plenty.ITEMFILTERS pf = PlentyFilterParams({ 'with': withparam, 'updatedBetween': start }) for pi, qty in plenty.search_items_lazy(pf): yield pi class PlentyVariationStockHistoryEntry(dict): fields = ['createdAt', 'quantity', 'reason', 'reasonString', 'storageLocationName', 'balance'] def __init__(self, psm, balance): self['createdAt'] = psm['createdAt'] self['quantity'] = psm['quantity'] self['reason'] = psm['reason'] self['reasonString'] = psm['reasonString'] self['storageLocationName'] = psm['storageLocationName'] self['balance'] = balance class PlentyVariationStockHistory: """Complete stock history for one variation""" def __init__(self, psml, current_stock): """psml - PlentyStockMovementList """ self.history = [] balance = current_stock for psm in sorted(psml, key=lambda x: x['createdAt'], reverse=True): e = PlentyVariationStockHistoryEntry(psm, balance) balance = balance - psm['quantity'] self.history.append(e) def to_csv(self, csvpath): with open(csvpath, 'w', newline='') as csvfile: writer = csv.DictWriter( csvfile, delimiter=';', fieldnames=PlentyVariationStockHistoryEntry.fields ) writer.writeheader() for row in self.history: writer.writerow(row) def get_available_timespans(self): """Returns timespans when the item was available in tuples in the form (fromDate, toDate). If the item ist currently available toDate is None""" def combine(timespans): combined = [] prev = None for e in chain(timespans, [(0, 0)]): if prev: if prev[1] != e[0]: combined.append(prev) prev = e else: prev = (prev[0], e[1]) else: prev = e return combined fromDate = None toDate = None timespans = [] # start with oldest for e in reversed(self.history): if e['balance'] > 0 and not fromDate: fromDate = e['createdAt'] if fromDate and e['balance'] == 0: toDate = e['createdAt'] if ((fromDate and toDate) and (fromDate != toDate)): timespans.append((fromDate, toDate)) fromDate = None toDate = None if fromDate and toDate is None: timespans.append((fromDate, toDate)) return combine(timespans) @classmethod def get_for_variation(cls, plenty, warehouseid, variationid): psml = PlentyStockMovementList.get_all(plenty, warehouseid, variationid) quantity = plenty.get_variation_stock_by_warehouse_value( variationid, warehouseid) quantity = quantity if quantity is not None else 0 return cls(psml, quantity) @classmethod def get_for_variation_from_date(cls, plenty, warehouseid, variationid, startdate): now = datetime.now().astimezone() psml = PlentyStockMovementList.get_all_for_timespan( plenty, warehouseid, variationid, startdate, now) quantity = plenty.get_variation_stock_by_warehouse_value( variationid, warehouseid) quantity = quantity if quantity is not None else 0 return cls(psml, quantity) class PlentyFilterParams(PlentyData): """Filter parameters such as timestamps stored as python objects can be easily converted to plenty json format here""" def plentyparamconvert(p): """convert a python type to a json plenty parameter""" if isinstance(p, datetime): dt = p.replace(microsecond=0).astimezone() return dt.isoformat() else: # try to convert datatype to json try: return p.json() except AttributeError: pass # pass ist raw return p def plentyfyparams(func): def wrapper(*args, **kwargs): newargs = [] for arg in args: newargs.append(plentyparamconvert(arg)) newkwargs = {} for k, v in kwargs.items(): newkwargs[k] = plentyparamconvert(v) return func(*newargs, **newkwargs) return wrapper class PlentyAPI(): """Map API calls and watch over limits and errors""" VARIATIONFILTERS = """properties,variationProperties,variationBarcodes, variationBundleComponents,variationComponentBundles,variationSalesPrices, marketItemNumbers,variationCategories,variationClients,variationMarkets, variationDefaultCategory,variationSuppliers,variationWarehouses,images, itemImages,variationAttributeValues,variationSkus,variationAdditionalSkus, unit,parent,item,stock""" ITEMFILTERS = """itemProperties,itemCrossSelling,variations,itemImages,\ itemShippingProfiles,ebayTitles""" def __init__(self, session): self.session = session.session self.baseurl = session.baseurl # read_limits = {'long': 166000, 'short': 200} # write_limits = {'long': 48000, 'short': 75} # be more conservative dont hammer the API as much read_limits = {'long': 166000, 'short': 180} write_limits = {'long': 48000, 'short': 50} self.limits = {'r': read_limits, 'w': write_limits} readcallsleft = read_limits.copy() writecallsleft = write_limits.copy() self.callsleft = {'r': readcallsleft, 'w': writecallsleft} read_decays = {'long': 0, 'short': 0} write_decays = {'long': 0, 'short': 0} self.decays = {'r': read_decays, 'w': write_decays} self.resourcelimits = {} self.resourcecallsleft = {} self.resourcedecays = {} def jsonifyparams(func): def wrapper(*args, **kwargs): try: args[1] = args[1].json() except AttributeError: pass return func(*args, **kwargs) return wrapper def jsonifydata(func): def wrapper(*args, **kwargs): try: kwargs['json'] = kwargs['json'].json() except AttributeError: pass return func(*args, **kwargs) return wrapper def get_required_delay(self, resource, access='r'): callsleft = self.callsleft[access] decay = self.decays[access] if callsleft['long'] <= 0: return decay['long'] elif callsleft['short'] <= 0: return decay['short'] elif resource in self.resourcecallsleft: if self.resourcecallsleft[resource] <= 0: return self.resourcedecays[resource] else: return 0 def update_limits_and_decays(self, resource, response): debug(self.limits['r']) debug(self.callsleft['r']) debug(self.decays['r']) if response.request.method == 'GET': access = 'r' else: access = 'w' h = response.headers limit = self.limits[access] left = self.callsleft[access] decay = self.decays[access] try: limit['short'] = int(h['X-Plenty-Global-Short-Period-Limit']) limit['long'] = int(h['X-Plenty-Global-Long-Period-Limit']) left['long'] = int(h['X-Plenty-Global-Long-Period-Calls-Left']) left['short'] = int(h['X-Plenty-Global-Short-Period-Calls-Left']) decay['long'] = int(h['X-Plenty-Global-Long-Period-Decay']) decay['short'] = int(h['X-Plenty-Global-Short-Period-Decay']) except KeyError as ke: warning(f"Exception {ke} encountered") if 'X-Plenty-Route-Limit' in h: # specifically throttled resource self.resourcelimits[resource] = int(h['X-Plenty-Route-Limit']) self.resourcecallsleft[resource] = int( h['X-Plenty-Route-Calls-Left']) self.resourcedecays[resource] = int(h['X-Plenty-Route-Decay']) def call_with_retry_on_429(func): def wrapper(*args, **kwargs): waited = 0 wait_on_rate_limit = 10 fail_after_seconds = 600 while True: r = func(*args, **kwargs) if r.status_code == 429: waited += wait_on_rate_limit if waited > fail_after_seconds: # let outside handel this return r time.sleep(wait_on_rate_limit) # wait double time one next try wait_on_rate_limit *= 2 else: return r return wrapper @call_with_retry_on_429 def get_with_retry(self, *args, **kwargs): return self.session.get(*args, **kwargs) @call_with_retry_on_429 def post_with_retry(self, *args, **kwargs): return self.session.post(*args, **kwargs) @jsonifyparams def make_single_get_request(self, resource, params={}): delay = self.get_required_delay(resource) if delay: debug('waiting {} seconds.'.format(delay)) time.sleep(delay) url = self.baseurl + resource debug('making call to resource "{}"'.format(resource)) response = self.get_with_retry(url, params=params) self.update_limits_and_decays(resource, response) return response def make_combined_get_request(self, resource, params={}, **kwargs): """Makes mutliple requests to get all result pages""" entries = [] if isinstance(params, PlentyData): params = params.json() params = params.copy() params.update(kwargs) while True: r = self.make_single_get_request(resource, params) r.raise_for_status() if not r.ok: warning('{} request to {} failed. Returning empty' 'result.'.format(r.request.method, r.request.url)) return [] else: json = r.json() if 'page' not in json: # seems to be non-paginated request return json entries.extend(json['entries']) if json['isLastPage']: return entries params['page'] = json['page'] + 1 @jsonifyparams def make_lazy_get_request(self, resource, params={}, returntype=PlentyObject, **kwargs): """Makes mutliple requests to get all result pages Returns iterator (returntype, remaining) tuples """ params = params.copy() params.update(kwargs) processed = 0 while True: r = self.make_single_get_request(resource, params) r.raise_for_status() json = r.json() for entry in json['entries']: processed += 1 yield (returntype(entry), json['totalsCount'] - processed) if 'page' not in json or json['isLastPage']: # non-paginated request or last page reached return params['page'] = json['page'] + 1 def make_single_delete_request(self, resource, json={}): delay = self.get_required_delay(resource) if delay: debug('waiting {} seconds.'.format(delay)) time.sleep(delay) url = self.baseurl + resource debug('making call to resource "{}"'.format(resource)) response = self.session.delete(url, json=json) self.update_limits_and_decays(resource, response) return response def make_single_put_request(self, resource, json={}): delay = self.get_required_delay(resource) if delay: debug('waiting {} seconds.'.format(delay)) time.sleep(delay) url = self.baseurl + resource debug('making call to resource "{}"'.format(resource)) response = self.session.put(url, json=json) self.update_limits_and_decays(resource, response) return response @jsonifydata def make_single_post_request(self, resource, json={}): delay = self.get_required_delay(resource) if delay: debug('waiting {} seconds.'.format(delay)) time.sleep(delay) url = self.baseurl + resource debug('making call to resource "{}"'.format(resource)) response = self.post_with_retry(url, json=json) self.update_limits_and_decays(resource, response) return response def make_combined_post_request(self, resource, json={}): """Makes mutliple requests to get all result pages""" entries = [] json = json.copy() while True: r = self.make_single_post_request(resource, json=json) r.raise_for_status() if not r.ok: warning('{} request to {} failed. Returning empty' 'result.'.format(r.request.method, r.request.url)) return [] else: returnedjson = r.json() if 'page' not in returnedjson: # seems to be non-paginated request return returnedjson entries.extend(returnedjson['entries']) if returnedjson['isLastPage']: return entries json['page'] = returnedjson['page'] + 1 time.sleep(3) def make_combined_post_request_lazy(self, resource, json={}): """Makes mutliple requests to get all result pages""" entries = [] json = json.copy() while True: r = self.make_single_post_request(resource, json=json) r.raise_for_status() if not r.ok: warning('{} request to {} failed. Returning empty' 'result.'.format(r.request.method, r.request.url)) return [] else: returnedjson = r.json() if 'page' not in returnedjson: # seems to be non-paginated request for e in returnedjson: yield e break for e in returnedjson['entries']: yield e if returnedjson['isLastPage']: break json['page'] = returnedjson['page'] + 1 def make_single_delete_request(self, resource, json={}): delay = self.get_required_delay(resource) if delay: debug('waiting {} seconds.'.format(delay)) time.sleep(delay) url = self.baseurl + resource debug('making call to resource "{}"'.format(resource)) response = self.session.delete(url, json=json) self.update_limits_and_decays(resource, response) return response def get_user(self): return self.make_combined_get_request('user') def get_item(self, itemid, withparam='all'): if withparam == 'all': fp = PlentyFilterParams({'with': self.ITEMFILTERS}) else: fp = PlentyFilterParams() response = self.make_single_get_request(f'items/{itemid}', params=fp.json()) response.raise_for_status() return response def put_pim_variation_property(self, variationid, propertyid, groupid, value): data = [ { "variationId": variationid, "propertyId": propertyid, "groupId": groupid, "value": value } ] response = self.make_single_put_request( f'pim/variations/properties', json=data) response.raise_for_status() return response def delete_pim_variation_property(self, variationid, propertyid, groupid): data = [ { "variationId": variationid, "propertyId": propertyid, "groupId": groupid, } ] response = self.make_single_delete_request( f'pim/variations/properties', json=data) response.raise_for_status() return response def put_pim_variation_text_property(self, variationid, propertyid, groupid, value, lang="DE"): data = [ { "variationId": variationid, "propertyId": propertyid, "groupId": groupid, "values": [ { "lang": lang, "value": value } ] } ] response = self.make_single_put_request( f'pim/variations/properties', json=data) response.raise_for_status() return response def post_property_relations(self, data): r = self.make_single_post_request(f'v2/properties/relations', json=data) r.raise_for_status() return r def post_property_selection(self, variationid, propertyid, groupid, selectionids): if type(selectionids) == int: selectionids = [selectionids] selectionvalues = [{'selectionId': i} for i in selectionids] data = [{ "type": "item", "targetId": variationid, "propertyId": propertyid, "groupId": groupid, "selectionValues": selectionvalues }] return self.post_property_relations(data) def get_listings_markets(self, **params): if 'withparam' in params: params['with'] = params['withparam'] fp = PlentyFilterParams(params) r = self.make_combined_get_request( f'listings/markets', **fp.json() ) return r def get_listings_markets_histories(self, **params): if 'withparam' in params: params['with'] = params['withparam'] fp = PlentyFilterParams(params) r = self.make_combined_get_request( f'listings/markets/histories', **fp.json() ) return r def get_listings_markets_lazy(self, **params): if 'withparam' in params: params['with'] = params['withparam'] fp = PlentyFilterParams(params) r = self.make_lazy_get_request( f'listings/markets', params=fp ) return r def get_listings_markets_mlid(self, mlid, withparam='properties'): fp = PlentyFilterParams({ 'with': withparam }) response = self.make_single_get_request( f'listings/markets/{mlid}', params=fp.json()) response.raise_for_status() return response def put_listings_markets(self, mlid, data): response = self.make_single_put_request( f'listings/markets/{mlid}', json=data.json() ) response.raise_for_status() return response def put_listings_markets_shippingprofile(self, mlid, referrerid, shippingprofileid): data = PlentyData({ 'id': mlid, 'referrerId': referrerid, 'properties': [{ 'listingMarketId': 21495, 'typeId': 1, 'value': shippingprofileid }] }) return self.put_listings_markets(mlid, data) def put_listings_markets_texts(self, mlid, description=None, title=None, lang='de'): data = PlentyData({ 'listingMarketId': mlid, 'lang': lang, }) if description is not None: data.update({'description': description}) if title is not None: data.update({'title': title}) response = self.make_single_put_request( f'listings/markets/texts/{mlid}/{lang}', json=data.json() ) response.raise_for_status() return response def put_listings_markets_histories_updates(self, mlids, options=['all']): data = PlentyData({ 'id': mlids, 'options': options }) response = self.make_single_put_request( f'listings/markets/histories/update', json=data.json() ) response.raise_for_status() return response def get_property_selection_names(self, selectionid): response = self.make_single_get_request( f'v2/properties/selections/{selectionid}/names') response.raise_for_status() return response def get_manufacturer(self, manufacturerid): response = self.make_single_get_request( f'items/manufacturers/{manufacturerid}') response.raise_for_status() return response def get_tag_relationships(self, relationshipvalue, tagtype='variation'): fp = PlentyFilterParams({'with': 'tag', 'tagType': tagtype, 'relationshipValue': relationshipvalue}) r = self.make_combined_get_request(f'tags/relationships', **fp.json()) return r def get_all_variations_for_item(self, itemid, withparam='all'): if withparam == 'all': fp = PlentyFilterParams({'with': self.VARIATIONFILTERS}) else: fp = PlentyFilterParams() r = self.make_combined_get_request(f'items/{itemid}/variations', **fp.json()) return r def get_orders(self, filterparams=None, **kwargs): if filterparams: ordersjson = self.make_combined_get_request('orders', **filterparams.json()) else: ordersjson = self.make_combined_get_request('orders', **kwargs) return PlentyOrderList(ordersjson) def get_orders_created_within_24h(self): """actual orders (without refunds) in he last 24 hours""" end = datetime.now() start = end - timedelta(hours=24) fp = PlentyFilterParams({'createdAtFrom': start, 'createdAtTo': end, 'orderType': 1}) return self.get_orders(**fp.json()) def get_orders_with_status(plenty, status): """get orders with specific status""" fp = PlentyFilterParams({'statusFrom': status, 'statusTo': status}) return plenty.get_orders(**fp.json()) def get_orders_with_status_range(plenty, statusfrom, statusto): """get orders with specific status range""" fp = PlentyFilterParams({'statusFrom': statusfrom, 'statusTo': statusto}) return plenty.get_orders(**fp.json()) def get_orders_by_external_id(self, externalid): fp = PlentyFilterParams({'externalOrderId': externalid, 'orderType': 1}) return self.get_orders(**fp.json()) def get_open_orders(self, preorder=False): """this will check for ready to be shipped orders based on status""" if preorder: statusto = 5.7 else: statusto = 5.089 fp = PlentyFilterParams({'statusFrom': 4, 'statusTo': statusto, 'orderType': 1}) return self.get_orders(**fp.json()) def get_order_document(self, orderid, dtype=""): resource = f'orders/{orderid}/documents/downloads/{dtype}' response = self.make_single_get_request(resource) response.raise_for_status() return response def get_order_documents_as_zip(self, orderids, filename=""): resource = f'orders/documents/downloads/as_zip' params = { 'ids[]': list(orderids), } if filename: params['filename'] = filename response = self.make_single_get_request(resource, params) response.raise_for_status() return response def get_order_documents(self, params={}, **kwargs): resource = f'orders/documents/find' return self.make_single_get_request(resource, params, **kwargs) def get_order_documents_lazy(self, params={}, **kwargs): resource = f'orders/documents/find' return self.make_lazy_get_request(resource, params, **kwargs) def get_package_numbers(self, orderid): resource = f'orders/{orderid}/packagenumbers' response = self.make_single_get_request(resource) return response.json() def post_shipping_package(self, orderid, packagenumber, packageid=1, weight=0, packagetype=0, volume=0): """Add a package with number to an order Privilege settings under: Berechtigungen->Aufträge->Versand->Versandpaket->Erstellen packagenumber - string """ data = PlentyData({ 'packageId': packageid, 'weight': weight, 'packageNumber': packagenumber, 'packageType': packagetype, 'volume': volume}) response = self.make_single_post_request(f'orders/{orderid}/shipping/' f'packages', json=data.json()) response.raise_for_status() return response def put_variation_barcode(self, itemid, variationid, barcodeid, data): path = (f'items/{itemid}/variations/{variationid}/variation_barcodes/' f'{barcodeid}') r = self.make_single_put_request(path, json=data.json()) r.raise_for_status() return r def delete_item_image(self, itemid, imageid): response = self.make_single_delete_request(f'items/{itemid}/images/' f'{imageid}') response.raise_for_status() return response def delete_order_item(self, orderid, orderitemid): response = self.make_single_delete_request(f'orders/{orderid}/items/' f'{orderitemid}') response.raise_for_status() return response def put_order(self, orderid, data): response = self.make_single_put_request(f'orders/{orderid}', json=data.json()) response.raise_for_status() return response def put_order_status(self, orderid, statusid): data = PlentyData({'statusId': statusid}) response = self.put_order(orderid, data) response.raise_for_status() return response def post_item(self, data): response = self.make_single_post_request(f'items', json=data.json()) response.raise_for_status() return response def post_order_comment(self, orderid, text, userid): data = PlentyData({ 'referenceType': 'order', 'referenceValue': orderid, 'text': text, 'userId': userid, 'isVisibleForContact': False }) response = self.make_single_post_request('comments', json=data.json()) response.raise_for_status() return response def post_item_variation_price(self, itemid, variationid, priceid, price): """Add a price to a variation Privilege settings under: Berechtigungen->Artikel->Variante->Preis->Erstellen priceid - short integer identifying the sales price, default is usually 1 price - the actual price value (float) """ data = PlentyData({'variationId': variationid, 'salesPriceId': priceid, 'price': price}) response = self.make_single_post_request(f'items/{itemid}/variations/' f'{variationid}/' f'variation_sales_prices', json=data.json()) response.raise_for_status() return response def post_tag_relationship(self, tagid, tagtype, targetid): """Attaches a tag to a resource. Privilege settings under: Berechtigungen->Einrichtung->Tag->Tag-Verknüpfung->Erstellen The targetid is the unique id identifier of the entity that should get the tag. For tagtype 'variation' this would be the 'id' of the variation. """ assert tagtype in ('category', 'ticket', 'variation', 'board', 'contact') data = PlentyData({'tagId': tagid, 'tagType': tagtype, 'relationshipValue': targetid}) r = self.make_single_post_request(f'tags/relationships', json=data.json()) r.raise_for_status() return r def post_item_shippingprofile(self, itemid, profileid): """Activate a shipping profile on an item profileid - short integer number identifying the shipping profile """ data = PlentyData({'itemId': itemid, 'profileId': profileid}) r = self.make_single_post_request(f'items/{itemid}/' f'item_shipping_profiles', json=data.json()) r.raise_for_status() return r def post_item_variation_market(self, itemid, variationid, marketid): data = PlentyData({'variationId': variationid, 'marketId': marketid}) r = self. make_single_post_request( f"items/{itemid}/variations/{variationid}/variation_markets", json=data.json() ) r.raise_for_status() return r def post_item_image_availability(self, itemid, imageid, value, typ='marketplace'): data = PlentyData({'imageId': imageid, 'type': typ, 'value': value}) r = self. make_single_post_request( f"items/{itemid}/images/{imageid}/availabilities", json=data.json() ) r.raise_for_status() return r def post_item_image(self, itemid, data): response = self.make_single_post_request(f'items/{itemid}/' f'images/upload', json=data.json()) response.raise_for_status() return response def post_order(self, data): response = self.make_single_post_request(f'orders', json=data.json()) response.raise_for_status() return response def post_address(self, data): response = self.make_single_post_request(f'accounts/addresses', json=data.json()) response.raise_for_status() return response def get_payments_by_search(self, params): response = self.make_single_get_request(f'payments/search', params=params) response.raise_for_status() return response def get_payment_properties(self, paymentid): response = self.make_single_get_request( f'payments/{paymentid}/properties') response.raise_for_status() return response def get_payments_for_order(self, oid): response = self.make_combined_get_request(f'payments/orders/{oid}') return response def post_payment(self, data): response = self.make_single_post_request(f'payments', json=data.json()) response.raise_for_status() return response def post_payment_order_relation(self, paymentid, orderid): now = datetime.now().replace(microsecond=0).astimezone().isoformat() data = PlentyData({ 'paymentId': paymentid, 'orderId': orderid, 'assignedAt': now }) uri = f'payment/{paymentid}/order/{orderid}/validation' response = self.make_single_post_request(uri, json=data.json()) response.raise_for_status() return response def get_address(self, aid): response = self.make_single_get_request(f'accounts/addresses/{aid}') response.raise_for_status() return response def search_variations(self, filterparams): response = self.make_combined_get_request( f"items/variations", params=filterparams.json() ) return response def search_items(self, filterparams): response = self.make_combined_get_request(f'items', params=filterparams) return response def search_items_lazy(self, filterparams): response = self.make_lazy_get_request(f'items', params=filterparams, returntype=PlentyItem) return response def search_variations_lazy(self, filterparams): response = self.make_lazy_get_request(f'items/variations', params=filterparams, returntype=PlentyVariation) return response def search_orders(self, params, withparam=""): if withparam: params.update({'with': withparam}) response = self.make_single_get_request(f'orders/search', params=params) response.raise_for_status() return response def search_orders_post(self, filterparams): response = self.make_combined_post_request(f'orders/search', filterparams) return response def search_orders_post_lazy(self, filterparams): response = self.make_combined_post_request_lazy(f'orders/search', filterparams) return response def get_order(self, orderid, withparam=""): if withparam: params = {'with': withparam} else: params = {} response = self.make_single_get_request(f'orders/{orderid}', params=params) response.raise_for_status() return response def get_order_properties_types(self, lang={'lang': ['de']}): response = self.make_single_get_request(f'orders/properties/types', params=lang) response.raise_for_status() return response def print_order_properties_types(self, lang={'lang': ['de']}): r = self.get_order_properties_types(lang) types = r.json() types.sort(key=lambda x: x['id']) for t in types: tid = t['id'] tname = t['names'][0]['name'] print(f"{tid:4}: {tname}") def get_variation(self, fp): response = self.make_single_get_request(f'items/variations', params=fp.json()) response.raise_for_status() return response def get_variations(self, fp): return self.make_combined_get_request(f'items/variations', params=fp.json()) def get_lazy_variation(self, fp): """Get variations on demand only. Returns an iterator over PlentyVariation objects """ return self.make_lazy_get_request(f'items/variations', params=fp, returntype=PlentyVariation) @plentyfyparams def get_status_history(self, orderId=None, statusId=None, createdAtFrom=None, createdAtTo=None, userId=None): """Get status history possible filterparams orderId: the ID of the order statusId: the ID of the status createdAtFrom: set a start date to get status history entries beginning with this date createdAtTo: set an end date to get status history entries up to this date userId: the ID of the user """ params = ['orderId', 'statusId', 'createdAtFrom', 'createdAtTo', 'userId'] args = locals() params = dict([(p, args[p]) for p in params if args[p]]) return self.make_combined_get_request(f'orders/status-history', params) def bulk_update_sales_prices(self, salespriceid, pricedict): """Bulk update sales prices. pricedict maps variationid->price""" data = ({"price": p, "salesPriceId": salespriceid, "variationId": v} for v, p in pricedict.items()) for chunk in itertools.batched(data, 50): r = self.make_single_put_request( 'items/variations/variation_sales_prices', json=chunk ) r.raise_for_status() return r def put_item(self, itemid, data): response = self.make_single_put_request(f'items/{itemid}', json=data.json()) response.raise_for_status() return response def put_item_variation(self, itemid, variationid, data): response = self.make_single_put_request(f'items/{itemid}/variations/' f'{variationid}', json=data.json()) response.raise_for_status() return response def touch_item_variation(self, itemid, variationid): """sets the updatedAt timestamp to now""" return self.put_item_variation(itemid, variationid, PlentyData({'updatedAt': ''})) def get_item_variation_description(self, itemid, variationid): resource = f"items/{itemid}/variations/{variationid}/descriptions/"\ f"de" response = self.make_single_get_request(resource) response.raise_for_status() return response.json() def put_item_variation_description(self, itemid, variationid, data, lang='de'): response = self.make_single_put_request(f'items/{itemid}/variations/' f'{variationid}/descriptions/' f'{lang}', json=data.json()) response.raise_for_status() return response def get_variation_by_id(self, vid, withparam=''): if not withparam: withparam = 'item' fp = PlentyFilterParams({'with': withparam, 'id': vid}) return self.get_variation(fp) def get_variation_by_number(self, number, withparam=''): if not withparam: withparam = 'item' fp = PlentyFilterParams({'with': withparam, 'numberExact': number}) return self.get_variation(fp) def get_variation_stock_by_warehouse(self, variationid, warehouseid): resource = f'stockmanagement/warehouses/{warehouseid}/stock' fp = PlentyFilterParams({'variationId': variationid}) response = self.make_single_get_request(resource, fp) response.raise_for_status() return response def get_variation_stock_by_warehouse_value(self, variationid, warehouseid): resource = f'stockmanagement/warehouses/{warehouseid}/stock' fp = PlentyFilterParams({'variationId': variationid}) response = self.make_single_get_request(resource, fp) response.raise_for_status() entries = response.json()['entries'] if entries: return entries[0]['stockPhysical'] else: # could not be determined for example for bundle return None def get_storage_location_by_warehouse(self, locationid, warehouseid): resource = f'stockmanagement/warehouses/{warehouseid}/management/'\ f'storageLocations/{locationid}' response = self.make_single_get_request(resource) response.raise_for_status() return response def get_stock_locations_by_warehouse(self, variationid, warehouseid): resource = f'stockmanagement/warehouses/{warehouseid}/stock/'\ f'storageLocations' fp = PlentyFilterParams({'variationId': variationid}) response = self.make_single_get_request(resource, fp) response.raise_for_status() return response def get_stock(self, fp): response = self.make_combined_get_request(f'stockmanagement/stock', **fp.json()) response.raise_for_status() return response def get_stock_movements(self, warehouseid, params={}, **kwargs): resource = f"stockmanagement/warehouses/{warehouseid}/stock/movements" response = self.make_combined_get_request(resource, params, **kwargs) return response def get_stock_movements_lazy(self, warehouseid, params={}, **kwargs): resource = f"stockmanagement/warehouses/{warehouseid}/stock/movements" response = self.make_lazy_get_request(resource, params, **kwargs) return response def put_stock_correction(self, variationid, warehouseid, stock, locid=0): resource = f"stockmanagement/warehouses/{warehouseid}/stock/correction" data = PlentyData({"corrections": [{"variationId": variationid, "quantity": stock, "storageLocationId": locid, "reasonId": 301}]}) response = self.make_single_put_request(resource, json=data.json()) if response.status_code == requests.codes.ok: info(f"Stock of {variationid} in {warehouseid} " f"at location {locid} changed to {stock}") return True else: error(f"Failed to change stock of {variationid} in " f"{warehouseid} at {locid} to {stock}") return False def get_bundle_components(self, itemid, variationid): resource = f"items/{itemid}/variations/{variationid}/"\ f"variation_bundles" response = self.make_single_get_request(resource) response.raise_for_status() return response.json() def put_book_outgoing_stock(self, variationid, warehouseid, quantity): if quantity >= 0: error(f"Negative quantity {quantity} for {variationid} " f"required.") return resource = f"stockmanagement/warehouses/{warehouseid}/stock/"\ f"bookOutgoingItems" data = PlentyData({"outgoingItems": [{"variationId": variationid, "quantity": quantity, "reasonId": 201, "currency": "EUR", "deliveredAt": datetime.now()}]}) # try default "storageLocationId": 0, response = self.make_single_put_request(resource, json=data.json()) if response.status_code == requests.codes.ok: info(f"Stock of {variationid} in {warehouseid} reduced by" f" {quantity}") return True else: error(f"Failed to reduce stock of {variationid} in " f"{warehouseid} by {quantity}") return False def change_stock_of_variation(self, variationid, warehouseid, difference): "change stock of variation" r = self.get_variation_stock_by_warehouse(variationid, warehouseid) try: # TODO this will not work if more storage locations than default in # use stock = r.json()['entries'][0]['stockPhysical'] except IndexError: warning(f"Stock for variation id '{variationid}' in warehouse " f"'{warehouseid} could not be determined") return None stock = stock + difference self.put_stock_correction(variationid, warehouseid, stock) def change_stock_of_variation_number(self, number, difference): r = self.get_variation_by_number(number) try: details = r.json()['entries'][0] except IndexError: warning(f"Variation number '{number}' not found") return None if details['bundleType'] != "bundle": # single item self.change_stock_of_variation(details['id'], details['mainWarehouseId'], difference) else: # bundle item components = self.get_bundle_components(details['itemId'], details['id']) for c in components: self.change_stock_of_variation(c['componentVariationId'], details['mainWarehouseId'], (c['componentQuantity'] * difference)) def book_outgoing_stock_of_variation_number(self, number, quantity, warehouseid_override=False): r = self.get_variation_by_number(number) try: details = r.json()['entries'][0] except IndexError: warning(f"Variation number '{number}' not found") return None if warehouseid_override: warehouseid = warehouseid_override else: warehouseid = details['mainWarehouseId'] if details['bundleType'] != "bundle": # single item self.put_book_outgoing_stock(details['id'], warehouseid, quantity) else: # bundle item components = self.get_bundle_components(details['itemId'], details['id']) for c in components: self.put_book_outgoing_stock(c['componentVariationId'], warehouseid, (c['componentQuantity'] * quantity)) def action_shell(args): with PlentySession('https://www.netzteilcheck.de/rest/', 'pyrest', 'PexC6HQNGJ247NC41Dnd') as plentysession: plenty = PlentyAPI(plentysession) import pdb;pdb.set_trace() ACTIONMAP = {'shell': action_shell} if __name__ == '__main__': parser = argparse.ArgumentParser(description='Plentymarkets REST ' 'interface') parser.add_argument('action', choices=ACTIONMAP.keys(), metavar="ACTION", help='Specify action.') parser.add_argument('identifier', nargs='?', metavar="IDENTIFIER", help='specify identifier i.e. orderid') parser.add_argument('-v', '--verbose', action='store_const', const='DEBUG', default='INFO', help='display more details') args = parser.parse_args() if args.verbose == 'DEBUG': logging.basicConfig(level=logging.DEBUG) # call main action ACTIONMAP[args.action](args)