salman/igny8
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fixing wp-igny8-integration

Browse Source
This commit is contained in:
alorig
2025-11-29 15:23:12 +05:00
parent 8d096b383a
commit 0549dea124
4 changed files with 320 additions and 298 deletions
Download Patch File Download Diff File Expand all files Collapse all files

212
backend/igny8_core/business/publishing/services/adapters/wordpress_adapter.py
View File

@@ -32,11 +32,18 @@ class WordPressAdapter(BaseAdapter):
Args:
content: Content instance or dict with content data
destination_config: {
# API Key method (preferred):
'site_url': str,
'api_key': str,
# OR username/password method:
'site_url': str,
'username': str,
'app_password': str,
'status': str (optional, default 'draft'),
'featured_image_url': str (optional)
# Optional:
'status': str (default 'draft'),
'featured_image_url': str
}
Returns:
@@ -49,60 +56,46 @@ class WordPressAdapter(BaseAdapter):
}
"""
try:
# Get WordPress client
client = self._get_client(destination_config)
# Extract content data
if hasattr(content, 'title'):
# Content model instance
title = content.title
# Stage 1 schema: content_html is the primary field
content_html = getattr(content, 'content_html', '') or getattr(content, 'html_content', '') or getattr(content, 'content', '')
content_html = getattr(content, 'content_html', '') or ''
elif isinstance(content, dict):
# Dict with content data
title = content.get('title', '')
content_html = content.get('content_html') or content.get('html_content') or content.get('content', '')
content_html = content.get('content_html', '')
else:
raise ValueError(f"Unsupported content type: {type(content)}")
# Get publishing options
status = destination_config.get('status', 'draft')
featured_image_url = destination_config.get('featured_image_url')
# Get site URL
site_url = destination_config.get('site_url')
if not site_url:
raise ValueError("site_url is required in destination_config")
# Publish to WordPress
result = client.create_post(
title=title,
content=content_html,
status=status,
featured_image_url=featured_image_url
)
# Check if using API key authentication
api_key = destination_config.get('api_key')
# Handle different response formats (for compatibility with mocks and real API)
if result.get('success') or result.get('id') or result.get('post_id'):
# Extract post ID from various possible fields
post_id = result.get('post_id') or result.get('id') or result.get('ID')
url = result.get('url') or result.get('link')
return {
'success': True,
'external_id': str(post_id) if post_id else None,
'url': url,
'published_at': datetime.now(),
'metadata': {
'post_id': post_id,
'status': status
}
}
if api_key:
# Use IGNY8 custom endpoint with API key
return self._publish_via_api_key(
site_url=site_url,
api_key=api_key,
content=content,
title=title,
content_html=content_html,
destination_config=destination_config
)
else:
return {
'success': False,
'external_id': None,
'url': None,
'published_at': None,
'metadata': {
'error': result.get('error', 'Unknown error')
}
}
# Use standard WordPress REST API with username/password
return self._publish_via_username_password(
site_url=site_url,
username=destination_config.get('username'),
app_password=destination_config.get('app_password'),
title=title,
content_html=content_html,
destination_config=destination_config
)
except Exception as e:
logger.error(
@@ -119,6 +112,139 @@ class WordPressAdapter(BaseAdapter):
}
}
def _publish_via_api_key(
self,
site_url: str,
api_key: str,
content: Any,
title: str,
content_html: str,
destination_config: Dict[str, Any]
) -> Dict[str, Any]:
"""
Publish via IGNY8 custom WordPress endpoint using API key.
This uses the /wp-json/igny8/v1/publish-content/ endpoint.
"""
import requests
from django.utils.html import strip_tags
# Generate excerpt
excerpt = ''
if content_html:
excerpt = strip_tags(content_html)[:150].strip()
if len(content_html) > 150:
excerpt += '...'
# Prepare payload
content_data = {
'content_id': content.id if hasattr(content, 'id') else None,
'title': title,
'content_html': content_html,
'excerpt': excerpt,
'status': destination_config.get('status', 'publish'),
}
# Add optional fields from content model
if hasattr(content, 'meta_title'):
content_data['seo_title'] = content.meta_title or ''
if hasattr(content, 'meta_description'):
content_data['seo_description'] = content.meta_description or ''
if hasattr(content, 'primary_keyword'):
content_data['primary_keyword'] = content.primary_keyword or ''
if hasattr(content, 'secondary_keywords'):
content_data['secondary_keywords'] = content.secondary_keywords or []
if hasattr(content, 'cluster') and content.cluster:
content_data['cluster_id'] = content.cluster.id
if hasattr(content, 'sector') and content.sector:
content_data['sector_id'] = content.sector.id
# Call WordPress endpoint
url = f"{site_url.rstrip('/')}/wp-json/igny8/v1/publish-content/"
headers = {
'Content-Type': 'application/json',
'X-IGNY8-API-KEY': api_key,
}
response = requests.post(url, json=content_data, headers=headers, timeout=30)
if response.status_code == 201:
wp_data = response.json().get('data', {})
return {
'success': True,
'external_id': str(wp_data.get('post_id')),
'url': wp_data.get('post_url'),
'published_at': datetime.now(),
'metadata': {
'post_id': wp_data.get('post_id'),
'status': destination_config.get('status', 'publish')
}
}
else:
error_msg = f"HTTP {response.status_code}: {response.text}"
return {
'success': False,
'external_id': None,
'url': None,
'published_at': None,
'metadata': {'error': error_msg}
}
def _publish_via_username_password(
self,
site_url: str,
username: str,
app_password: str,
title: str,
content_html: str,
destination_config: Dict[str, Any]
) -> Dict[str, Any]:
"""
Publish via standard WordPress REST API using username/password.
"""
if not username or not app_password:
raise ValueError("username and app_password are required when not using API key")
# Get WordPress client
client = WordPressClient(site_url, username, app_password)
# Get publishing options
status = destination_config.get('status', 'draft')
featured_image_url = destination_config.get('featured_image_url')
# Publish to WordPress
result = client.create_post(
title=title,
content=content_html,
status=status,
featured_image_url=featured_image_url
)
# Handle response
if result.get('success') or result.get('post_id'):
post_id = result.get('post_id') or result.get('id')
url = result.get('url') or result.get('link')
return {
'success': True,
'external_id': str(post_id) if post_id else None,
'url': url,
'published_at': datetime.now(),
'metadata': {
'post_id': post_id,
'status': status
}
}
else:
return {
'success': False,
'external_id': None,
'url': None,
'published_at': None,
'metadata': {
'error': result.get('error', 'Unknown error')
}
}
def test_connection(
self,
config: Dict[str, Any]

13
backend/igny8_core/business/publishing/services/publisher_service.py
View File

@@ -138,8 +138,8 @@ class PublisherService:
if not adapter:
raise ValueError(f"No adapter found for destination: {destination}")
# Get destination config (for now, basic config - can be extended)
destination_config = {'account': account}
# Get destination config
destination_config = {}
# If content has site, try to get integration config
if hasattr(content, 'site') and content.site:
@@ -151,8 +151,13 @@ class PublisherService:
).first()
if integration:
destination_config.update(integration.config_json)
destination_config.update(integration.get_credentials())
# Merge config_json and credentials_json
destination_config.update(integration.config_json or {})
destination_config.update(integration.get_credentials() or {})
# Ensure site_url is set (from config or from site model)
if not destination_config.get('site_url'):
destination_config['site_url'] = content.site.url
# Publish via adapter
result = adapter.publish(content, destination_config)