diff --git a/marketplace/clients/facebook/client.py b/marketplace/clients/facebook/client.py index 01215f57..40b59e64 100644 --- a/marketplace/clients/facebook/client.py +++ b/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( diff --git a/marketplace/clients/vtex/client.py b/marketplace/clients/vtex/client.py index dad8e10e..c1bb926f 100644 --- a/marketplace/clients/vtex/client.py +++ b/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, + } diff --git a/marketplace/services/vtex/private/products/service.py b/marketplace/services/vtex/private/products/service.py index 6a2cc3c3..d71574b9 100644 --- a/marketplace/services/vtex/private/products/service.py +++ b/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 # ================================ diff --git a/marketplace/services/vtex/utils/__init__.py b/marketplace/services/vtex/utils/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/marketplace/services/vtex/utils/data_processor.py b/marketplace/services/vtex/utils/data_processor.py new file mode 100644 index 00000000..d10635af --- /dev/null +++ b/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