#!/usr/bin/env python3 # A little tool to make REST requests to the shopware API import argparse import csv import collections import itertools import logging import os.path import time import urllib.request import sys from collections import Counter, defaultdict from datetime import datetime, timedelta from functools import wraps from itertools import chain from logging import debug, info, error from uuid import uuid4 from requests_oauthlib import OAuth2Session, requests from requests.exceptions import HTTPError from oauthlib.oauth2 import LegacyApplicationClient, BackendApplicationClient from oauthlib.oauth2.rfc6749.errors import TokenExpiredError warnings_present = False SHOPWARE_PRODUCT_RELATIONSHIPS = ( 'canonicalProduct', 'categories', 'categoriesRo', 'children', 'cmsPage', 'configuratorSettings', 'cover', 'crossSellingAssignedProducts', 'crossSellings', 'customFieldSets', 'deliveryTime', 'downloads', 'featureSet', 'mainCategories', 'manufacturer', 'media', 'options', 'orderLineItems', 'prices', 'productReviews', 'properties', 'searchKeywords', 'seoUrls', 'streams', 'tags', 'tax', 'translations', 'unit', 'visibilities', 'wishlists' ) VATFACTORTABLE = { '018eeb96f3567336a3b9db72ca7d7b31': 1.19, '018eeb96f3567336a3b9db72cae92d34': 1.07 } VATFACTORTABLEREVERSE = dict((v, k) for k, v in VATFACTORTABLE.items()) CURRENCYIDTABLE = { 'EUR': 'b7d2554b0ce847cd82f3ac9bd1c0dfca' } AVAILABILITIES = { 0: '018eeb97057d712a8ceec8ca618826a6', 1: '018eeb96f34c72bf9b4d9ee84060fb03', 2: '018eeb96f34c72bf9b4d9ee8406ffdd4', 3: '018eeb96f34d731c89683defc3522637', 4: '018eeb96f34d731c89683defc4369ce6', 5: '01967c44ea927daf998bfe65c8716f79', 6: '01967c4579f678f18919c7e60f14963e', 7: '01967c45be637e4db5ea88b96f649c14', 8: '01967c4692ac72bdbfec4ee6151924f0', 9: '01967c46f6be7b4f981d40c59501f5d8' } AVAILABILITIESREVERSE = dict((v, k) for k, v in AVAILABILITIES.items()) class ShopwareError(Exception): pass def warning(*args): global warnings_present warnings_present = True logging.warning(*args) class ShopwareSession(): """Authenticate and hold details to the current session""" clientid = 'sr.py' def __init__(self, baseurl, clientid, clientsecret, scope=[], languageid=None): self.baseurl = baseurl self.clientid = clientid self.clientsecret = clientsecret self.scope = scope self.languageid = languageid self.writedelayinseconds = 0.1 self.readdelayinseconds = 0.1 def open_session(self): client = BackendApplicationClient(client_id=self.clientid) self.session = OAuth2Session(client=client) self.token = self.session.fetch_token(self.baseurl + 'oauth/token', client_id=self.clientid, client_secret=self.clientsecret) def __enter__(self): self.open_session() return self def __exit__(self, objtype, value, traceback): # shopware does not seem to need a logout pass def refresh_token_if_nescessary(method): @wraps(method) def _impl(self, *args, **kwargs): try: return method(self, *args, **kwargs) except TokenExpiredError: self.open_session() return method(self, *args, **kwargs) return _impl def write_delay(self): time.sleep(self.writedelayinseconds) def read_delay(self): time.sleep(self.readdelayinseconds) @refresh_token_if_nescessary def get(self, apipath): self.read_delay() if self.languageid: headers = {'sw-language-id': self.languageid} else: headers = {} r = self.session.get(self.baseurl + apipath, headers=headers) r.raise_for_status() return r def delete(self, apipath): self.write_delay() if self.languageid: headers = {'sw-language-id': self.languageid} else: headers = {} r = self.session.delete(self.baseurl + apipath, headers=headers) r.raise_for_status() return r @refresh_token_if_nescessary def post_json(self, apipath, json): self.read_delay() if self.languageid: headers = {'sw-language-id': self.languageid} else: headers = {} r = self.session.post(self.baseurl + apipath, json=json, headers=headers) r.raise_for_status() return r @refresh_token_if_nescessary def post_json_paginated(self, apipath, json, limit, page): self.read_delay() if self.languageid: headers = {'sw-language-id': self.languageid} else: headers = {} json = json.copy() json['limit'] = limit json['page'] = page r = self.session.post(self.baseurl + apipath, json=json, headers=headers) r.raise_for_status() return r @refresh_token_if_nescessary def post_file(self, apipath, contenttype, data): self.read_delay() if self.languageid: headers = {'sw-language-id': self.languageid} else: headers = {} headers['Content-Type'] = contenttype r = self.session.post(self.baseurl + apipath, data=data, headers=headers) r.raise_for_status() return r def get_json(self, apipath): r = self.get(apipath) r.raise_for_status() return r.json() def get_json_paginated(self, apipath, limit, page): self.read_delay() if self.languageid: headers = {'sw-language-id': self.languageid} else: headers = {} params = {'limit': limit, 'page': page} r = self.session.get(self.baseurl + apipath, params=params, headers=headers) r.raise_for_status() return r @refresh_token_if_nescessary def patch_json(self, apipath, json): self.write_delay() if self.languageid: headers = {'sw-language-id': self.languageid} else: headers = {} r = self.session.patch(self.baseurl + apipath, json=json, headers=headers) r.raise_for_status() return r def list_products(self, searchfilter=None): if not searchfilter: page = 1 limit = 500 data = [] while True: r = self.get_json_paginated('product', limit, page) d = r.json() data.extend(d['data']) if len(d['data']) < limit: return data page += 1 else: page = 1 limit = 500 data = [] while True: r = self.post_json_paginated( 'search/product', searchfilter, limit, page ) d = r.json() data.extend(d['data']) if len(d['data']) < limit: return data page += 1 def list_stock(self): sf = {'includes': {'product': ['id', 'name', 'productNumber', 'stock', 'active', 'isCloseout', 'shippingFree']}, 'sort': [{'field': 'productNumber', 'order': 'ASC'}]} return self.list_products(sf) def search_manufacturer(self, name): searchfilter = {'type': 'equals', 'field': 'name', 'value': name} r = self.post_json('search/product-manufacturer', json={'filter': [searchfilter]}) r.raise_for_status() return r.json() def get_product(self, uuid): return self.get_json(f'product/{uuid}') def get_product_post(self, uuid, body={}): body_extra = { 'ids': [uuid] } body = body.copy() body.update(body_extra) return self.post_json(f'search/product', json=body).json() def create_product_minimal(self, product_number): body = { 'name': 'minimal', 'productNumber': product_number, 'stock': 0, 'taxId': VATFACTORTABLEREVERSE[1.19], 'price': [ { 'currencyId': CURRENCYIDTABLE['EUR'], 'gross': 10000, 'net': 10000/1.19, 'linked': False } ] } return self.post_json('product', json=body) def create_product_visibility(self, payload): return self.post_json('product-visibility', json=payload) def create_media(self, payload): uuid = payload['id'] # create empty media self.post_json('media', json={'id': uuid}) # upload from url extension = payload['url'].rsplit('.', maxsplit=1)[1].lower() if extension == 'jpeg': extension = 'jpg' r = urllib.request.urlopen(payload['url']) contenttype = r.headers.get_content_type() f = self.post_file( f"_action/media/{uuid}/upload?extension={extension}", contenttype, r.read() ) # change filename self.patch_json(f"media/{uuid}", json={'fileName': payload['fileName']}) def create_product_media(self, payload): return self.post_json('product-media', json=payload) def create_order_delivery(self, payload): return self.post_json('order-delivery', json=payload) def get_media(self, entityid): return self.get_json(f'media/{entityid}') def delete_product_visibility(self, entityid): return self.delete(f"product-visibility/{entityid}") def delete_product_media(self, entityid): return self.delete(f"product-media/{entityid}") def delete_media(self, entityid): return self.delete(f"media/{entityid}") def delete_product_category(self, productid, categoryid): return self.delete(f"product/{productid}/categories/{categoryid}") def get_product_by_wmnumber(self, wmnumber): """Alias for old scripts""" return self.get_product_by_number(wmnumber) def get_product_by_number(self, prodno, relationships=[]): searchfilter = {'type': 'equals', 'field': 'productNumber', 'value': prodno} assocdict = dict(((rel, {}) for rel in relationships)) r = self.post_json('search/product', json={'filter': [searchfilter], 'associations': assocdict}) return r def get_order_by_swnumber(self, swnumber): searchfilter = {'type': 'equals', 'field': 'orderNumber', 'value': swnumber} r = self.post_json('search/order', json={'filter': [searchfilter]}) json = r.json() # change in place if len(json['data']) > 0: json['data'] = json['data'][0] json['included'] = json['included'][0] return ShopwareOrder(json, session=self) return None def get_country_iso_by_order_address(self, uuid): r = self.get(f'order-address/{uuid}/country') return r.json()['data'][0]['attributes']['iso'] def get_order_by_uuid(self, uuid): r = self.get(f'order/{uuid}') return ShopwareOrder(r.json(), session=self) def get_order_deliveries(self, orderuuid): searchfilter = {'type': 'equals', 'field': 'orderId', 'value': orderuuid} r = self.post_json('search/order-delivery', json={'filter': [searchfilter]}) json = r.json() return json def get_state_machine_state(self, uuid): r = self.get(f'state-machine-state/{uuid}') return r def patch_order_delivery(self, uuid, json): r = self.patch_json(f'order-delivery/{uuid}', json=json) r.raise_for_status() return r def patch_product(self, uuid, json): r = self.patch_json(f'product/{uuid}', json=json) r.raise_for_status() return r def set_order_state(self, uuid, action_name): json = {'sendMail': False} r = self.post_json(f'_action/order/{uuid}/state/{action_name}', json=json) return r def set_order_payment_state(self, uuid, action_name): json = {'sendMail': False} r = self.post_json( f'_action/order_transaction/{uuid}/state/{action_name}', json=json) return r def set_order_delivery_state(self, uuid, action_name): json = {'sendMail': False} r = self.post_json(f'_action/order_delivery/{uuid}' f'/state/{action_name}', json=json) return r class ShopwareOrder(dict): PREFETCH_FIELDS = ['addresses', 'lineItems', 'transactions', 'order-customer'] FILTER_ORDER_BY_IDS = { "associations": { "deliveries": { "associations": { "stateMachineState": {} } }, "transactions": { "associations": { "stateMachineState": {} } } }, "filter": [ { 'type': 'equals', 'field': 'orderNumber', 'value': '' } ] } def __init__(self, *args, session=None, prefetch=True, **kwargs): super().__init__(*args, **kwargs) self.session = session self.patches = defaultdict(list) self.deletes = defaultdict(list) self.creates = defaultdict(list) # prefetch interesting fields if prefetch: self.prefetch() def prefetch(self): """get additional order details""" for fn in self.PREFETCH_FIELDS: self[fn] = self.get_field(fn) def get_field(self, fieldname): oid = self['data']['id'] r = self.session.get(f'order/{oid}/{fieldname}') json = r.json() if fieldname == 'addresses': for address in json['data']: uuid = address['id'] ciso = self.session.get_country_iso_by_order_address(uuid) address['attributes']['countryiso'] = ciso return json def get_state_name(self): stateid = self['data']['attributes']['stateId'] try: first = self['included'][0] except KeyError: included = [self['included']] else: included = self['included'] d = dict(((e['id'], e) for e in included)) return d[stateid]['attributes']['technicalName'] def is_paid(self): stateid = self['transactions']['data'][0]['attributes']['stateId'] return (stateid == '018eeb96f3c9722fa2457fe0bd25b9f8' or stateid == '018eeb96f3c9722fa2457fe0bf70bbbd') def get_payment_method_id(self): return self['transactions']['data'][0]['attributes']['paymentMethodId'] def get_state_name(self): stateid = self['data']['attributes']['stateId'] r = self.session.get_state_machine_state(stateid) return r.json()['data']['attributes']['technicalName'] def set_state_name(self, action_name): uuid = self['data']['id'] r = self.session.set_order_state(uuid, action_name) return r def get_delivery_state_name(self): uuid = self['data']['id'] r = self.session.get_order_deliveries(uuid) stateid = r['data'][0]['attributes']['stateId'] r = self.session.get_state_machine_state(stateid) return r.json()['data']['attributes']['technicalName'] def set_delivery_state_name(self, action_name): uuid = self['data']['id'] r = self.session.get_order_deliveries(uuid) if len(r['data']) > 1: warning(f'more than one delivery found for order {uuid}') duuid = r['data'][0]['id'] r = self.session.set_order_delivery_state(duuid, action_name) return r def set_payment_state_name(self, action_name): puuid = self.get_first_included_by_type('order_transaction')['id'] r = self.session.set_order_payment_state(puuid, action_name) return r def get_delivery_address(self): if len(self['addresses']['data']) == 1: return self['addresses']['data'][0] else: for address in self['addresses']['data']: if (address['id'] != self['data']['attributes']['billingAddressId']): return address def get_billing_address(self): for address in self['addresses']['data']: if (address['id'] == self['data']['attributes']['billingAddressId']): return address def get_first_included_by_type(self, typestr): for inc in self['included']: if inc['type'] == typestr: return inc def get_tracking_codes(self): orderdelivery = self.get_first_included_by_type('order_delivery') if orderdelivery is not None: return orderdelivery['attributes']['trackingCodes'] else: return [] def print_summary(self): orderno = self['data']['attributes']['orderNumber'] firstname = self.get_billing_address()['attributes']['firstName'] lastname = self.get_billing_address()['attributes']['lastName'] print(f'{orderno}: {firstname} {lastname}') for li in self['lineItems']['data']: quantity = li['attributes']['quantity'] label = li['attributes']['label'] print(f'{quantity}x {label}') def add_patch(self, patchtype, patch): """add patch to be applied on save""" self.patches[patchtype].append(patch) def add_delete(self, deletetype, delete): """add deletion to be applied on save""" self.deletes[deletetype].append(delete) def add_create(self, createtype, create): """add creation to be applied on save""" self.creates[createtype].append(create) def set_delivery(self, trackingcodes): delivery = self.get_first_included_by_type('order_delivery') self.add_patch('order_delivery', { 'id': delivery['id'], 'trackingCodes': trackingcodes }) def save(self): # create new relations for createtype in self.creates: if createtype == 'order_delivery': for create in self.creates[createtype]: self.session.create_order_delivery(create) for patch in self.patches['order_delivery']: self.session.patch_order_delivery(self.order_delivery_id, patch) @property def order_delivery_id(self): orderdelivery = self.get_first_included_by_type('order_delivery') return orderdelivery['id'] @classmethod def get_by_orderid(cls, orderid, sws): orderfilter = cls.FILTER_ORDER_BY_IDS.copy() orderfilter['filter'][0]['value'] = orderid r = sws.post_json('search/order', orderfilter) json = r.json() raworders = json['data'] if raworders: entry = json.copy() entry['data'] = raworders[0] return cls(entry, session=sws) class ShopwareOrderList(list): FILTER_ORDER_PAIDOPEN = { "associations": { "deliveries": { "associations": { "stateMachineState": {} } }, "transactions": { "associations": { "stateMachineState": {} } } }, "filter": [{ 'type': 'multi', 'operator': 'and', 'queries': [ { 'type': 'equals', 'field': 'transactions.stateMachineState.technicalName', 'value': 'paid' }, { 'type': 'equals', 'field': 'deliveries.stateMachineState.technicalName', 'value': 'open' }, { 'type': 'equals', 'field': 'stateMachineState.technicalName', 'value': 'open' } ] }] } FILTER_ORDER_VORKASSEOPEN = { "associations": { "deliveries": { "associations": { "stateMachineState": {} } }, "transactions": { } }, "filter": [{ 'type': 'multi', 'operator': 'and', 'queries': [ { 'type': 'equalsAny', 'field': 'transactions.paymentMethodId', 'value': ['71fdafaca1e840df8686f8313bce4126', '018eeb96f34c72bf9b4d9ee8403d0f0b'] }, { 'type': 'equals', 'field': 'deliveries.stateMachineState.' 'technicalName', 'value': 'open' }, { 'type': 'equals', 'field': 'stateMachineState.technicalName', 'value': 'open' } ] }] } FILTER_ORDER_INPROGRESS = { "filter": [ { 'type': 'equals', 'field': 'stateMachineState.technicalName', 'value': 'in_progress' } ] } FILTER_ORDER_PREORDER = { "filter": [ { 'type': 'equals', 'field': 'stateMachineState.technicalName', 'value': 'preordered' } ] } FILTER_ORDER_OPEN_CANCELLED = { "associations": { "deliveries": { "associations": { "stateMachineState": {} } }, "transactions": { "associations": { "stateMachineState": {} } } }, "filter": [{ 'type': 'multi', 'operator': 'and', 'queries': [ { 'type': 'equals', 'field': 'transactions.stateMachineState.technicalName', 'value': 'cancelled' }, { 'type': 'equals', 'field': 'deliveries.stateMachineState.technicalName', 'value': 'open' }, { 'type': 'equals', 'field': 'stateMachineState.technicalName', 'value': 'open' } ] }] } FILTER_IMPORTABLE_ORDERS= { "associations": { "deliveries": { "associations": { "stateMachineState": {} } }, "transactions": { "associations": { "stateMachineState": {} } } }, "filter": [{ 'type': 'multi', 'operator': 'or', 'queries': [ { 'type': 'multi', 'operator': 'and', 'queries': [ { 'type': 'equals', 'field': 'transactions.stateMachineState.technicalName', 'value': 'paid' }, { 'type': 'equals', 'field': 'deliveries.stateMachineState.technicalName', 'value': 'open' }, { 'type': 'equals', 'field': 'stateMachineState.technicalName', 'value': 'open' } ] }, { 'type': 'multi', 'operator': 'and', 'queries': [ { 'type': 'equalsAny', 'field': 'transactions.paymentMethodId', 'value': ['71fdafaca1e840df8686f8313bce4126', '018eeb96f34c72bf9b4d9ee8403d0f0b'] }, { 'type': 'equals', 'field': 'transactions.stateMachineState.technicalName', 'value': 'open' }, { 'type': 'equalsAny', 'field': 'stateMachineState.technicalName', 'value': ['open', 'complete', 'in_progress'] } ] }, { 'type': 'multi', 'operator': 'and', 'queries': [ { 'type': 'equalsAny', 'field': 'transactions.paymentMethodId', 'value': ['71fdafaca1e840df8686f8313bce4126', '018eeb96f34c72bf9b4d9ee8403d0f0b'] }, { 'type': 'equals', 'field': 'deliveries.stateMachineState.technicalName', 'value': 'open' }, { 'type': 'equalsAny', 'field': 'stateMachineState.technicalName', 'value': ['open', 'complete', 'in_progress'] } ] }, ] }] } @classmethod def get_open_cancelled_orders(cls, sws): r = sws.post_json('search/order', cls.FILTER_ORDER_OPEN_CANCELLED) json = r.json() raworders = json['data'] ol = cls() for ro in raworders: entry = json.copy() entry['data'] = ro ol.append(ShopwareOrder(entry, prefetch=False, session=sws)) return ol @classmethod def get_paid_open_orders(cls, sws): r = sws.post_json('search/order', cls.FILTER_ORDER_PAIDOPEN) json = r.json() raworders = json['data'] ol = cls() for ro in raworders: entry = json.copy() entry['data'] = ro ol.append(ShopwareOrder(entry, prefetch=False, session=sws)) return ol @classmethod def get_open_vorkasse_orders(cls, sws): r = sws.post_json('search/order', cls.FILTER_ORDER_VORKASSEOPEN) json = r.json() raworders = json['data'] ol = cls() for ro in raworders: entry = json.copy() entry['data'] = ro ol.append(ShopwareOrder(entry, prefetch=False, session=sws)) return ol @classmethod def get_in_progress_orders(cls, sws): r = sws.post_json('search/order', cls.FILTER_ORDER_INPROGRESS) json = r.json() raworders = json['data'] ol = cls() for ro in raworders: entry = json.copy() entry['data'] = ro ol.append(ShopwareOrder(entry, prefetch=False, session=sws)) return ol @classmethod def get_preordered_orders(cls, sws): r = sws.post_json('search/order', cls.FILTER_ORDER_PREORDER) json = r.json() raworders = json['data'] ol = cls() for ro in raworders: entry = json.copy() entry['data'] = ro ol.append(ShopwareOrder(entry, prefetch=False, session=sws)) return ol @classmethod def get_syncable_ordernumbers(cls, sws): r = sws.post_json('search/order', cls.FILTER_IMPORTABLE_ORDERS) importable = [d['attributes']['orderNumber'] for d in r.json()['data']] r = sws.post_json('search/order', cls.FILTER_ORDER_PREORDER) preordered = [d['attributes']['orderNumber'] for d in r.json()['data']] r = sws.post_json('search/order', cls.FILTER_ORDER_INPROGRESS) inprogress = [d['attributes']['orderNumber'] for d in r.json()['data']] return importable + preordered + inprogress class ShopwareMedia(): """Shopware media entry i.e. images""" def __init__(self, raw=None, session=None): self.session = session self.raw = raw self.attr = raw['attributes'] self.id = raw['id'] @property def md5checksum(self): """The md5 checksum is part of the filename by convention The md5 sum should be the last 32 characters in the basename of the file. """ fn = self.attr['fileName'] if fn and len(fn) >= 32: return fn[-32:] @classmethod def list_from_product(cls, product, session=None): media = [] for pm in product.included['media']: media.append(cls(pm, session)) return media class ShopwareProductMedia(): """Relation between product and media """ def __init__(self, raw=None, session=None): self.session = session self.raw = raw self.attr = raw['attributes'] self.id = raw['id'] self.mediaid = self.attr['mediaId'] @classmethod def list_from_product(cls, product, session=None): productmedia = [] for pm in product.included['product_media']: productmedia.append(cls(pm, session)) return productmedia @property def position(self): return self.attr['position'] class ShopwareProduct(): """Representation of a product in shopware""" def __init__(self, raw=None, session=None, relations=[]): self.session = session self.raw = raw self.attr = raw['data'][0]['attributes'] self.data = raw['data'][0] self.relations = relations self.patches = defaultdict(list) self.deletes = defaultdict(list) self.creates = defaultdict(list) self.included = defaultdict(list) for rel in self.raw['included']: self.included[rel['type']].append(rel) self.media_by_id = dict( ((m.id, m) for m in ShopwareMedia.list_from_product(self, session)) ) self.product_media_by_id = dict( (pm.mediaid, pm) for pm in ShopwareProductMedia.list_from_product(self, session) ) self.media = sorted(self.media_by_id.values(), key=lambda x: self.product_media_by_id[x.id].position) def save(self): # create new relations for createtype in self.creates: if createtype == 'product_visibility': for create in self.creates[createtype]: self.session.create_product_visibility(create) if createtype == 'media': for create in self.creates[createtype]: self.session.create_media(create) if createtype == 'product_media': for create in self.creates[createtype]: self.session.create_product_media(create) if createtype == 'order_delivery': for create in self.creates[createtype]: self.session.create_order_delivery(create) # delete relations for deletetype in self.deletes: if deletetype == 'product_category': for delete in self.deletes[deletetype]: self.session.delete_product_category(self.id, delete) if deletetype == 'product_visibility': for delete in self.deletes[deletetype]: self.session.delete_product_visibility(delete) elif deletetype == 'product_media': for delete in self.deletes[deletetype]: self.session.delete_product_media(delete) elif deletetype == 'media': for delete in self.deletes[deletetype]: try: self.session.delete_media(delete) except HTTPError as e: warning(f"Deletion of media {delete} for product " f"{self.product_number} failed. Probably " f"still in use elsewhere.") # apply patches for patchtype in filter(lambda x: x != 'product', self.patches.keys()): raise ShopwareError(f"{patchtype} not implemented") if 'product' in self.patches: merged_patch = {} for patch in self.patches['product']: if 'customFields' in patch and 'customFields' in merged_patch: merged_patch['customFields'].update(patch['customFields']) else: merged_patch.update(patch) self.session.patch_product(self.id, merged_patch) def add_patch(self, patchtype, patch): """add patch to be applied on save""" self.patches[patchtype].append(patch) def add_delete(self, deletetype, delete): """add deletion to be applied on save""" self.deletes[deletetype].append(delete) def add_create(self, createtype, create): """add creation to be applied on save""" self.creates[createtype].append(create) def delete_media_by_id(self, mediaid): """Deletes product media relation and media itself""" self.add_delete('product_media', self.product_media_by_id[mediaid].id) self.add_delete('media', mediaid) def delete_product_visibility(self, entityid): return self.delete(f"product-visibility/{entityid}") def add_media(self, position, url, md5checksum, prefix): """upload and link media to this product""" mediaid = uuid4().hex self.add_create('media', {'id': mediaid, 'url': url, 'fileName': f"{prefix}{md5checksum}"}) productmediaid = uuid4().hex self.add_create('product_media', {'id': productmediaid, 'productId': self.id, 'mediaId': mediaid, 'position': position}) if position == 0: # always default to position 0 for cover self.add_patch('product', {'coverId': productmediaid}) def replace_media_by_id(self, mediaid, position, url, md5checksum, prefix): self.delete_media_by_id(mediaid) self.add_media(position, url, md5checksum, prefix) def set_active(self, status): p = { 'active': status } self.add_patch('product', p) def set_taxid(self, taxid): p = { 'taxId': taxid } self.add_patch('product', p) def set_closeout(self, status): p = { 'isCloseout': status } self.add_patch('product', p) def set_ean(self, ean): p = { 'ean': ean } self.add_patch('product', p) def set_name(self, name): p = { 'name': name } self.add_patch('product', p) def set_description(self, desc): p = { 'description': desc } self.add_patch('product', p) def set_gross_price(self, price, listprice=None, regulationprice=None): if listprice is None: listprice = price if regulationprice is None: regulationprice = price net = price / VATFACTORTABLE.get(self.attr['taxId'], 1.19) listpricenet = listprice / VATFACTORTABLE.get( self.attr['taxId'], 1.19) regulationpricenet = regulationprice / VATFACTORTABLE.get( self.attr['taxId'], 1.19) p = { 'price': [ { 'currencyId': CURRENCYIDTABLE['EUR'], 'gross': price, 'net': net, 'linked': True, 'listPrice': { 'currencyId': CURRENCYIDTABLE['EUR'], 'gross': listprice, 'net': listpricenet, 'linked': True, }, 'regulationPrice': { 'currencyId': CURRENCYIDTABLE['EUR'], 'gross': regulationprice, 'net': regulationpricenet, 'linked': True, } } ] } self.add_patch('product', p) def set_dimensions(self, dimensions): p = { 'length': dimensions[0], 'width': dimensions[1], 'height': dimensions[2], 'weight': dimensions[3] } self.add_patch('product', p) def set_manufacturer_by_name(self, name): man = self.session.search_manufacturer(name) if man['data']: manufacturerid = man['data'][0]['id'] else: raise ShopwareError(f"Manufacturer {name} not found.") p = { 'manufacturerId': manufacturerid } self.add_patch('product', p) def set_eu_responsible_person(self, person): p = { 'customFields': {'custom_fromplenty_extra_euresponsible': person} } self.add_patch('product', p) def set_releasedate(self, isodatestr): p = { 'releaseDate': isodatestr } self.add_patch('product', p) def set_channels(self, channels): deletechannels = self.active_channels - channels for visibility in self.included['product_visibility']: if visibility['attributes']['salesChannelId'] in deletechannels: self.add_delete('product_visibility', visibility['id']) addchannels = channels - self.active_channels uuids = [] for channel in addchannels: uuid = uuid4().hex uuids.append(uuid) c = { 'id': uuid, 'productId': self.id, 'salesChannelId': channel, 'visibility': 30 } self.add_create('product_visibility', c) def set_availability(self, index): p = { 'deliveryTimeId': AVAILABILITIES[index] } self.add_patch('product', p) def set_free_shipping(self, value): p = { 'shippingFree': value } self.add_patch('product', p) def set_categories(self, catids): if self.categories: for c in self.categories: self.add_delete('product_category', c) p = { 'categories': [{'id': c} for c in catids] } self.add_patch('product', p) def set_warnings(self, text): p = { 'customFields': {'custom_fromplenty_extra_warnings': text} } self.add_patch('product', p) def set_features(self, features): # sanetize content features = [f.replace(';', '|') for f in features] p = { 'customFields': {'zenit_atmos_features': ';'.join(features)} } self.add_patch('product', p) @property def categories(self): if self.attr['categoryIds']: return set(self.attr['categoryIds']) else: return set() @property def free_shipping(self): return self.attr['shippingFree'] @property def description(self): return self.attr['description'] @property def dimensions(self): return (self.attr['length'], self.attr['width'], self.attr['height'], self.attr['weight']) @property def manufacturername(self): man = self.included['product_manufacturer'] if man: return man[0]['attributes']['name'] @property def eu_responsible_person(self): if self.attr['customFields']: return self.attr['customFields'].get( 'custom_fromplenty_extra_euresponsible', None) @property def features(self): if self.attr['customFields']: f = self.attr['customFields'].get('zenit_atmos_features', '') return [] if f is None else f.split(';') @property def warnings(self): if self.attr['customFields']: return self.attr['customFields'].get( 'custom_fromplenty_extra_warnings', None) @property def ean(self): return self.attr['ean'] @property def releasedate(self): return self.attr['releaseDate'] @property def availability(self): if self.attr['deliveryTimeId']: return AVAILABILITIESREVERSE[self.attr['deliveryTimeId']] @property def is_closeout(self): return self.attr['isCloseout'] @property def active_channels(self): channels = set() for e in self.included['product_visibility']: channels.add(e['attributes']['salesChannelId']) return channels @property def gross_price(self): return self.attr['price'][0]['gross'] @property def list_price(self): if self.attr['price'][0]['listPrice']: return self.attr['price'][0]['listPrice']['gross'] @property def product_number(self): return self.attr['productNumber'] @property def name(self): return self.attr['name'] @property def id(self): return self.data['id'] @property def is_variation(self): return bool(self.attr['parentId']) @property def has_variations(self): return bool(self.attr['childCount']) @classmethod def get_product_by_number(cls, session, prodno, relationships=[]): if isinstance(relationships, str) and relationships == 'all': relationships = SHOPWARE_PRODUCT_RELATIONSHIPS prod = session.get_product_by_number(prodno, relationships) if prod.json()['data']: return cls(prod.json(), session, relationships) class StockSync(): def __init__(self, csvstock, swstock, ignorelist): self.csvstock = csvstock self.swstock = {} self.swstockoriginal = {} self.swuuid = {} self.ignorelist = ignorelist for prod in swstock: attrs = prod['attributes'] self.swstock[attrs['productNumber']] = attrs['stock'] self.swstockoriginal[attrs['productNumber']] = prod self.swuuid[attrs['productNumber']] = prod['id'] def showdiff(self): allstock = self.csvstock.copy() allstock.update(self.swstock) for number in sorted(allstock.keys()): csvstockqty = self.csvstock.get(number, 'NA') swstockqty = self.swstock.get(number, 'NA') if str(csvstockqty) != str(swstockqty): if str(swstockqty) == "NA" or swstockqty > 0: if csvstockqty == "NA" or csvstockqty < 10: print(f'{number}: CSV={csvstockqty} SW={swstockqty}') def updatestock(self, sws, prodno, newstock): r = sws.patch_product(self.swuuid[prodno], {'stock': newstock}) if r.ok: info(f"SW stock for {prodno} updated to {newstock}") else: error(f"SW stock for {prodno} NOT updated. Server reply " f"'{r.reason}'") def sync(self, sws): csvstock = self.csvstock swstock = self.swstock for prodno in sorted(set(csvstock.keys()) | set(swstock.keys())): if prodno in self.ignorelist: continue if prodno in swstock: attrs = self.swstockoriginal[prodno]['attributes'] if (not attrs['isCloseout'] and attrs['active']): warning(f"{prodno} has 'Abverkauf' deactivated but is " f"active. Unlimited sales are possible regardless " f"of stock") if prodno in swstock and prodno in csvstock: swqty = swstock[prodno] csvqty = csvstock[prodno] if (csvqty > 0 and not attrs['active']): warning(f"CSV stock for {prodno} positive ({csvqty}) but " f"item is not active on SW") if csvstock[prodno] != swstock[prodno]: info(f"updating SW stock for {prodno} from {swqty} to " f"{csvqty}") self.updatestock(sws, prodno, csvqty) elif prodno in csvstock and prodno not in swstock: if csvstock[prodno] > 0: warning(f"CSV contains postive stock for {prodno} but no " f"matching product can be found in SW") else: if attrs['active'] and swstock[prodno] > 0: warning(f"{prodno} not in CSV. Oversale possible.") class CSVSource: def __init__(self, csvfilenames): self.rows = [] if isinstance(csvfilenames, str): csvfilenames = [csvfilenames] for csvfilename in csvfilenames: if not os.path.exists(csvfilename): # assume its a url debug(f'downloading {csvfilename}') csvfilename, headers = urllib.request.urlretrieve(csvfilename) reader = csv.DictReader(open(csvfilename, newline=''), delimiter=';') for row in reader: self.rows.append(row) class StockCSVgomazing(CSVSource): def get_available_stock(self): swqty = defaultdict(int) for row in self.rows: vid = row['Variation.id'] qty = (int(row['VariationStock.netStock.Halle'] or '0') + int(row['VariationStock.netStock.Vorverkaufslager'] or '0')) variationid = int(vid) swqty[f"GM{variationid:05}"] = qty return swqty def action_inspect(args, sws): # products = sws.list_products() r = sws.get_product_by_wmnumber('GM10021') import code code.interact(local=dict(globals(), **locals())) def action_liststockcsv(args, sws): searchfilter = {'includes': {'product': ['id', 'name', 'productNumber', 'stock', 'active', 'isCloseout', 'shippingFree']}, 'sort': [{'field': 'productNumber', 'order': 'ASC'}]} print('productNumber;name;stock;active;isCloseout;shippingFree') for p in sws.list_products(searchfilter)['data']: a = p['attributes'] print(f'{a["productNumber"]};{a["name"]};{a["stock"]};{a["active"]};' f'{a["isCloseout"]};{a["shippingFree"]}') def action_syncstock(args, sws): if args.target == 'gomazing': stockcsv = StockCSVgomazing(args.stockcsv) else: warning(f"Target {args.target} not implemented.") stockqty = stockcsv.get_available_stock() swstock = sws.list_stock() sync = StockSync(stockqty, swstock, args.ignore) sync.sync(sws) TARGETS = { 'gomazing': { 'uri': 'https://shop.gomazing.de/api/', 'user': 'SWIASTBMV2ZKNWNJZ3RHZNLUDA', 'secret': 'UlN6MFRTN1BxT0p5WkdYVVVsTGtZcWFnZTdLMDVTVlVZeHFHUUI' } } if __name__ == '__main__': parser = argparse.ArgumentParser(description='Shopware6 REST ' 'interface') parser.add_argument('target', choices=TARGETS.keys(), help='shopware target instance.') parser.add_argument('--logfile', default=None, help='file for logging') parser.add_argument('-q', '--quiet', action='store_true', help='no output except for warnings and worse') parser.add_argument('--failonwarnings', action='store_true', help='exit with error status if any warnings present') parser.add_argument('-v', '--verbose', action='store_const', const='DEBUG', default='INFO', help='display more details') subparsers = parser.add_subparsers(required=True) syncstockparser = subparsers.add_parser( 'syncstock', help='update shopware stock from provided CSVs') syncstockparser.add_argument( 'stockcsv', nargs='+', help='local csv file(s) or link(s) with stock details') syncstockparser.add_argument( '--ignore', nargs='+', default=[], help='Shopware product numbers to explicitly ignore') syncstockparser.set_defaults(func=action_syncstock) inspectparser = subparsers.add_parser( 'inspect', help='interactive prompt to probe and inspect api') inspectparser.set_defaults(func=action_inspect) liststockcsvparser = subparsers.add_parser( 'liststockcsv', help='list products in CSV form') liststockcsvparser.set_defaults(func=action_liststockcsv) args = parser.parse_args() LOGFORMAT = '%(asctime)s %(levelname)s %(name)s %(message)s' if args.verbose == 'DEBUG': logging.basicConfig(filename=args.logfile, level=logging.DEBUG, format=LOGFORMAT) elif args.verbose == 'INFO' and not args.quiet: logging.basicConfig(filename=args.logfile, level=logging.INFO, format=LOGFORMAT) elif args.verbose == 'INFO' and args.quiet: logging.basicConfig(filename=args.logfile, level=logging.WARNING, format=LOGFORMAT) # call main action t = TARGETS[args.target] with ShopwareSession(t['uri'], t['user'], t['secret']) as sws: args.func(args, sws) if warnings_present and args.failonwarnings: sys.exit(1)