add cosmohome client with exact export of combinations and products

src/telecaster/Clients/BaseClient.py

@@ -2,12 +2,12 @@ import requests
 
 
 class BaseClient:
-    def get(self, url):
-        response = requests.get(url)
+    def get(self, url, timeout: int = 5):
+        response = requests.get(url, timeout=timeout)
         if response and response.status_code == 200:
             return response.content
         else:
-            print('Something went wrong')
+            print(response.text)
             return response.status_code
 
     def post(self, url, body):

src/telecaster/Clients/CosmohomeClient.py

@@ -0,0 +1,204 @@
+import hashlib
+import struct
+from collections import defaultdict
+
+from telecaster.Clients.PrestaShopClient import PrestaShopClient
+
+CANONICAL_DOMAIN = "https://cosmohome.gr"
+TOKEN = "IK2GQGGKH4LUVYR9C2TSF6XC51F1J1C3"
+
+COLOR_PRODUCT_OPTION_IDS = ['5', '17', '11']
+SIZE_PRODUCT_OPTION_IDS = ['1', '13', '16', '3']
+
+TAX = 0.24
+
+
+def numHash(s, length=None):
+    hash_object = hashlib.md5(s.encode('ISO-8859-1')).digest()
+    hash_str, _ = struct.unpack('>II', hash_object[:8])
+    if length and isinstance(length, int):
+        hash_str = str(hash_str)[:length]
+    return str(hash_str)
+
+
+def get_nested(data, map_list):
+    for k in map_list:
+        data = data.get(k, None)
+        if data is None:
+            return None
+    return data
+
+
+class CosmohomeClient(PrestaShopClient):
+    def __init__(self, base_url=None, token=None):
+        super().__init__(base_url or CANONICAL_DOMAIN, token or TOKEN)
+        self.caches = defaultdict(dict)
+
+    def update_cache(self, cache_key, values: list, id_key_path):
+        cache_dict = self.caches[cache_key]
+        for item in values:
+            key = get_nested(item, id_key_path)
+            cache_dict[key] = item
+
+    def load_cache(self):
+        categories = self.get_categories()  # {'display': '[id,name, link_rewrite, id_parent]'})
+        features = self.get_features()
+        feature_values = self.get_feature_values()
+        product_options = self.get_product_options()
+        product_option_values = self.get_product_option_values()
+        # tax_rule_groups = self.get_tax_rule_groups()
+        # tax_rules = self.get_tax_rules()
+        # taxes = self.get_taxes()
+
+        categories = map(lambda x: x['category'], categories)
+        features = map(lambda x: x['product_feature'], features)
+        feature_values = map(lambda x: x['product_feature_value'], feature_values)
+        product_options = map(lambda x: x['product_option'], product_options)
+        product_option_values = map(lambda x: x['product_option_value'],
+                                    product_option_values)
+        # tax_rule_groups = map(lambda x: x['tax_rule_group'], tax_rule_groups)
+        # tax_rules = map(lambda x: x['tax_rule'], tax_rules)
+        # taxes = map(lambda x: x['tax'], taxes)
+
+        self.update_cache('categories', categories, ['id'])
+        self.update_cache('features', features, ['id'])
+        self.update_cache('feature_values', feature_values, ['id'])
+        self.update_cache('product_options', product_options, ['id'])
+        self.update_cache('product_option_values', product_option_values, ['id'])
+        # self.update_cache('tax_rule_groups', tax_rule_groups, ['id'])
+        # self.update_cache('tax_rules', tax_rules, ['id'])
+        # self.update_cache('taxes', taxes, ['id'])
+
+    def parse_product(self, product, combinations):
+        product = product.get('product')
+        product_combinations = self.get_product_combinations(product, combinations)
+
+        # Move this to pre-computed cache when perform batches
+        id_product_attributes = [c.get('id') for c in product_combinations]
+        sp = self.get_specific_prices(
+            {'filter[id_product_attribute]': f"{'|'.join(id_product_attributes)}"})
+        sp = list(map(lambda x: x['specific_price'], sp))
+        self.update_cache('specific_prices', sp, ['id_product_attribute'])
+        self.update_cache('specific_prices', sp, ['id_product'])
+        ## END OF Specific prices calculations
+
+        serialized_product = {
+            "productId": product.get('id'),
+            "title": self.sanitize_language(product.get('name')),
+            "productURL": product.get('link_rewrite'),
+            # add to the url the attributes like #/attr_id/attr_id/...
+            "Category": self.get_category_tree(product.get('id_category_default'),
+                                               min_depth=1),
+            "Category_ID": product.get('id_category_default'),
+            "weight": product.get('weight'),
+            "Manufacturer": product.get('manufacturer_name'),
+            "MPN": product.get('reference') or product.get('mpn'),
+            "Barcode": product.get('ean13'),
+            "Price": self.calculate_final_price(product),
+            "image": self.generate_image_url(product),
+            "stock": "Y" if int(product.get('quantity')) > 0 else "N",
+            "Availability": "Available from 1 to 3 days" if int(
+                product.get('quantity')) > 0 else "",
+            "Quantity": product.get('quantity'),
+            "Color": self.get_color(product),
+            "Size": self.get_size(product),
+        }
+
+        serialized_combinations = []
+        for combination in product_combinations:
+            serialized_combinations.append({
+                **serialized_product,
+                "productId": product.get('id') + numHash(combination.get('reference'), 10),
+                "Barcode": combination.get('ean13') or serialized_product.get(
+                    'Barcode'),
+                "Name": combination.get('name'),
+                "weight": product.get('weight'),
+                "MPN": combination.get('reference') or combination.get(
+                    'mpn') or serialized_product.get('MPN'),
+
+                "Price": self.calculate_final_price(
+                    combination,
+                    float(product.get('price')) + float(combination.get('price'))
+                ),
+                "Quantity": combination.get('quantity') or serialized_product.get(
+                    'Quantity'),
+
+                "Color": self.get_color(product) or serialized_product.get('Color'),
+                "Size": self.get_size(product) or serialized_product.get('Size'),
+            })
+
+        return [serialized_product] if not combinations else serialized_combinations
+
+    def sanitize_language(self, value):
+        if isinstance(value, str):
+            return value.strip()
+        elif isinstance(value, dict):
+            return list(value.values())[0]
+        else:
+            return value
+
+    def get_category_tree(self, category_id, min_depth=0):
+        categories = self.caches['categories']
+        category = categories[category_id]
+        category_tree = []
+        while category and int(category.get('level_depth')) > min_depth:
+            if category:
+                category_tree.insert(0, category['name'])
+                category = categories.get(category.get('id_parent'))
+            else:
+                break
+
+        return [self.sanitize_language(c) for c in category_tree]
+
+    def generate_image_url(self, product):
+        image_id = product.get('id_default_image')
+        link_rewrite = self.sanitize_language(product.get('link_rewrite'))
+        return f"{CANONICAL_DOMAIN}/{image_id}-thickbox_default/{link_rewrite}.jpg"
+
+    def get_color(self, product):
+        options = product.get('associations').get('product_option_values')
+        for option in options:
+            option = option.get('product_option_value')
+            option_value = self.caches['product_option_values'].get(option.get('id'))
+            if option_value and option_value.get(
+                    'id_attribute_group') in COLOR_PRODUCT_OPTION_IDS:
+                return self.sanitize_language(option_value.get('name'))
+        return None
+
+    def get_size(self, product):
+        options = product.get('associations').get('product_option_values')
+        for option in options:
+            option = option.get('product_option_value')
+            option_value = self.caches['product_option_values'].get(option.get('id'))
+            if option_value and option_value.get(
+                    'id_attribute_group') in SIZE_PRODUCT_OPTION_IDS:
+                return self.sanitize_language(option_value.get('name'))
+        return None
+
+    def get_product_combinations(self, product, combinations):
+        if combinations:
+            combinations = map(lambda x: x['combination'], combinations)
+            self.update_cache('combinations', combinations, ['id'])
+
+        combinations = self.caches['combinations']
+
+        product_combinations = product.get('associations').get('combinations')
+        product_combination_ids = map(lambda x: x['combination']['id'],
+                                      product_combinations)
+
+        return [combinations.get(cid) for cid in product_combination_ids]
+
+    def calculate_final_price(self, product, price=None):
+        if not price:
+            price = product.get('price')
+
+        price_with_tax = float(price) * (1 + TAX)
+
+        sp = self.caches['specific_prices'].get(product.get('id'))
+        discount = sp.get('reduction')
+        type = sp.get('reduction_type')
+        if type == 'percentage':
+            final_price_with_tax = price_with_tax * (1 - float(discount))
+        else:
+            final_price_with_tax = price_with_tax - float(discount)
+        return round(final_price_with_tax, 2)

