feat: extract ct data into csv command

commerce_coordinator/apps/commercetools/clients.py

@@ -5,7 +5,7 @@
 import datetime
 import logging
 from types import SimpleNamespace
-from typing import Generic, List, Optional, Tuple, TypedDict, TypeVar, Union
+from typing import Dict, Generic, List, Optional, Tuple, TypedDict, TypeVar, Union
 
 import requests
 from commercetools import Client as CTClient
@@ -42,6 +42,7 @@
 from commercetools.platform.models.state import State as CTLineItemState
 from django.conf import settings
 from openedx_filters.exceptions import OpenEdxFilterException
+from requests.exceptions import HTTPError
 
 from commerce_coordinator.apps.commercetools.catalog_info.constants import (
     DEFAULT_ORDER_EXPANSION,
@@ -832,3 +833,73 @@ def is_first_time_discount_eligible(self, email: str, code: str) -> bool:
                                        err, f"Unable to check if user {email} is eligible for a "
                                             f"first time discount", True)
             return True
+
+
+class CTCustomAPIClient:
+    """Custom Commercetools API Client using requests."""
+
+    def __init__(self):
+        """
+        Initialize the Commercetools client with configuration from Django settings.
+        """
+        self.config = settings.COMMERCETOOLS_CONFIG
+        self.access_token = self._get_access_token()
+
+    def _get_access_token(self) -> str:
+        """
+        Retrieve an access token using client credentials flow for Commercetools.
+
+        Returns:
+            str: Access token for API requests.
+        """
+        auth_url = self.config["authUrl"]+'/oauth/token'
+        auth = (self.config["clientId"], self.config["clientSecret"])
+        data = {
+            "grant_type": "client_credentials",
+            "scope": self.config['scopes'],
+        }
+
+        response = requests.post(auth_url, auth=auth, data=data)
+
+        response.raise_for_status()
+        return response.json()["access_token"]
+
+    def _make_request(
+            self,
+            method: str,
+            endpoint: str,
+            params: Optional[Dict] = None,
+            json: Optional[Dict] = None,
+    ) -> Union[Dict, List]:
+        """
+        Make an HTTP request to the Commercetools API.
+
+        Args:
+            method (str): HTTP method (e.g., "GET", "POST").
+            endpoint (str): API endpoint (e.g., "/cart-discounts").
+            params (Optional[Dict]): Query parameters.
+            json (Optional[Dict]): JSON payload for POST/PUT requests.
+
+        Returns:
+            Union[Dict, List]: JSON response from the API or None if the request fails.
+        """
+        url = f"{self.config['apiUrl']}/{self.config['projectKey']}/{endpoint}"
+        headers = {
+            "Authorization": f"Bearer {self.access_token}",
+            "Content-Type": "application/json",
+        }
+        try:
+            response = requests.request(method, url, headers=headers, params=params, json=json)
+            response.raise_for_status()
+            return response.json()
+        except HTTPError as err:
+            if response is not None:
+                response_message = response.json().get('message', 'No message provided.')
+                logger.error(
+                    "API request for endpoint: %s failed with error: %s and message: %s",
+                    endpoint, err, response_message
+                )
+            else:
+                logger.error("API request for endpoint: %s failed with error: %s", endpoint, err)
+
+            return None

commerce_coordinator/apps/commercetools/management/commands/create_commercetools_product.py

