Creates treatment class for Vtex API products and adds methos to priv…

…ate vtex service

marketplace/clients/facebook/client.py

@@ -54,7 +54,7 @@ def create_product_feed(self, product_catalog_id, name):
 
         return response.json()
 
-    def upload_product_feed(self, feed_id, file):
+    def upload_product_feed(self, feed_id, file, update_only=False):
         url = self.get_url + f"{feed_id}/uploads"
 
         headers = self._get_headers()
@@ -65,7 +65,10 @@ def upload_product_feed(self, feed_id, file):
                 file.content_type,
             )
         }
-        response = self.make_request(url, method="POST", headers=headers, files=files)
+        params = {"update_only": update_only}
+        response = self.make_request(
+            url, method="POST", headers=headers, params=params, files=files
+        )
         return response.json()
 
     def create_product_feed_via_url(

marketplace/clients/vtex/client.py

@@ -34,38 +34,67 @@ def search_product_by_sku_id(self, skuid, domain, sellerid=1):
 
 
 class VtexPrivateClient(VtexAuthorization, VtexCommonClient):
-    def get_products_sku_ids(self, domain):
+    def is_valid_credentials(self, domain):
+        try:
+            url = (
+                f"https://{domain}/api/catalog_system/pvt/products/GetProductAndSkuIds"
+            )
+            headers = self._get_headers()
+            response = self.make_request(url, method="GET", headers=headers)
+            return response.status_code == 200
+        except Exception:
+            return False
+
+    def list_all_products_sku_ids(self, domain, page_size=1000):
         all_skus = []
-        page_size = 250
-        _from = 1
-        _to = page_size
+        page = 1
 
         while True:
-            url = f"https://{domain}/api/catalog_system/pvt/products/GetProductAndSkuIds?_from={_from}&_to={_to}"
+            url = f"https://{domain}/api/catalog_system/pvt/sku/stockkeepingunitids?page={page}&pagesize={page_size}"
             headers = self._get_headers()
             response = self.make_request(url, method="GET", headers=headers)
 
-            data = response.json().get("data")
-            if data:
-                for sku_ids in data.values():
-                    all_skus.extend(sku_ids)
-
-                total_products = response.json().get("range", {}).get("total", 0)
-                if _to >= total_products:
-                    break
-                _from += page_size
-                _to += page_size
-                _to = min(_to, total_products)  # To avoid overshooting the total count
-            else:
+            sku_ids = response.json()
+            if not sku_ids:
                 break
 
+            all_skus.extend(sku_ids)
+            page += 1
+
         return all_skus
 
-    def is_valid_credentials(self, domain):
-        try:
-            url = f"https://{domain}/api/catalog_system/pvt/products/GetProductAndSkuIds"
-            headers = self._get_headers()
-            response = self.make_request(url, method="GET", headers=headers)
-            return response.status_code == 200
-        except Exception:
-            return False
+    def list_active_sellers(self, domain):
+        url = f"https://{domain}/api/seller-register/pvt/sellers"
+        headers = self._get_headers()
+        response = self.make_request(url, method="GET", headers=headers)
+        sellers_data = response.json()
+        return [seller["id"] for seller in sellers_data["items"] if seller["isActive"]]
+
+    def get_product_details(self, sku_id, domain):
+        url = (
+            f"https://{domain}/api/catalog_system/pvt/sku/stockkeepingunitbyid/{sku_id}"
+        )
+        headers = self._get_headers()
+        response = self.make_request(url, method="GET", headers=headers)
+        return response.json()
+
+    def pub_simulate_cart_for_seller(self, sku_id, seller_id, domain):
+        cart_simulation_url = f"https://{domain}/api/checkout/pub/orderForms/simulation"
+        payload = {"items": [{"id": sku_id, "quantity": 1, "seller": seller_id}]}
+
+        response = self.make_request(cart_simulation_url, method="POST", json=payload)
+        simulation_data = response.json()
+
+        if simulation_data["items"]:
+            item_data = simulation_data["items"][0]
+            return {
+                "is_available": item_data["availability"] == "available",
+                "price": item_data["price"],
+                "list_price": item_data["listPrice"],
+            }
+        else:
+            return {
+                "is_available": False,
+                "price": 0,
+                "list_price": 0,
+            }

marketplace/services/vtex/private/products/service.py

@@ -1,44 +1,45 @@
 """
-Service for interacting with VTEX private APIs that require authentication.
+Service for managing product operations with VTEX private APIs.
 
-This service is responsible for validating domain and credentials against VTEX private APIs.
-It encapsulates the logic for domain validation and credentials checking, ensuring that only
-valid and authenticated requests are processed for private VTEX operations.
+This service interacts with VTEX's private APIs for product-related operations. It handles
+domain validation, credentials verification, product listing, and updates from webhook notifications.
 
 Attributes:
-    client: A configured client instance that is capable of communicating with VTEX private APIs.
+    client: A client instance for VTEX private APIs communication.
+    data_processor: DataProcessor instance for processing product data.
 
 Public Methods:
-    check_is_valid_domain(domain): Validates the provided domain to ensure it is recognized by VTEX.
-        Raises a CredentialsValidationError if the domain is not valid.
-
-    validate_private_credentials(domain): Validates the credentials stored in the client for the given domain.
-        Returns True if the credentials are valid, False otherwise.
-
-Private Methods:
-    _is_domain_valid(domain): Performs a check against the VTEX API to determine if the provided domain is valid.
+    check_is_valid_domain(domain): Validates if a domain is recognized by VTEX.
+    validate_private_credentials(domain): Checks if stored credentials for a domain are valid.
+    list_all_products(domain): Lists all products from a domain. Returns processed product data.
+    get_product_details(sku_id, domain): Retrieves details for a specific SKU.
+    simulate_cart_for_seller(sku_id, seller_id, domain): Simulates a cart for a seller and SKU.
+    update_product_info(domain, webhook_payload): Updates product info based on webhook payload.
 
 Exceptions:
-    CredentialsValidationError: Raised when the provided domain
-    or credentials are not valid according to VTEX's standards.
+    CredentialsValidationError: Raised for invalid domain or credentials.
 
 Usage:
-    To use this service, instantiate it with a client that has the necessary API credentials (app_key and app_token).
-    The client should implement methods for checking domain validity and credentials.
+    Instantiate with a client having API credentials. Use methods for product operations with VTEX.
 
 Example:
-    client = VtexPrivateClient(app_key="your-app-key", app_token="your-app-token")
+    client = VtexPrivateClient(app_key="key", app_token="token")
     service = PrivateProductsService(client)
-    is_valid = service.validate_private_credentials("your-domain.vtex.com")
+    is_valid = service.validate_private_credentials("domain.vtex.com")
     if is_valid:
-        # Proceed with operations that require valid credentials
+        products = service.list_all_products("domain.vtex.com")
+        # Use products data as needed
 """
 from marketplace.services.vtex.exceptions import CredentialsValidationError
+from marketplace.services.vtex.utils.data_processor import DataProcessor
 
 
 class PrivateProductsService:
-    def __init__(self, client):
+    def __init__(self, client, data_processor=DataProcessor):
         self.client = client
+        self.data_processor = data_processor
+        # TODO: Check if it makes sense to leave the domain instantiated
+        # so that the domain parameter is removed from the methods
 
     # ================================
     # Public Methods
@@ -54,6 +55,40 @@ def validate_private_credentials(self, domain):
         self.check_is_valid_domain(domain)
         return self.client.is_valid_credentials(domain)
 
+    def list_all_products(self, domain):
+        active_sellers = self.client.list_active_sellers(domain)
+        skus_ids = self.client.list_all_products_sku_ids(domain)
+
+        data = self.data_processor.process_product_data(
+            skus_ids, active_sellers, self, domain
+        )
+        return data
+
+    def get_product_details(self, sku_id, domain):
+        return self.client.get_product_details(sku_id, domain)
+
+    def simulate_cart_for_seller(self, sku_id, seller_id, domain):
+        return self.client.pub_simulate_cart_for_seller(
+            sku_id, seller_id, domain
+        )  # TODO: Change to pvt_simulate_cart_for_seller
+
+    def update_product_info(self, domain, webhook_payload):
+        updated_products = []
+
+        sku_id = webhook_payload["IdSku"]
+        price_modified = webhook_payload["PriceModified"]
+        stock_modified = webhook_payload["StockModified"]
+        other_changes = webhook_payload["HasStockKeepingUnitModified"]
+
+        seller_ids = self.client.list_active_sellers(domain)
+
+        if price_modified or stock_modified or other_changes:
+            updated_products = self.data_processor.process_product_data(
+                [sku_id], seller_ids, self, domain, update_product=True
+            )
+
+        return updated_products
+
     # ================================
     # Private Methods
     # ================================

marketplace/services/vtex/utils/data_processor.py

@@ -0,0 +1,157 @@
+from dataclasses import dataclass
+import csv
+import io
+from typing import List
+import dataclasses
+
+
+@dataclass
+class FacebookProductDTO:
+    id: str
+    title: str
+    description: str
+    availability: str
+    condition: str
+    price: str
+    link: str
+    image_link: str
+    brand: str
+    sale_price: str
+    product_details: dict
+
+    def get_multiplier(self):
+        return self.product_details.get("UnitMultiplier", 1.0)
+
+    def get_weight(self):
+        return self.product_details["Dimension"]["weight"]
+
+    def calculates_by_weight(self):
+        return self.product_details["MeasurementUnit"] != "un"
+
+
+@dataclass
+class VtexProductDTO:  # TODO: Implement This VtexProductDTO
+    pass
+
+
+class DataProcessor:
+    SEPARATOR = "#"
+    CURRENCY = "BRL"
+
+    @staticmethod
+    def create_unique_product_id(sku_id, seller_id):
+        return f"{sku_id}{DataProcessor.SEPARATOR}{seller_id}"
+
+    @staticmethod
+    def decode_unique_product_id(unique_product_id):
+        sku_id, seller_id = unique_product_id.split(DataProcessor.SEPARATOR)
+        return sku_id, seller_id
+
+    @staticmethod
+    def extract_fields(product_details, availability_details) -> FacebookProductDTO:
+        price = (
+            availability_details["price"]
+            if availability_details["price"] is not None
+            else 0
+        )
+        list_price = (
+            availability_details["list_price"]
+            if availability_details["list_price"] is not None
+            else 0
+        )
+        return FacebookProductDTO(
+            id=product_details["Id"],
+            title=product_details["SkuName"],
+            description=product_details["SkuName"],
+            availability="in stock"
+            if availability_details["is_available"]
+            else "out of stock",
+            condition="new",
+            price=list_price,
+            link="",  # URL do produto (a ser preenchida)
+            image_link=product_details["ImageUrl"],
+            brand=product_details.get("BrandName", "N/A"),
+            sale_price=price,
+            product_details=product_details,
+        )
+
+    @staticmethod
+    def format_price(price):
+        """Formats the price to the standard 'XX.XX BRL'."""
+        formatted_price = f"{price / 100:.2f} {DataProcessor.CURRENCY}"  # TODO: Move CURRENCY to business layer
+        return formatted_price
+
+    @staticmethod
+    def format_fields(
+        seller_id, product: FacebookProductDTO
+    ):  # TODO: Move this method rules to business layer
+        if product.calculates_by_weight():
+            # Apply price calculation logic per weight/unit
+            DataProcessor.calculate_price_by_weight(product)
+
+        # Format the price for all products
+        product.price = DataProcessor.format_price(product.price)
+        product.sale_price = DataProcessor.format_price(product.sale_price)
+        product.id = DataProcessor.create_unique_product_id(product.id, seller_id)
+
+        return product
+
+    @staticmethod
+    def calculate_price_by_weight(product: FacebookProductDTO):
+        unit_multiplier = product.get_multiplier()
+        product_weight = product.get_weight()
+        weight = product_weight * unit_multiplier
+        product.price = product.price * unit_multiplier
+        product.sale_price = product.sale_price * unit_multiplier
+        product.description += f" - {weight}g"
+
+    @staticmethod
+    def process_product_data(
+        skus_ids, active_sellers, service, domain, update_product=False
+    ):
+        facebook_products = []
+        for sku_id in skus_ids:
+            product_details = service.get_product_details(sku_id, domain)
+
+            for seller_id in active_sellers:
+                availability_details = service.simulate_cart_for_seller(
+                    sku_id, seller_id, domain
+                )
+
+                if update_product is False:
+                    if not availability_details["is_available"]:
+                        continue
+
+                extracted_product_dto = DataProcessor.extract_fields(
+                    product_details, availability_details
+                )
+                formatted_product_dto = DataProcessor.format_fields(
+                    seller_id, extracted_product_dto
+                )
+                facebook_products.append(formatted_product_dto)
+
+        return facebook_products
+
+    def products_to_csv(products: List[FacebookProductDTO]) -> str:
+        output = io.StringIO()
+        fieldnames = [
+            field.name
+            for field in dataclasses.fields(FacebookProductDTO)
+            if field.name != "product_details"
+        ]
+        writer = csv.DictWriter(output, fieldnames=fieldnames)
+        writer.writeheader()
+        for product in products:
+            row = dataclasses.asdict(product)
+            row.pop(
+                "product_details", None
+            )  # TODO: should change this logic before going to production
+            writer.writerow(row)
+
+        return output.getvalue()
+
+    @staticmethod
+    def generate_csv_file(csv_content: str) -> io.BytesIO:
+        csv_bytes = csv_content.encode("utf-8")
+        csv_memory = io.BytesIO(csv_bytes)
+        return csv_memory