594 lines
20 KiB
Python
594 lines
20 KiB
Python
"""
|
|
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, List
|
|
from django.conf import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class WordPressClient:
|
|
"""
|
|
WordPress REST API client for content publishing and sync.
|
|
"""
|
|
|
|
def __init__(self, site_url: str, username: str = None, app_password: str = None):
|
|
"""
|
|
Initialize WordPress client.
|
|
|
|
Args:
|
|
site_url: WordPress site URL (e.g., https://example.com)
|
|
username: WordPress username or application password username
|
|
app_password: WordPress application password
|
|
"""
|
|
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
|
|
self.session = requests.Session()
|
|
|
|
# Set up authentication if provided
|
|
if username and app_password:
|
|
self.session.auth = (username, app_password)
|
|
|
|
def test_connection(self) -> Dict[str, Any]:
|
|
"""
|
|
Test connection to WordPress site.
|
|
|
|
Returns:
|
|
Dict with 'success', 'message', 'wp_version'
|
|
"""
|
|
try:
|
|
response = self.session.get(f"{self.api_base}/")
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
return {
|
|
'success': True,
|
|
'message': 'Connection successful',
|
|
'wp_version': data.get('version', 'Unknown'),
|
|
}
|
|
return {
|
|
'success': False,
|
|
'message': f"HTTP {response.status_code}",
|
|
'wp_version': None,
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"WordPress connection test failed: {e}")
|
|
return {
|
|
'success': False,
|
|
'message': str(e),
|
|
'wp_version': None,
|
|
}
|
|
|
|
def create_post(
|
|
self,
|
|
title: str,
|
|
content: str,
|
|
status: str = 'draft',
|
|
featured_image_url: Optional[str] = None,
|
|
**kwargs
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Create a new WordPress post.
|
|
|
|
Args:
|
|
title: Post title
|
|
content: Post content (HTML or blocks)
|
|
status: Post status ('draft', 'publish', 'pending')
|
|
featured_image_url: URL of featured image (must be uploaded first)
|
|
**kwargs: Additional post fields (excerpt, categories, etc.)
|
|
|
|
Returns:
|
|
Dict with 'success', 'post_id', 'url', 'error'
|
|
"""
|
|
try:
|
|
post_data = {
|
|
'title': title,
|
|
'content': content,
|
|
'status': status,
|
|
**kwargs
|
|
}
|
|
|
|
if featured_image_url:
|
|
# Convert URL to media ID if needed
|
|
media_id = self._get_media_id_from_url(featured_image_url)
|
|
if media_id:
|
|
post_data['featured_media'] = media_id
|
|
|
|
response = self.session.post(f"{self.api_base}/posts", json=post_data)
|
|
|
|
if response.status_code in [200, 201]:
|
|
data = response.json()
|
|
return {
|
|
'success': True,
|
|
'post_id': data.get('id'),
|
|
'url': data.get('link'),
|
|
'error': None,
|
|
}
|
|
return {
|
|
'success': False,
|
|
'post_id': None,
|
|
'url': None,
|
|
'error': f"HTTP {response.status_code}: {response.text}",
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"WordPress post creation failed: {e}")
|
|
return {
|
|
'success': False,
|
|
'post_id': None,
|
|
'url': None,
|
|
'error': str(e),
|
|
}
|
|
|
|
def upload_image(self, image_url: str, filename: str = None) -> Dict[str, Any]:
|
|
"""
|
|
Upload an image to WordPress media library.
|
|
|
|
Args:
|
|
image_url: URL of image to upload
|
|
filename: Optional filename
|
|
|
|
Returns:
|
|
Dict with 'success', 'media_id', 'url', 'error'
|
|
"""
|
|
try:
|
|
# Download image
|
|
img_response = requests.get(image_url)
|
|
if img_response.status_code != 200:
|
|
return {
|
|
'success': False,
|
|
'media_id': None,
|
|
'url': None,
|
|
'error': f"Failed to download image: HTTP {img_response.status_code}",
|
|
}
|
|
|
|
# Upload to WordPress
|
|
files = {
|
|
'file': (filename or 'image.jpg', img_response.content, img_response.headers.get('content-type', 'image/jpeg'))
|
|
}
|
|
|
|
response = self.session.post(f"{self.api_base}/media", files=files)
|
|
|
|
if response.status_code in [200, 201]:
|
|
data = response.json()
|
|
return {
|
|
'success': True,
|
|
'media_id': data.get('id'),
|
|
'url': data.get('source_url'),
|
|
'error': None,
|
|
}
|
|
return {
|
|
'success': False,
|
|
'media_id': None,
|
|
'url': None,
|
|
'error': f"HTTP {response.status_code}: {response.text}",
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"WordPress image upload failed: {e}")
|
|
return {
|
|
'success': False,
|
|
'media_id': None,
|
|
'url': None,
|
|
'error': str(e),
|
|
}
|
|
|
|
def _get_media_id_from_url(self, url: str) -> Optional[int]:
|
|
"""Helper to get media ID from URL (if already uploaded)."""
|
|
# TODO: Implement media lookup by URL
|
|
return None
|
|
|
|
def sync_settings(self, settings_data: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Sync settings to WordPress via custom IGNY8 endpoint.
|
|
|
|
Args:
|
|
settings_data: Settings dictionary to sync
|
|
|
|
Returns:
|
|
Dict with 'success', 'message', 'error'
|
|
"""
|
|
try:
|
|
response = self.session.post(
|
|
f"{self.igny8_api_base}/sync-settings",
|
|
json=settings_data
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
return {
|
|
'success': True,
|
|
'message': 'Settings synced successfully',
|
|
'error': None,
|
|
}
|
|
return {
|
|
'success': False,
|
|
'message': None,
|
|
'error': f"HTTP {response.status_code}: {response.text}",
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"WordPress settings sync failed: {e}")
|
|
return {
|
|
'success': False,
|
|
'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 []
|
|
|