@@ -0,0 +1,65 @@
+import json
+from commercetools.platform.models import (
+    ProductDraft,
+    ProductVariantDraft,
+    ProductTypeResourceIdentifier,
+    ProductPriceModeEnum
+)
+from commerce_coordinator.apps.commercetools.management.commands._ct_api_client_command import (
+    CommercetoolsAPIClientCommand,
+)
+
+
+product_json=''' 
+
+
+
+'''
+
+class Command(CommercetoolsAPIClientCommand):
+    help = "Create a commercetools product from a JSON string or file"
+
+    def handle(self, *args, **options):
+        if product_json:
+            try:
+                product_data = json.loads(product_json)
+            except json.JSONDecodeError as e:
+                self.stderr.write(f"Invalid JSON format: {e}")
+                return
+        else:
+            print("\n\n\n\nNo JSON data provided.\n\n\n\n")
+            return
+
+        product_type_data = product_data.get("productType")
+        if product_type_data:
+            product_type = ProductTypeResourceIdentifier(id=product_type_data["id"])
+        else:
+            print("\n\n\n\nMissing productType data.\n\n\n\n")
+            return
+
+        master_variant_data = product_data.get("masterVariant")
+        variants_data = product_data.get("variants", [])
+
+        master_variant = ProductVariantDraft(**master_variant_data)
+        variants = [ProductVariantDraft(**variant) for variant in variants_data]
+
+        product_draft_data = {
+            "key": product_data.get("key"),
+            "name": product_data.get("name"),
+            "description": product_data.get("description"),
+            "slug": product_data.get("slug"),
+            "price_mode": ProductPriceModeEnum.STANDALONE,
+            "publish": product_data.get("publish"),
+            "tax_category": product_data.get("taxCategory"),
+            "master_variant": master_variant,
+            "variants": variants,
+            "product_type": product_type
+        }
+
+        try:
+            product_draft = ProductDraft(**product_draft_data)
+            created_product = self.ct_api_client.base_client.products.create(draft=product_draft)
+            print(f"\n\n\n\nSuccessfully created product with ID: {created_product.id}")
+        except Exception as e:
+            print(f"\n\n\n\nError creating product: {e}")
+

commerce_coordinator/apps/commercetools/management/commands/extract_commercetools_data_to_csv.py

@@ -0,0 +1,139 @@
+from datetime import datetime
+import csv
+from enum import Enum
+from itertools import product
+
+from commerce_coordinator.apps.commercetools.management.commands._ct_api_client_command import (
+    CommercetoolsAPIClientCommand
+)
+
+
+# Enum for product types
+class ProductType(Enum):
+    EDX_COURSE_ENTITLEMENT = "edx_course_entitlement"
+    EDX_PROGRAM = "edx_program"
+    OC_SELF_PACED = "oc_self_paced"
+    EDX_COURSE = "oc_self_paced"
+
+STAGE_PRODUCT_TYPE_ID_MAPPING = {
+    ProductType.EDX_COURSE_ENTITLEMENT.value: "12e5510c-a4d6-4301-9caf-17053e57ff71",
+    ProductType.EDX_PROGRAM.value: "79fb6abe-8373-4dec-a8d1-51242b1798b8",
+    ProductType.OC_SELF_PACED.value: "9f8ec882-043a-4225-8811-00ac5acfd580"
+}
+
+PROD_PRODUCT_TYPE_ID_MAPPING = {
+    ProductType.EDX_COURSE_ENTITLEMENT.value: "9f1f189a-4d79-4eaa-9c6e-cfcb61aa779f",
+    ProductType.EDX_PROGRAM.value: "c6a2d629-a50e-4d88-bd01-ab05a0617eae",
+    ProductType.EDX_COURSE.value: "b241ac79-fee2-461d-b714-8f3c4a1c4c0e"
+}
+
+class Command(CommercetoolsAPIClientCommand):
+    help = "Fetch and verify course attributes from CommerceTools"
+
+    def handle(self, *args, **options):
+        # Specify product type to fetch
+        product_type = ProductType.EDX_PROGRAM
+
+        # Fetch products based on type
+        products = self.fetch_products(product_type)
+
+        # Write data to CSV
+        self.write_attributes_to_csv(products, product_type)
+
+    def fetch_products(self, product_type):
+        limit = 500
+        offset = 0
+        products = []
+
+        product_type_id = PROD_PRODUCT_TYPE_ID_MAPPING.get(product_type.value)
+
+        while True:
+            products_result = self.ct_api_client.base_client.products.query(
+                limit=limit,
+                offset=offset,
+                where=f"productType(id=\"{product_type_id}\")"
+            )
+            for product in products_result.results:
+                attributes = self.extract_product_attributes(product, product_type.value)
+                products.extend(attributes)
+
+            if products_result.offset + products_result.limit >= products_result.total:
+                break
+            offset += limit
+
+        return products
+
+    def extract_product_attributes(self, product, product_type):
+        # Extract common product-level attributes
+        common_attributes = {
+            "product_type": product_type,
+            "product_id": product.id,
+            "product_key": product.key,
+            "published_status": product.master_data.published,
+            "name": product.master_data.current.name.get('en-US', ''),
+            "slug": product.master_data.current.slug.get('en-US', ''),
+            "description": (
+                product.master_data.current.description.get('en-US', '')
+                if product.master_data.current.description
+                else ''
+            ),
+            "date_created": product.created_at,
+            "master_variant_key": product.master_data.current.master_variant.key,
+            "master_variant_sku": product.master_data.current.master_variant.sku,
+            "master_variant_image_url": (
+                product.master_data.current.master_variant.images[0].url
+                if product.master_data.current.master_variant.images
+                else None
+            ),
+        }
+
+        product_rows = []  # This will hold the product and variant rows
+
+        # Add the master variant attributes
+        if len(product.master_data.current.variants) == 0:
+            master_variant_attributes = {attr.name: attr.value for attr in
+                                         product.master_data.current.master_variant.attributes}
+            product_rows.append({**common_attributes, **master_variant_attributes})
+
+        # Add attributes for each variant and create a separate row, including variant_key and variant_sku
+        for variant in product.master_data.current.variants:
+            variant_attributes = {attr.name: attr.value for attr in variant.attributes}
+            variant_row = {
+                **common_attributes,
+                "variant_key": variant.key,  # Add variant_key
+                "variant_sku": variant.sku,  # Add variant_sku
+                "variant_image_url": (
+                    variant.images[0].url
+                    if variant.images
+                    else None
+                ),
+                **variant_attributes,
+            }
+            # Create a new row for each variant, combining common product data with variant-specific attributes
+            product_rows.append(variant_row)
+
+        return product_rows
+
+    def write_attributes_to_csv(self, products, product_type):
+        if not products:
+            print(f"No products found for type {product_type}.")
+            return
+
+        # Dynamically extract all unique keys across all product dictionaries
+        keys = set()
+        for product in products:
+            keys.update(product.keys())
+
+        # Convert keys set back to a list and sort them if you want a consistent order
+        keys = sorted(list(keys))
+
+        # Define CSV filename with product type and date
+        filename = f"{product_type.value}_attributes_{datetime.now().strftime('%Y%m%d')}.csv"
+
+        # Write to CSV
+        with open(filename, "w", newline="") as output_file:
+            dict_writer = csv.DictWriter(output_file, fieldnames=keys)
+            dict_writer.writeheader()
+            dict_writer.writerows(products)
+
+        print(f"\n\n\n\n\n\n\nCSV file '{filename}' written successfully with {len(products)} records.")