src/telecaster/Clients/PrestaShopClient.py

@@ -9,14 +9,22 @@ class PrestaShopClient(BaseClient):
         self.token = token
         self.base_url = base_url
 
-    def build_url(self, url, params={}):
-        updated_params = {**params, 'ws_key': self.token}
+    def build_url(self, url, params=None):
+        updated_params = {**(params or {}), 'ws_key': self.token}
         return f'{self.base_url}/api{url}?{urllib.parse.urlencode(updated_params)}'
 
-    def get_products(self, params={}):
+    def request(self, url, params=None, timeout=20):
+        constructed_params = {'display': 'full', **(params or {})}
+        full_url = self.build_url(url, constructed_params)
+        xml_content = self.get(full_url, timeout=timeout)
+        xml = ElementTree.fromstring(xml_content)
+        result = [parse_xml_to_json(child) for child in list(xml)[0]]
+        return result
+
+    def get_products(self, params=None):
         constructed_params = {
-            'display': '[name, id, mpn, id_default_image, id_category_default,price,wholesale_price,manufacturer_name,ean13,delivery_in_stock,additional_shipping_cost,description,quantity, link_rewrite]',
-            **params,
+            'display': 'full',
+            **(params or {}),
         }
         full_url = self.build_url('/products', constructed_params)
         # we need to parse these since the response is in XML
