stage 4-1
This commit is contained in:
alorig
@@ -1,10 +1,11 @@
|
||||
"""
|
||||
WordPress Integration Service
|
||||
Handles communication with WordPress sites via REST API
|
||||
Stage 4: Enhanced with taxonomy and WooCommerce product support
|
||||
"""
|
||||
import logging
|
||||
import requests
|
||||
from typing import Dict, Any, Optional
|
||||
from typing import Dict, Any, Optional, List
|
||||
from django.conf import settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -26,6 +27,7 @@ class WordPressClient:
|
||||
"""
|
||||
self.site_url = site_url.rstrip('/')
|
||||
self.api_base = f"{self.site_url}/wp-json/wp/v2"
|
||||
self.woocommerce_api_base = f"{self.site_url}/wp-json/wc/v3" # WooCommerce API
|
||||
self.igny8_api_base = f"{self.site_url}/wp-json/igny8/v1" # Custom IGNY8 endpoints
|
||||
self.username = username
|
||||
self.app_password = app_password
|
||||
@@ -215,4 +217,377 @@ class WordPressClient:
|
||||
'message': None,
|
||||
'error': str(e),
|
||||
}
|
||||
|
||||
# Stage 4: Taxonomy Methods
|
||||
|
||||
def get_categories(self, per_page: int = 100, page: int = 1) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Get WordPress categories.
|
||||
|
||||
Args:
|
||||
per_page: Number of categories per page
|
||||
page: Page number
|
||||
|
||||
Returns:
|
||||
List of category dictionaries
|
||||
"""
|
||||
try:
|
||||
response = self.session.get(
|
||||
f"{self.api_base}/categories",
|
||||
params={'per_page': per_page, 'page': page}
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
categories = response.json()
|
||||
return [
|
||||
{
|
||||
'id': cat.get('id'),
|
||||
'name': cat.get('name', ''),
|
||||
'slug': cat.get('slug', ''),
|
||||
'description': cat.get('description', ''),
|
||||
'count': cat.get('count', 0),
|
||||
'parent': cat.get('parent', 0),
|
||||
'link': cat.get('link', '')
|
||||
}
|
||||
for cat in categories
|
||||
]
|
||||
logger.warning(f"Failed to fetch categories: HTTP {response.status_code}")
|
||||
return []
|
||||
except Exception as e:
|
||||
logger.error(f"Error fetching WordPress categories: {e}")
|
||||
return []
|
||||
|
||||
def create_category(
|
||||
self,
|
||||
name: str,
|
||||
slug: Optional[str] = None,
|
||||
description: Optional[str] = None,
|
||||
parent_id: Optional[int] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Create a WordPress category.
|
||||
|
||||
Args:
|
||||
name: Category name
|
||||
slug: Category slug (auto-generated if not provided)
|
||||
description: Category description
|
||||
parent_id: Parent category ID
|
||||
|
||||
Returns:
|
||||
Dict with 'success', 'category_id', 'error'
|
||||
"""
|
||||
try:
|
||||
category_data = {'name': name}
|
||||
if slug:
|
||||
category_data['slug'] = slug
|
||||
if description:
|
||||
category_data['description'] = description
|
||||
if parent_id:
|
||||
category_data['parent'] = parent_id
|
||||
|
||||
response = self.session.post(
|
||||
f"{self.api_base}/categories",
|
||||
json=category_data
|
||||
)
|
||||
|
||||
if response.status_code in [200, 201]:
|
||||
data = response.json()
|
||||
return {
|
||||
'success': True,
|
||||
'category_id': data.get('id'),
|
||||
'slug': data.get('slug'),
|
||||
'error': None,
|
||||
}
|
||||
return {
|
||||
'success': False,
|
||||
'category_id': None,
|
||||
'slug': None,
|
||||
'error': f"HTTP {response.status_code}: {response.text}",
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"WordPress category creation failed: {e}")
|
||||
return {
|
||||
'success': False,
|
||||
'category_id': None,
|
||||
'slug': None,
|
||||
'error': str(e),
|
||||
}
|
||||
|
||||
def get_tags(self, per_page: int = 100, page: int = 1) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Get WordPress tags.
|
||||
|
||||
Args:
|
||||
per_page: Number of tags per page
|
||||
page: Page number
|
||||
|
||||
Returns:
|
||||
List of tag dictionaries
|
||||
"""
|
||||
try:
|
||||
response = self.session.get(
|
||||
f"{self.api_base}/tags",
|
||||
params={'per_page': per_page, 'page': page}
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
tags = response.json()
|
||||
return [
|
||||
{
|
||||
'id': tag.get('id'),
|
||||
'name': tag.get('name', ''),
|
||||
'slug': tag.get('slug', ''),
|
||||
'description': tag.get('description', ''),
|
||||
'count': tag.get('count', 0),
|
||||
'link': tag.get('link', '')
|
||||
}
|
||||
for tag in tags
|
||||
]
|
||||
logger.warning(f"Failed to fetch tags: HTTP {response.status_code}")
|
||||
return []
|
||||
except Exception as e:
|
||||
logger.error(f"Error fetching WordPress tags: {e}")
|
||||
return []
|
||||
|
||||
def create_tag(
|
||||
self,
|
||||
name: str,
|
||||
slug: Optional[str] = None,
|
||||
description: Optional[str] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Create a WordPress tag.
|
||||
|
||||
Args:
|
||||
name: Tag name
|
||||
slug: Tag slug (auto-generated if not provided)
|
||||
description: Tag description
|
||||
|
||||
Returns:
|
||||
Dict with 'success', 'tag_id', 'error'
|
||||
"""
|
||||
try:
|
||||
tag_data = {'name': name}
|
||||
if slug:
|
||||
tag_data['slug'] = slug
|
||||
if description:
|
||||
tag_data['description'] = description
|
||||
|
||||
response = self.session.post(
|
||||
f"{self.api_base}/tags",
|
||||
json=tag_data
|
||||
)
|
||||
|
||||
if response.status_code in [200, 201]:
|
||||
data = response.json()
|
||||
return {
|
||||
'success': True,
|
||||
'tag_id': data.get('id'),
|
||||
'slug': data.get('slug'),
|
||||
'error': None,
|
||||
}
|
||||
return {
|
||||
'success': False,
|
||||
'tag_id': None,
|
||||
'slug': None,
|
||||
'error': f"HTTP {response.status_code}: {response.text}",
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"WordPress tag creation failed: {e}")
|
||||
return {
|
||||
'success': False,
|
||||
'tag_id': None,
|
||||
'slug': None,
|
||||
'error': str(e),
|
||||
}
|
||||
|
||||
# Stage 4: WooCommerce Product Methods
|
||||
|
||||
def get_products(
|
||||
self,
|
||||
per_page: int = 100,
|
||||
page: int = 1,
|
||||
status: str = 'publish'
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Get WooCommerce products.
|
||||
|
||||
Args:
|
||||
per_page: Number of products per page
|
||||
page: Page number
|
||||
status: Product status ('publish', 'draft', etc.)
|
||||
|
||||
Returns:
|
||||
List of product dictionaries
|
||||
"""
|
||||
try:
|
||||
response = self.session.get(
|
||||
f"{self.woocommerce_api_base}/products",
|
||||
params={
|
||||
'per_page': per_page,
|
||||
'page': page,
|
||||
'status': status
|
||||
}
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
products = response.json()
|
||||
return [
|
||||
{
|
||||
'id': prod.get('id'),
|
||||
'name': prod.get('name', ''),
|
||||
'slug': prod.get('slug', ''),
|
||||
'sku': prod.get('sku', ''),
|
||||
'type': prod.get('type', 'simple'),
|
||||
'status': prod.get('status', 'publish'),
|
||||
'description': prod.get('description', ''),
|
||||
'short_description': prod.get('short_description', ''),
|
||||
'price': prod.get('price', ''),
|
||||
'regular_price': prod.get('regular_price', ''),
|
||||
'sale_price': prod.get('sale_price', ''),
|
||||
'on_sale': prod.get('on_sale', False),
|
||||
'stock_status': prod.get('stock_status', 'instock'),
|
||||
'stock_quantity': prod.get('stock_quantity'),
|
||||
'categories': prod.get('categories', []),
|
||||
'tags': prod.get('tags', []),
|
||||
'images': prod.get('images', []),
|
||||
'attributes': prod.get('attributes', []),
|
||||
'variations': prod.get('variations', []),
|
||||
'permalink': prod.get('permalink', '')
|
||||
}
|
||||
for prod in products
|
||||
]
|
||||
logger.warning(f"Failed to fetch products: HTTP {response.status_code}")
|
||||
return []
|
||||
except Exception as e:
|
||||
logger.error(f"Error fetching WooCommerce products: {e}")
|
||||
return []
|
||||
|
||||
def create_product(self, product_data: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""
|
||||
Create a WooCommerce product.
|
||||
|
||||
Args:
|
||||
product_data: Product data dictionary
|
||||
|
||||
Returns:
|
||||
Dict with 'success', 'product_id', 'error'
|
||||
"""
|
||||
try:
|
||||
response = self.session.post(
|
||||
f"{self.woocommerce_api_base}/products",
|
||||
json=product_data
|
||||
)
|
||||
|
||||
if response.status_code in [200, 201]:
|
||||
data = response.json()
|
||||
return {
|
||||
'success': True,
|
||||
'product_id': data.get('id'),
|
||||
'slug': data.get('slug'),
|
||||
'error': None,
|
||||
}
|
||||
return {
|
||||
'success': False,
|
||||
'product_id': None,
|
||||
'slug': None,
|
||||
'error': f"HTTP {response.status_code}: {response.text}",
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"WooCommerce product creation failed: {e}")
|
||||
return {
|
||||
'success': False,
|
||||
'product_id': None,
|
||||
'slug': None,
|
||||
'error': str(e),
|
||||
}
|
||||
|
||||
def get_product_categories(self, per_page: int = 100, page: int = 1) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Get WooCommerce product categories.
|
||||
|
||||
Args:
|
||||
per_page: Number of categories per page
|
||||
page: Page number
|
||||
|
||||
Returns:
|
||||
List of product category dictionaries
|
||||
"""
|
||||
try:
|
||||
response = self.session.get(
|
||||
f"{self.woocommerce_api_base}/products/categories",
|
||||
params={'per_page': per_page, 'page': page}
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
categories = response.json()
|
||||
return [
|
||||
{
|
||||
'id': cat.get('id'),
|
||||
'name': cat.get('name', ''),
|
||||
'slug': cat.get('slug', ''),
|
||||
'description': cat.get('description', ''),
|
||||
'count': cat.get('count', 0),
|
||||
'parent': cat.get('parent', 0),
|
||||
'image': cat.get('image', {}).get('src') if cat.get('image') else None
|
||||
}
|
||||
for cat in categories
|
||||
]
|
||||
logger.warning(f"Failed to fetch product categories: HTTP {response.status_code}")
|
||||
return []
|
||||
except Exception as e:
|
||||
logger.error(f"Error fetching WooCommerce product categories: {e}")
|
||||
return []
|
||||
|
||||
def get_product_attributes(self) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Get WooCommerce product attributes.
|
||||
|
||||
Returns:
|
||||
List of product attribute dictionaries with terms
|
||||
"""
|
||||
try:
|
||||
# Get attributes
|
||||
response = self.session.get(f"{self.woocommerce_api_base}/products/attributes")
|
||||
|
||||
if response.status_code != 200:
|
||||
logger.warning(f"Failed to fetch product attributes: HTTP {response.status_code}")
|
||||
return []
|
||||
|
||||
attributes = response.json()
|
||||
result = []
|
||||
|
||||
for attr in attributes:
|
||||
attr_id = attr.get('id')
|
||||
# Get terms for this attribute
|
||||
terms_response = self.session.get(
|
||||
f"{self.woocommerce_api_base}/products/attributes/{attr_id}/terms"
|
||||
)
|
||||
|
||||
terms = []
|
||||
if terms_response.status_code == 200:
|
||||
terms_data = terms_response.json()
|
||||
terms = [
|
||||
{
|
||||
'id': term.get('id'),
|
||||
'name': term.get('name', ''),
|
||||
'slug': term.get('slug', '')
|
||||
}
|
||||
for term in terms_data
|
||||
]
|
||||
|
||||
result.append({
|
||||
'id': attr_id,
|
||||
'name': attr.get('name', ''),
|
||||
'slug': attr.get('slug', ''),
|
||||
'type': attr.get('type', 'select'),
|
||||
'order_by': attr.get('order_by', 'menu_order'),
|
||||
'has_archives': attr.get('has_archives', False),
|
||||
'terms': terms
|
||||
})
|
||||
|
||||
return result
|
||||
except Exception as e:
|
||||
logger.error(f"Error fetching WooCommerce product attributes: {e}")
|
||||
return []
|
||||
|
||||
|
||||
Reference in New Issue
Block a user