commerce_coordinator/apps/commercetools/management/commands/update_commercetools_product.py

@@ -0,0 +1,67 @@
+import json
+from commerce_coordinator.apps.commercetools.management.commands._ct_api_client_command import CommercetoolsAPIClientCommand
+from commerce_coordinator.apps.commercetools.clients import CTCustomAPIClient
+
+product_id = '21533075-1710-4cbc-88d2-a18ba4176fdd'
+product_json = ''' 
+
+{
+        "version": 9,
+        "actions": [
+            {
+              "action": "setDescription",
+              "description": {
+                "en-US": "<p>This course reflects the most current version of the PMP exam, based on the Project Management Institute, Inc's (PMI) current exam content outline (ECO). This will not only prepare you for the PMP, but also teach you valuable project management skills useful for program managers and project managers in this PMP training course.</p><p>Covers PMBOK Guide 7th Edition, PMBOK Guide 6th Edition, Agile Practice Guide, and more. Passing the PMP certification exam is an important step for any individual with project management experience looking to advance their career in the field and wants to earn this valuable credential.  This course covers content related to predictive (traditional), adaptive (Agile), and hybrid projects. Each course allows you to earn up to 10 professional development units (PDUs). When you enroll as a verified learner, you'll have immediate access to study materials for your PMP exam prep.</p><p>This course is taught by Instructor and global keynote speaker, Crystal Richards.  Crystal is an experienced project manager and has over 20 years of experience working with project teams in the private-sector and public-sector. Crystal's client work has been in the healthcare and federal government. She specializes in project management training and is a PMI PMP and PMI-ACP credential holder. Crystal has taught foundational project management courses and project management certification boot camp courses to thousands of students around the world both in the classroom and online.</p><p>The entire course will include the following:<ul><li>Earn 35 PDUs/Contact Hours by completing the entire course as required by PMI</li><li>Content based on the current PMP Examination Content Outline</li><li>Expert guidance completing the PMP application to meet exam eligibility requirements</li><li>Explanation of the project management processes</li><li>Discussion of key project management topics such as scope management, cost management,</li> schedule management, and risk management</li><li>Demonstrate use of key formulas, charts, and graphs</li><li>Strong foundation in Agile project management such as scrum, XP, and Kanban</li><li>Exposure to challenging exam questions on practice exams-including 'wordy' questions, questions with formulas, and questions with more than one correct answer</li><li>Guidance on the logistical details to sit for the exam such as information on the exam fee for PMI members and non-members, paying for PMI membership, prerequisites, and information on test centers.</li></ul></p><p>The course is broken up into 4 modules:<ul><li>PMP Prep: Project Management Principles - This module will provide an overview of predictive, Agile, and hybrid project management methodologies.  The module will also delve into key project roles, and key concepts such as tailoring, progressive elaboration, and rolling wave planning.</li><li>PMP Prep: Managing People with Power Skills - Linked to the Leadership skill area of the PMI Talent Triangle®, this module will place focus on managing the expectations and relationships of the people involved in projects. Participants will need to demonstrate the knowledge, skills and behaviors to guide, motivate and/or direct others to achieve a goal. Key skills related to people include planning resource needs, managing stakeholder expectations, and communications planning and execution.  This module will also delve into 'power skills' such as negotiations, active listening, emotional intelligence, and servant leadership.</li><li>PMP Prep: Determining Ways of Working for Technical Project Management - Linked to the Technical skill area of the PMI Talent Triangle®, this module focuses on the technical aspects of successfully managing projects.  Topics will delve into the core skills of scope, cost, and schedule management and integrating these concepts to develop a master project plan.  Participants will also need to demonstrate an understanding of quality, risk, and procurement management and use techniques such as earned value, critical path methodology, and general data gathering and analysis techniques.</li><li>PMP Prep: Gaining Business Acumen for Project Managers- Linked to the Strategic and Business Management skill area of the PMI Talent Triangle®, this module will highlight the connection between projects and organizational strategy.  Participants will need to demonstrate knowledge of and expertise in the industry/organization, so as to align the project goals and objectives to the organizational goals and enhance performance to better deliver business outcomes.  Additional topics in this module will include compliance management and an understanding of how internal and external factors impact project outcomes.</li></ul></p>"
+              }
+            },
+            {
+                "action": "publish"
+            }
+        ]
+    }
+
+
+'''
+
+class Command(CommercetoolsAPIClientCommand):
+    help = "Update a commercetools product from a JSON string or file"
+
+    def handle(self, *args, **options):
+        if product_json:
+            try:
+                product_data = json.loads(product_json)
+            except json.JSONDecodeError as e:
+                self.stderr.write(f"Invalid JSON format: {e}")
+                return
+        else:
+            print("\n\n\n\nNo JSON data provided.\n\n\n\n")
+            return
+
+        version = product_data.get("version")
+        actions = product_data.get("actions")
+
+        if not product_id or not version or not actions:
+            print("\n\n\n\nMissing product ID, version, or actions.\n\n\n\n")
+            return
+
+        # Initialize the custom commercetools client
+        ct_client = CTCustomAPIClient()
+
+        # Prepare the data for updating the product
+        update_payload = {
+            "version": version,
+            "actions": actions
+        }
+
+        # Make the request to update the product
+        endpoint = f"products/{product_id}"
+        response = ct_client._make_request(
+            method="POST",
+            endpoint=endpoint,
+            json=update_payload
+        )
+
+        if response:
+            print(f"\n\n\n\n\nSuccessfully updated product with ID: {response.get('id')}\n\n\n\n\n")
+        else:
+            print("\n\n\n\n\nError updating product.\n\n\n\n\n\n")