@@ -29,16 +37,34 @@ class PrestaShopClient(BaseClient):
 
         return result
 
-    def get_categories(self, params={}):
-        constructed_params = {'display': '[id,name, link_rewrite]', **params}
-        full_url = self.build_url('/categories', constructed_params)
-        xml_categories_content = self.get(full_url)
-        xml_categories = ElementTree.fromstring(xml_categories_content)
-        categories = list(xml_categories)[0]
-        # result = {}
-        # for child in categories:
-        #     child_ojb = parse_xml_to_json(child)
-        #     result[child_ojb['id']] = child_ojb
-        # Alternatively if we wanna convert to Array check use the below
-        result = [parse_xml_to_json(child) for child in categories]
-        return result
+    def get_categories(self, params=None):
+        return self.request('/categories', params)
+
+    def get_features(self, params=None):
+        return self.request('/product_features', params)
+    def get_feature_values(self, params=None):
+        return self.request('/product_feature_values', params)
+
+    def get_combinations(self, params=None):
+        return self.request('/combinations', params)
+
+    def get_product_options(self, params=None):
+        return self.request('/product_options', params)
+
+    def get_product_option_values(self, params=None):
+        return self.request('/product_option_values', params)
+
+    def get_images(self, params=None):
+        return self.request('/images', params)
+
+    def get_tax_rule_groups(self, params=None):
+        return self.request('/tax_rule_groups', params)
+
+    def get_tax_rules(self, params=None):
+        return self.request('/tax_rules', params)
+
+    def get_taxes(self, params=None):
+        return self.request('/taxes', params)
+
+    def get_specific_prices(self, params=None):
+        return self.request('/specific_prices', params)

src/telecaster/parsers/parse_xml_to_json.py

@@ -1,13 +1,15 @@
 def parse_xml_to_json(xml):
-    response = {}
-    for child in list(xml):
-        if len(list(child)) > 0:
-            response[child.tag] = parse_xml_to_json(child)
-        else:
-            response[child.tag] = child.text or ''
-
-    return response
+    if not len(list(xml)):
+        if xml.attrib.get('id'):
+            return {xml.attrib.get('id'): xml.text}
+        return {xml.tag: xml.text}
 
+    parsed_children = [parse_xml_to_json(child) for child in list(xml)]
+    if xml.attrib.get('nodeType'):
+        return {xml.tag: parsed_children}
+    else:
+        merged_children = {k: v for d in parsed_children for k, v in d.items()}
+        return {xml.tag: merged_children}
 
 def parse_prestashop_xml_products(xml):
     response = {}

src/telecaster/tests/test_cosmohome_xml.py

@@ -7,12 +7,8 @@ def test_post_method(django_db_setup):
     # Create a client to make requests
     client = APIClient()
 
-    # Define your test url and token
-    test_url = 'http://test.com'
-    test_token = 'test_token'
-
     # Make a post request to your view
-    response = client.post(reverse('xml-generator-view'), {'url': test_url, 'token': test_token})  # Replace 'xml_generator_view' with your actual view name
+    response = client.get(reverse('cosmohome-generate-xml'))  # Replace 'xml_generator_view' with your actual view name
 
     # Assert the status code is 200
     assert response.status_code == 200

