Wpp-cloud products catalog cart

.coveragerc

@@ -9,7 +9,9 @@ omit =
     marketplace/flows/*
     marketplace/connect/*
     marketplace/wpp_templates/requests.py
+    marketplace/core/types/channels/whatsapp_cloud/services/*
+    marketplace/clients/*
     *urls.py*
 
 [report]
-fail_under = 70
+fail_under = 75

marketplace/clients/base.py

@@ -0,0 +1,41 @@
+import requests
+
+from marketplace.clients.exceptions import CustomAPIException
+
+
+class RequestClient:
+    def make_request(
+        self,
+        url: str,
+        method: str,
+        headers=None,
+        data=None,
+        params=None,
+        files=None,
+        json=None,
+    ):
+        if data and json:
+            raise ValueError(
+                "Cannot use both 'data' and 'json' arguments simultaneously."
+            )
+
+        response = requests.request(
+            method=method,
+            url=url,
+            headers=headers,
+            json=json,
+            data=data,
+            timeout=60,
+            params=params,
+            files=files,
+        )
+        if response.status_code >= 500:
+            raise CustomAPIException(status_code=response.status_code)
+        elif response.status_code >= 400:
+            try:
+                detail = response.json()
+            except ValueError:
+                detail = response.text
+            raise CustomAPIException(detail=detail, status_code=response.status_code)
+
+        return response

marketplace/clients/exceptions.py

@@ -0,0 +1,7 @@
+from rest_framework.exceptions import APIException
+
+
+class CustomAPIException(APIException):
+    def __init__(self, detail=None, code=None, status_code=None):
+        super().__init__(detail, code)
+        self.status_code = status_code or self.status_code

marketplace/clients/facebook/client.py

@@ -0,0 +1,217 @@
+import time
+import json
+
+from django.conf import settings
+
+from marketplace.clients.base import RequestClient
+
+WHATSAPP_VERSION = settings.WHATSAPP_VERSION
+ACCESS_TOKEN = settings.WHATSAPP_SYSTEM_USER_ACCESS_TOKEN
+
+
+class FacebookAuthorization:
+    BASE_URL = f"https://graph.facebook.com/{WHATSAPP_VERSION}/"
+
+    def __init__(self):
+        self.access_token = ACCESS_TOKEN
+
+    def _get_headers(self):
+        headers = {"Authorization": f"Bearer {self.access_token}"}
+        return headers
+
+    @property
+    def get_url(self):
+        return self.BASE_URL
+
+
+class FacebookClient(FacebookAuthorization, RequestClient):
+    def create_catalog(self, business_id, name, category=None):
+        url = self.get_url + f"{business_id}/owned_product_catalogs"
+        data = {"name": name}
+        if category:
+            data["vertical"] = category
+
+        headers = self._get_headers()
+        response = self.make_request(url, method="POST", headers=headers, data=data)
+
+        return response.json()
+
+    def destroy_catalog(self, catalog_id):
+        url = self.get_url + f"{catalog_id}"
+
+        headers = self._get_headers()
+        response = self.make_request(url, method="DELETE", headers=headers)
+
+        return response.json().get("success")
+
+    def create_product_feed(self, product_catalog_id, name):
+        url = self.get_url + f"{product_catalog_id}/product_feeds"
+
+        data = {"name": name}
+        headers = self._get_headers()
+        response = self.make_request(url, method="POST", headers=headers, data=data)
+
+        return response.json()
+
+    def upload_product_feed(self, feed_id, file):
+        url = self.get_url + f"{feed_id}/uploads"
+
+        headers = self._get_headers()
+        files = {
+            "file": (
+                file.name,
+                file,
+                file.content_type,
+            )
+        }
+        response = self.make_request(url, method="POST", headers=headers, files=files)
+        return response.json()
+
+    def create_product_feed_via_url(
+        self, product_catalog_id, name, feed_url, file_type, interval, hour
+    ):  # TODO: adjust this method
+        url = self.get_url + f"{product_catalog_id}/product_feeds"
+
+        schedule = {"interval": interval, "url": feed_url, "hour": str(hour)}
+
+        data = {"name": name, "schedule": json.dumps(schedule), "file_type": file_type}
+
+        headers = self._get_headers()
+        response = self.make_request(url, method="POST", headers=headers, data=data)
+        return response.json()
+
+    def get_upload_status(self, feed_id, max_attempts=10, wait_time=30):
+        """
+        Checks the upload status using long polling.
+
+        Args:
+            upload_id (str): The ID of the upload.
+            max_attempts (int): Maximum number of polling attempts. Default is 10.
+            wait_time (int): Wait time in seconds between polling attempts. Default is 30 seconds.
+
+        Returns:
+            bool or str: True if 'end_time' is found, otherwise a formatted error message.
+        """
+        url = self.get_url + f"{feed_id}/uploads"
+        headers = self._get_headers()
+
+        attempts = 0
+        while attempts < max_attempts:
+            response = self.make_request(url, method="GET", headers=headers)
+            data = response.json()
+
+            if data.get("data") and data["data"][0].get("end_time"):
+                return True
+
+            time.sleep(wait_time)
+            attempts += 1
+
+        total_wait_time = wait_time * max_attempts
+        return (
+            f"Unable to retrieve the upload completion status for feed {feed_id}. "
+            f"Waited for a total of {total_wait_time} seconds."
+        )
+
+    def list_products_by_feed(self, feed_id):
+        url = self.get_url + f"{feed_id}/products"
+
+        headers = self._get_headers()
+        response = self.make_request(url, method="GET", headers=headers)
+
+        return response.json()
+
+    def list_all_products_by_feed(self, feed_id):
+        url = self.get_url + f"{feed_id}/products"
+        headers = self._get_headers()
+        all_products = []
+
+        while url:
+            response = self.make_request(url, method="GET", headers=headers).json()
+            all_products.extend(response.get("data", []))
+            url = response.get("paging", {}).get("next")
+
+        return all_products
+
+    def list_all_catalogs(self, wa_business_id):
+        url = self.get_url + f"{wa_business_id}/owned_product_catalogs"
+        headers = self._get_headers()
+        all_catalog_ids = []
+
+        while url:
+            response = self.make_request(url, method="GET", headers=headers).json()
+            catalog_data = response.get("data", [])
+            all_catalog_ids.extend([item["id"] for item in catalog_data])
+            url = response.get("paging", {}).get("next")
+
+        return all_catalog_ids
+
+    def destroy_feed(self, feed_id):
+        url = self.get_url + f"{feed_id}"
+
+        headers = self._get_headers()
+        response = self.make_request(url, method="DELETE", headers=headers)
+
+        return response.json().get("success")
+
+    def get_connected_catalog(self, waba_id):
+        url = self.get_url + f"{waba_id}/product_catalogs"
+        headers = self._get_headers()
+        response = self.make_request(url, method="GET", headers=headers)
+        return response.json()
+
+    def enable_catalog(self, waba_id, catalog_id):
+        url = self.get_url + f"{waba_id}/product_catalogs"
+        data = {"catalog_id": catalog_id}
+        headers = self._get_headers()
+        response = self.make_request(url, method="POST", headers=headers, data=data)
+        return response.json()
+
+    def disable_catalog(self, waba_id, catalog_id):
+        url = self.get_url + f"{waba_id}/product_catalogs"
+        data = {"catalog_id": catalog_id, "method": "delete"}
+        headers = self._get_headers()
+        response = self.make_request(url, method="POST", headers=headers, data=data)
+        return response.json()
+
+    def get_catalog_details(self, catalog_id):
+        url = self.get_url + f"{catalog_id}"
+        params = {"fields": "name,vertical"}
+        headers = self._get_headers()
+        response = self.make_request(url, method="GET", headers=headers, params=params)
+
+        return response.json()
+
+    def _update_commerce_settings(self, wa_phone_number_id, **settings):
+        url = self.BASE_URL + f"{wa_phone_number_id}/whatsapp_commerce_settings"
+        headers = self._get_headers()
+        response = self.make_request(url, method="POST", headers=headers, data=settings)
+        return response.json()
+
+    def toggle_cart(self, wa_phone_number_id, enable=True):
+        return self._update_commerce_settings(
+            wa_phone_number_id, is_cart_enabled=enable
+        )
+
+    def toggle_catalog_visibility(self, wa_phone_number_id, make_visible=True):
+        return self._update_commerce_settings(
+            wa_phone_number_id, is_catalog_visible=make_visible
+        )
+
+    def get_wpp_commerce_settings(self, wa_phone_number_id):
+        """
+        Returns:
+            "data": [
+                {
+                    "is_cart_enabled": true,
+                    "is_catalog_visible": true,
+                    "id": "270925148880242"
+                }
+            ]
+        Or:
+            "data": []
+        """
+        url = self.BASE_URL + f"{wa_phone_number_id}/whatsapp_commerce_settings"
+
+        headers = self._get_headers()
+        response = self.make_request(url, method="GET", headers=headers)
+        return response.json()

marketplace/clients/flows/client.py

@@ -0,0 +1,54 @@
+"""Client for connection with flows"""
+
+
+from django.conf import settings
+from marketplace.clients.base import RequestClient
+
+
+class InternalAuthentication(RequestClient):
+    def __get_module_token(self):
+        data = {
+            "client_id": settings.OIDC_RP_CLIENT_ID,
+            "client_secret": settings.OIDC_RP_CLIENT_SECRET,
+            "grant_type": "client_credentials",
+        }
+        request = self.make_request(
+            url=settings.OIDC_OP_TOKEN_ENDPOINT, method="POST", data=data
+        )
+
+        token = request.json().get("access_token")
+
+        return f"Bearer {token}"
+
+    @property
+    def headers(self):
+        return {
+            "Content-Type": "application/json; charset: utf-8",
+            "Authorization": self.__get_module_token(),
+        }
+
+
+class FlowsClient(RequestClient):
+    def __init__(self):
+        self.base_url = settings.FLOWS_REST_ENDPOINT
+        self.authentication_instance = InternalAuthentication()
+
+    def detail_channel(self, flow_object_uuid):
+        url = f"{self.base_url}/api/v2/internals/channel/{str(flow_object_uuid)}"
+
+        response = self.make_request(
+            url, method="GET", headers=self.authentication_instance.headers
+        )
+        return response.json()
+
+    def update_config(self, data, flow_object_uuid):
+        payload = {"config": data}
+        url = f"{self.base_url}/api/v2/internals/channel/{flow_object_uuid}/"
+
+        response = self.make_request(
+            url,
+            method="PATCH",
+            headers=self.authentication_instance.headers,
+            json=payload,
+        )
+        return response

marketplace/core/types/channels/whatsapp_cloud/catalogs/tests/test_catalog_model.py

@@ -0,0 +1,55 @@
+import uuid
+
+from django.test import TestCase
+from django.contrib.auth import get_user_model
+from django.core.exceptions import ValidationError
+
+from marketplace.applications.models import App
+from marketplace.wpp_products.models import Catalog
+
+
+User = get_user_model()
+
+
+class CreateCatalogErrorTestCase(TestCase):
+    def setUp(self):
+        self.user = User.objects.create_superuser(email="user@marketplace.ai")
+
+    def test_create_catalog_with_error(self):
+        app = App.objects.create(
+            code="wpp",
+            created_by=self.user,
+            project_uuid=str(uuid.uuid4()),
+            platform=App.PLATFORM_WENI_FLOWS,
+        )
+        with self.assertRaises(ValidationError) as context:
+            Catalog.objects.create(
+                app=app,
+                facebook_catalog_id="0123456789",
+                name="catalog test error",
+                category="commerce",
+            )
+
+        expected_error_message = "The App must be a 'WhatsApp Cloud' AppType."
+        self.assertIn(expected_error_message, str(context.exception))
+
+
+class CreateCatalogSuccessTestCase(TestCase):
+    def setUp(self):
+        self.user = User.objects.create_superuser(email="user@marketplace.ai")
+
+    def test_create_catalog_success(self):
+        app = App.objects.create(
+            code="wpp-cloud",
+            created_by=self.user,
+            project_uuid=str(uuid.uuid4()),
+            platform=App.PLATFORM_WENI_FLOWS,
+        )
+
+        catalog = Catalog.objects.create(
+            app=app,
+            facebook_catalog_id="0123456789",
+            name="catalog test success",
+            category="commerce",
+        )
+        self.assertEqual(app.uuid, catalog.app.uuid)