src/telecaster/urls.py

@@ -17,11 +17,13 @@ from django.contrib import admin
 from django.urls import path, re_path
 from rest_framework import routers
 from .views import XmlGeneratorView
-
+from .views.CosmohomeView import CosmohomeView
 
 router = routers.SimpleRouter(trailing_slash=False)
 
+router.register(r'cosmohome', CosmohomeView, basename='cosmohome')
+
 urlpatterns = [
     re_path(r'^generate/xml', view=XmlGeneratorView.as_view(), name='xml-generator-view'),
     path('admin/', admin.site.urls),
-]
+] + router.urls

src/telecaster/views/CosmohomeView.py

@@ -0,0 +1,60 @@
+from rest_framework import views, viewsets
+from rest_framework.decorators import action
+from rest_framework.request import Request
+from rest_framework.response import Response
+from rest_framework_xml.renderers import XMLRenderer
+from django.http import JsonResponse
+
+from ..Clients.CosmohomeClient import CosmohomeClient
+from ..Clients.PrestaShopClient import PrestaShopClient
+from ..serializers import XmlGeneratorSerializer
+from ..sanitizers.prestashop_backoffice_api_sanitizer import PrestashopBackOfficeApiSanitizer
+
+
+class CosmohomeView(viewsets.ViewSet):
+    renderer_classes = [
+        XMLRenderer,
+    ]
+
+    @classmethod
+    @action(methods=['get'], detail=False, url_path='generate/xml')
+    def generate_xml(cls, request: Request, *args, **kwargs) -> JsonResponse:
+        client = CosmohomeClient()
+        client.load_cache()
+
+        products = client.get_products({'limit': 5, 'filter[active]': 1, 'filter[id]': 1615})
+        combinations = client.get_combinations({
+            'filter[id_product]': f"[{'|'.join([p.get('product').get('id') for p in products])}]"
+        })
+
+        client.parse_product(products[0], combinations)
+
+        # retrieve categories first since they are needed for products
+        # categories = prestashop_client.get_categories({
+        #     'display': '[id,name, link_rewrite]'
+        # })
+        # features = prestashop_client.get_features()
+        # feature_values = prestashop_client.get_feature_values()
+        # products = prestashop_client.get_products({'limit': 5, 'filter[active]': 1, 'filter[id]': 1615})
+        # combinations = prestashop_client.get_combinations({
+        #     'filter[id_product]': f"[{'|'.join([p.get('product').get('id') for p in products])}]"
+        # })
+        #
+        # product_options = prestashop_client.get_product_options()
+        # product_option_values = prestashop_client.get_product_option_values()
+
+        sanitizer = PrestashopBackOfficeApiSanitizer()
+
+        # sanitized_categories = sanitizer.categories(categories)
+        #
+        # categories_dict = {category.category_id: vars(category) for category in sanitized_categories}
+        # sanitized_products = sanitizer.products(products)
+        #
+        # # serialize response
+        # serializer = XmlGeneratorSerializer()
+        # response_xml = serializer.for_xml(
+        #     products=sanitized_products, categories=categories_dict, site_url=url
+        # )
+
+        # return Response(response_xml, content_type='text/xml')
+        return Response("ok")

src/telecaster/views/XmlGeneratorView.py

@@ -25,8 +25,19 @@ class XmlGeneratorView(views.APIView):
         prestashop_client = PrestaShopClient(base_url=url, token=token)
 
         # retrieve categories first since they are needed for products
-        categories = prestashop_client.get_categories()
-        products = prestashop_client.get_products({'limit': 5})
+        categories = prestashop_client.get_categories({
+            'display': '[id,name, link_rewrite]'
+        })
+        features = prestashop_client.get_features()
+        feature_values = prestashop_client.get_feature_values()
+        products = prestashop_client.get_products({'limit': 5, 'filter[active]': 1, 'filter[id]': 1615})
+        combinations = prestashop_client.get_combinations({
+            'filter[id_product]': f"[{'|'.join([p.get('product').get('id') for p in products])}]"
+        })
+
+        product_options = prestashop_client.get_product_options()
+        product_option_values = prestashop_client.get_product_option_values()
+
         sanitizer = PrestashopBackOfficeApiSanitizer()
 
         sanitized_categories = sanitizer.categories(categories)