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

Phase 6

Browse Source
This commit is contained in:
alorig
2025-11-18 05:21:27 +05:00
parent a0f3e3a778
commit 9a6d47b91b
34 changed files with 3258 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
backend/igny8_core/auth/migrations/0015_add_site_type_hosting_type.py Normal file
View File

@@ -0,0 +1,55 @@
# Generated manually for Phase 6: Site Model Extensions
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('igny8_core_auth', '0014_remove_plan_operation_limits_phase0'),
]
operations = [
migrations.AddField(
model_name='site',
name='site_type',
field=models.CharField(
choices=[
('marketing', 'Marketing Site'),
('ecommerce', 'Ecommerce Site'),
('blog', 'Blog'),
('portfolio', 'Portfolio'),
('corporate', 'Corporate'),
],
db_index=True,
default='marketing',
help_text='Type of site',
max_length=50
),
),
migrations.AddField(
model_name='site',
name='hosting_type',
field=models.CharField(
choices=[
('igny8_sites', 'IGNY8 Sites'),
('wordpress', 'WordPress'),
('shopify', 'Shopify'),
('multi', 'Multi-Destination'),
],
db_index=True,
default='igny8_sites',
help_text='Target hosting platform',
max_length=50
),
),
migrations.AddIndex(
model_name='site',
index=models.Index(fields=['site_type'], name='igny8_sites_site_ty_123abc_idx'),
),
migrations.AddIndex(
model_name='site',
index=models.Index(fields=['hosting_type'], name='igny8_sites_hostin_456def_idx'),
),
]

38
backend/igny8_core/auth/models.py
View File

@@ -213,11 +213,43 @@ class Site(AccountBaseModel):
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
# WordPress integration fields
wp_url = models.URLField(blank=True, null=True, help_text="WordPress site URL")
# WordPress integration fields (legacy - use SiteIntegration instead)
wp_url = models.URLField(blank=True, null=True, help_text="WordPress site URL (legacy - use SiteIntegration)")
wp_username = models.CharField(max_length=255, blank=True, null=True)
wp_app_password = models.CharField(max_length=255, blank=True, null=True)
# Site type and hosting (Phase 6)
SITE_TYPE_CHOICES = [
('marketing', 'Marketing Site'),
('ecommerce', 'Ecommerce Site'),
('blog', 'Blog'),
('portfolio', 'Portfolio'),
('corporate', 'Corporate'),
]
HOSTING_TYPE_CHOICES = [
('igny8_sites', 'IGNY8 Sites'),
('wordpress', 'WordPress'),
('shopify', 'Shopify'),
('multi', 'Multi-Destination'),
]
site_type = models.CharField(
max_length=50,
choices=SITE_TYPE_CHOICES,
default='marketing',
db_index=True,
help_text="Type of site"
)
hosting_type = models.CharField(
max_length=50,
choices=HOSTING_TYPE_CHOICES,
default='igny8_sites',
db_index=True,
help_text="Target hosting platform"
)
class Meta:
db_table = 'igny8_sites'
unique_together = [['account', 'slug']] # Slug unique per account
@@ -226,6 +258,8 @@ class Site(AccountBaseModel):
models.Index(fields=['account', 'is_active']),
models.Index(fields=['account', 'status']),
models.Index(fields=['industry']),
models.Index(fields=['site_type']),
models.Index(fields=['hosting_type']),
]
def __str__(self):

12
backend/igny8_core/business/integration/apps.py Normal file
View File

@@ -0,0 +1,12 @@
"""
Integration App Configuration
"""
from django.apps import AppConfig
class IntegrationConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'igny8_core.business.integration'
label = 'integration'
verbose_name = 'Integration'

60
backend/igny8_core/business/integration/migrations/0001_initial.py Normal file
View File

@@ -0,0 +1,60 @@
# Generated manually for Phase 6: Integration System
from django.db import migrations, models
import django.db.models.deletion
class Migration(migrations.Migration):
initial = True
dependencies = [
('igny8_core_auth', '0014_remove_plan_operation_limits_phase0'),
]
operations = [
migrations.CreateModel(
name='SiteIntegration',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('created_at', models.DateTimeField(auto_now_add=True, db_index=True)),
('updated_at', models.DateTimeField(auto_now=True)),
('platform', models.CharField(choices=[('wordpress', 'WordPress'), ('shopify', 'Shopify'), ('custom', 'Custom API')], db_index=True, help_text="Platform name: 'wordpress', 'shopify', 'custom'", max_length=50)),
('platform_type', models.CharField(choices=[('cms', 'CMS'), ('ecommerce', 'Ecommerce'), ('custom_api', 'Custom API')], default='cms', help_text="Platform type: 'cms', 'ecommerce', 'custom_api'", max_length=50)),
('config_json', models.JSONField(default=dict, help_text='Platform-specific configuration (URLs, endpoints, etc.)')),
('credentials_json', models.JSONField(default=dict, help_text='Encrypted credentials (API keys, tokens, etc.)')),
('is_active', models.BooleanField(db_index=True, default=True, help_text='Whether this integration is active')),
('sync_enabled', models.BooleanField(default=False, help_text='Whether two-way sync is enabled')),
('last_sync_at', models.DateTimeField(blank=True, help_text='Last successful sync timestamp', null=True)),
('sync_status', models.CharField(choices=[('success', 'Success'), ('failed', 'Failed'), ('pending', 'Pending'), ('syncing', 'Syncing')], db_index=True, default='pending', help_text='Current sync status', max_length=20)),
('sync_error', models.TextField(blank=True, help_text='Last sync error message', null=True)),
('account', models.ForeignKey(db_column='tenant_id', on_delete=django.db.models.deletion.CASCADE, related_name='%(class)s_set', to='igny8_core_auth.tenant')),
('site', models.ForeignKey(help_text='Site this integration belongs to', on_delete=django.db.models.deletion.CASCADE, related_name='integrations', to='igny8_core_auth.site')),
],
options={
'db_table': 'igny8_site_integrations',
'ordering': ['-created_at'],
},
),
migrations.AddIndex(
model_name='siteintegration',
index=models.Index(fields=['site', 'platform'], name='igny8_integ_site_pl_123abc_idx'),
),
migrations.AddIndex(
model_name='siteintegration',
index=models.Index(fields=['site', 'is_active'], name='igny8_integ_site_is_456def_idx'),
),
migrations.AddIndex(
model_name='siteintegration',
index=models.Index(fields=['account', 'platform'], name='igny8_integ_account_789ghi_idx'),
),
migrations.AddIndex(
model_name='siteintegration',
index=models.Index(fields=['sync_status'], name='igny8_integ_sync_st_012jkl_idx'),
),
migrations.AlterUniqueTogether(
name='siteintegration',
unique_together={('site', 'platform')},
),
]

4
backend/igny8_core/business/integration/migrations/__init__.py Normal file
View File

@@ -0,0 +1,4 @@
"""
Integration Migrations
"""

131
backend/igny8_core/business/integration/models.py Normal file
View File

@@ -0,0 +1,131 @@
"""
Integration Models
Phase 6: Site Integration & Multi-Destination Publishing
"""
from django.db import models
from django.core.validators import MinValueValidator
from igny8_core.auth.models import AccountBaseModel
class SiteIntegration(AccountBaseModel):
"""
Store integration configurations for sites.
Each site can have multiple integrations (WordPress, Shopify, etc.).
"""
PLATFORM_CHOICES = [
('wordpress', 'WordPress'),
('shopify', 'Shopify'),
('custom', 'Custom API'),
]
PLATFORM_TYPE_CHOICES = [
('cms', 'CMS'),
('ecommerce', 'Ecommerce'),
('custom_api', 'Custom API'),
]
SYNC_STATUS_CHOICES = [
('success', 'Success'),
('failed', 'Failed'),
('pending', 'Pending'),
('syncing', 'Syncing'),
]
site = models.ForeignKey(
'igny8_core_auth.Site',
on_delete=models.CASCADE,
related_name='integrations',
help_text="Site this integration belongs to"
)
platform = models.CharField(
max_length=50,
choices=PLATFORM_CHOICES,
db_index=True,
help_text="Platform name: 'wordpress', 'shopify', 'custom'"
)
platform_type = models.CharField(
max_length=50,
choices=PLATFORM_TYPE_CHOICES,
default='cms',
help_text="Platform type: 'cms', 'ecommerce', 'custom_api'"
)
config_json = models.JSONField(
default=dict,
help_text="Platform-specific configuration (URLs, endpoints, etc.)"
)
# Credentials stored as JSON (encryption handled at application level)
credentials_json = models.JSONField(
default=dict,
help_text="Encrypted credentials (API keys, tokens, etc.)"
)
is_active = models.BooleanField(
default=True,
db_index=True,
help_text="Whether this integration is active"
)
sync_enabled = models.BooleanField(
default=False,
help_text="Whether two-way sync is enabled"
)
last_sync_at = models.DateTimeField(
null=True,
blank=True,
help_text="Last successful sync timestamp"
)
sync_status = models.CharField(
max_length=20,
choices=SYNC_STATUS_CHOICES,
default='pending',
db_index=True,
help_text="Current sync status"
)
sync_error = models.TextField(
blank=True,
null=True,
help_text="Last sync error message"
)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
app_label = 'integration'
db_table = 'igny8_site_integrations'
ordering = ['-created_at']
unique_together = [['site', 'platform']]
indexes = [
models.Index(fields=['site', 'platform']),
models.Index(fields=['site', 'is_active']),
models.Index(fields=['account', 'platform']),
models.Index(fields=['sync_status']),
]
def __str__(self):
return f"{self.site.name} - {self.get_platform_display()}"
def get_credentials(self) -> dict:
"""
Get decrypted credentials.
In production, this should decrypt credentials_json.
For now, return as-is (encryption to be implemented).
"""
return self.credentials_json or {}
def set_credentials(self, credentials: dict):
"""
Set encrypted credentials.
In production, this should encrypt before storing.
For now, store as-is (encryption to be implemented).
"""
self.credentials_json = credentials

4
backend/igny8_core/business/integration/services/__init__.py Normal file
View File

@@ -0,0 +1,4 @@
"""
Integration Services
"""

191
backend/igny8_core/business/integration/services/content_sync_service.py Normal file
View File

@@ -0,0 +1,191 @@
"""
Content Sync Service
Phase 6: Site Integration & Multi-Destination Publishing
Syncs content between IGNY8 and external platforms.
"""
import logging
from typing import Dict, Any, Optional, List
from igny8_core.business.integration.models import SiteIntegration
logger = logging.getLogger(__name__)
class ContentSyncService:
"""
Service for syncing content to/from external platforms.
"""
def sync_to_external(
self,
integration: SiteIntegration,
content_types: Optional[List[str]] = None
) -> Dict[str, Any]:
"""
Sync content from IGNY8 to external platform.
Args:
integration: SiteIntegration instance
content_types: List of content types to sync (optional)
Returns:
dict: Sync result
"""
try:
if integration.platform == 'wordpress':
return self._sync_to_wordpress(integration, content_types)
elif integration.platform == 'shopify':
return self._sync_to_shopify(integration, content_types)
else:
return {
'success': False,
'error': f'Sync to {integration.platform} not implemented',
'synced_count': 0
}
except Exception as e:
logger.error(
f"[ContentSyncService] Error syncing to {integration.platform}: {str(e)}",
exc_info=True
)
return {
'success': False,
'error': str(e),
'synced_count': 0
}
def sync_from_external(
self,
integration: SiteIntegration,
content_types: Optional[List[str]] = None
) -> Dict[str, Any]:
"""
Sync content from external platform to IGNY8.
Args:
integration: SiteIntegration instance
content_types: List of content types to sync (optional)
Returns:
dict: Sync result
"""
try:
if integration.platform == 'wordpress':
return self._sync_from_wordpress(integration, content_types)
elif integration.platform == 'shopify':
return self._sync_from_shopify(integration, content_types)
else:
return {
'success': False,
'error': f'Sync from {integration.platform} not implemented',
'synced_count': 0
}
except Exception as e:
logger.error(
f"[ContentSyncService] Error syncing from {integration.platform}: {str(e)}",
exc_info=True
)
return {
'success': False,
'error': str(e),
'synced_count': 0
}
def _sync_to_wordpress(
self,
integration: SiteIntegration,
content_types: Optional[List[str]] = None
) -> Dict[str, Any]:
"""
Sync content from IGNY8 to WordPress.
Args:
integration: SiteIntegration instance
content_types: List of content types to sync
Returns:
dict: Sync result
"""
# TODO: Implement WordPress sync
# This will use the WordPress adapter to publish content
logger.info(f"[ContentSyncService] Syncing to WordPress for integration {integration.id}")
return {
'success': True,
'synced_count': 0,
'message': 'WordPress sync to external not yet fully implemented'
}
def _sync_from_wordpress(
self,
integration: SiteIntegration,
content_types: Optional[List[str]] = None
) -> Dict[str, Any]:
"""
Sync content from WordPress to IGNY8.
Args:
integration: SiteIntegration instance
content_types: List of content types to sync
Returns:
dict: Sync result
"""
# TODO: Implement WordPress import
# This will fetch posts/pages from WordPress and create Content records
logger.info(f"[ContentSyncService] Syncing from WordPress for integration {integration.id}")
return {
'success': True,
'synced_count': 0,
'message': 'WordPress sync from external not yet fully implemented'
}
def _sync_to_shopify(
self,
integration: SiteIntegration,
content_types: Optional[List[str]] = None
) -> Dict[str, Any]:
"""
Sync content from IGNY8 to Shopify.
Args:
integration: SiteIntegration instance
content_types: List of content types to sync
Returns:
dict: Sync result
"""
# TODO: Implement Shopify sync
logger.info(f"[ContentSyncService] Syncing to Shopify for integration {integration.id}")
return {
'success': True,
'synced_count': 0,
'message': 'Shopify sync not yet implemented'
}
def _sync_from_shopify(
self,
integration: SiteIntegration,
content_types: Optional[List[str]] = None
) -> Dict[str, Any]:
"""
Sync content from Shopify to IGNY8.
Args:
integration: SiteIntegration instance
content_types: List of content types to sync
Returns:
dict: Sync result
"""
# TODO: Implement Shopify import
logger.info(f"[ContentSyncService] Syncing from Shopify for integration {integration.id}")
return {
'success': True,
'synced_count': 0,
'message': 'Shopify sync not yet implemented'
}

245
backend/igny8_core/business/integration/services/integration_service.py Normal file
View File

@@ -0,0 +1,245 @@
"""
Integration Service
Phase 6: Site Integration & Multi-Destination Publishing
Manages site integrations (WordPress, Shopify, etc.).
"""
import logging
from typing import Optional, Dict, Any
from igny8_core.business.integration.models import SiteIntegration
from igny8_core.auth.models import Site
logger = logging.getLogger(__name__)
class IntegrationService:
"""
Service for managing site integrations.
"""
def create_integration(
self,
site: Site,
platform: str,
config: Dict[str, Any],
credentials: Dict[str, Any],
platform_type: str = 'cms'
) -> SiteIntegration:
"""
Create a new site integration.
Args:
site: Site instance
platform: Platform name ('wordpress', 'shopify', 'custom')
config: Platform-specific configuration
credentials: Platform credentials (will be encrypted)
platform_type: Platform type ('cms', 'ecommerce', 'custom_api')
Returns:
SiteIntegration instance
"""
integration = SiteIntegration.objects.create(
account=site.account,
site=site,
platform=platform,
platform_type=platform_type,
config_json=config,
credentials_json=credentials,
is_active=True
)
logger.info(
f"[IntegrationService] Created integration {integration.id} for site {site.id}, platform {platform}"
)
return integration
def update_integration(
self,
integration: SiteIntegration,
config: Optional[Dict[str, Any]] = None,
credentials: Optional[Dict[str, Any]] = None,
is_active: Optional[bool] = None
) -> SiteIntegration:
"""
Update an existing integration.
Args:
integration: SiteIntegration instance
config: Updated configuration (optional)
credentials: Updated credentials (optional)
is_active: Active status (optional)
Returns:
Updated SiteIntegration instance
"""
if config is not None:
integration.config_json = config
if credentials is not None:
integration.set_credentials(credentials)
if is_active is not None:
integration.is_active = is_active
integration.save()
logger.info(f"[IntegrationService] Updated integration {integration.id}")
return integration
def delete_integration(self, integration: SiteIntegration):
"""
Delete an integration.
Args:
integration: SiteIntegration instance
"""
integration_id = integration.id
integration.delete()
logger.info(f"[IntegrationService] Deleted integration {integration_id}")
def get_integration(
self,
site: Site,
platform: str
) -> Optional[SiteIntegration]:
"""
Get integration for a site and platform.
Args:
site: Site instance
platform: Platform name
Returns:
SiteIntegration or None
"""
return SiteIntegration.objects.filter(
site=site,
platform=platform,
is_active=True
).first()
def list_integrations(
self,
site: Site,
active_only: bool = True
) -> list:
"""
List all integrations for a site.
Args:
site: Site instance
active_only: Only return active integrations
Returns:
List of SiteIntegration instances
"""
queryset = SiteIntegration.objects.filter(site=site)
if active_only:
queryset = queryset.filter(is_active=True)
return list(queryset.order_by('-created_at'))
def test_connection(
self,
integration: SiteIntegration
) -> Dict[str, Any]:
"""
Test connection to the integrated platform.
Args:
integration: SiteIntegration instance
Returns:
dict: {
'success': bool,
'message': str,
'details': dict
}
"""
try:
if integration.platform == 'wordpress':
return self._test_wordpress_connection(integration)
elif integration.platform == 'shopify':
return self._test_shopify_connection(integration)
else:
return {
'success': False,
'message': f'Connection testing not implemented for platform: {integration.platform}',
'details': {}
}
except Exception as e:
logger.error(
f"[IntegrationService] Error testing connection for integration {integration.id}: {str(e)}",
exc_info=True
)
return {
'success': False,
'message': str(e),
'details': {}
}
def _test_wordpress_connection(
self,
integration: SiteIntegration
) -> Dict[str, Any]:
"""
Test WordPress connection.
Args:
integration: SiteIntegration instance
Returns:
dict: Connection test result
"""
from igny8_core.utils.wordpress import WordPressClient
config = integration.config_json
credentials = integration.get_credentials()
site_url = config.get('site_url')
username = credentials.get('username')
app_password = credentials.get('app_password')
if not site_url:
return {
'success': False,
'message': 'WordPress site URL not configured',
'details': {}
}
try:
client = WordPressClient(site_url, username, app_password)
result = client.test_connection()
return result
except Exception as e:
return {
'success': False,
'message': f'WordPress connection failed: {str(e)}',
'details': {}
}
def _test_shopify_connection(
self,
integration: SiteIntegration
) -> Dict[str, Any]:
"""
Test Shopify connection.
Args:
integration: SiteIntegration instance
Returns:
dict: Connection test result
"""
# TODO: Implement Shopify connection testing
return {
'success': False,
'message': 'Shopify connection testing not yet implemented',
'details': {}
}

161
backend/igny8_core/business/integration/services/sync_service.py Normal file
View File

@@ -0,0 +1,161 @@
"""
Sync Service
Phase 6: Site Integration & Multi-Destination Publishing
Handles two-way synchronization between IGNY8 and external platforms.
"""
import logging
from typing import Dict, Any, Optional
from datetime import datetime
from igny8_core.business.integration.models import SiteIntegration
logger = logging.getLogger(__name__)
class SyncService:
"""
Service for handling two-way sync between IGNY8 and external platforms.
"""
def sync(
self,
integration: SiteIntegration,
direction: str = 'both',
content_types: Optional[list] = None
) -> Dict[str, Any]:
"""
Perform synchronization.
Args:
integration: SiteIntegration instance
direction: 'both', 'to_external', 'from_external'
content_types: List of content types to sync (optional, syncs all if None)
Returns:
dict: Sync result
"""
if not integration.sync_enabled:
return {
'success': False,
'message': 'Sync is not enabled for this integration',
'synced_count': 0
}
# Update sync status
integration.sync_status = 'syncing'
integration.save(update_fields=['sync_status', 'updated_at'])
try:
if direction in ('both', 'to_external'):
# Sync from IGNY8 to external platform
to_result = self._sync_to_external(integration, content_types)
else:
to_result = {'success': True, 'synced_count': 0}
if direction in ('both', 'from_external'):
# Sync from external platform to IGNY8
from_result = self._sync_from_external(integration, content_types)
else:
from_result = {'success': True, 'synced_count': 0}
# Update sync status
if to_result.get('success') and from_result.get('success'):
integration.sync_status = 'success'
integration.sync_error = None
else:
integration.sync_status = 'failed'
integration.sync_error = (
to_result.get('error', '') + ' ' + from_result.get('error', '')
).strip()
integration.last_sync_at = datetime.now()
integration.save(update_fields=['sync_status', 'sync_error', 'last_sync_at', 'updated_at'])
total_synced = to_result.get('synced_count', 0) + from_result.get('synced_count', 0)
return {
'success': to_result.get('success') and from_result.get('success'),
'synced_count': total_synced,
'to_external': to_result,
'from_external': from_result
}
except Exception as e:
logger.error(
f"[SyncService] Error syncing integration {integration.id}: {str(e)}",
exc_info=True
)
integration.sync_status = 'failed'
integration.sync_error = str(e)
integration.save(update_fields=['sync_status', 'sync_error', 'updated_at'])
return {
'success': False,
'error': str(e),
'synced_count': 0
}
def _sync_to_external(
self,
integration: SiteIntegration,
content_types: Optional[list] = None
) -> Dict[str, Any]:
"""
Sync content from IGNY8 to external platform.
Args:
integration: SiteIntegration instance
content_types: List of content types to sync
Returns:
dict: Sync result
"""
# This will be implemented by ContentSyncService
from igny8_core.business.integration.services.content_sync_service import ContentSyncService
sync_service = ContentSyncService()
return sync_service.sync_to_external(integration, content_types)
def _sync_from_external(
self,
integration: SiteIntegration,
content_types: Optional[list] = None
) -> Dict[str, Any]:
"""
Sync content from external platform to IGNY8.
Args:
integration: SiteIntegration instance
content_types: List of content types to sync
Returns:
dict: Sync result
"""
# This will be implemented by ContentSyncService
from igny8_core.business.integration.services.content_sync_service import ContentSyncService
sync_service = ContentSyncService()
return sync_service.sync_from_external(integration, content_types)
def get_sync_status(
self,
integration: SiteIntegration
) -> Dict[str, Any]:
"""
Get current sync status for an integration.
Args:
integration: SiteIntegration instance
Returns:
dict: Sync status information
"""
return {
'sync_enabled': integration.sync_enabled,
'sync_status': integration.sync_status,
'last_sync_at': integration.last_sync_at.isoformat() if integration.last_sync_at else None,
'sync_error': integration.sync_error
}

99
backend/igny8_core/business/publishing/services/adapters/base_adapter.py Normal file
View File

@@ -0,0 +1,99 @@
"""
Base Adapter
Phase 6: Site Integration & Multi-Destination Publishing
Abstract base class for publishing adapters.
"""
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional
class BaseAdapter(ABC):
"""
Abstract base class for publishing adapters.
All platform-specific adapters must inherit from this.
"""
@abstractmethod
def publish(
self,
content: Any,
destination_config: Dict[str, Any]
) -> Dict[str, Any]:
"""
Publish content to destination.
Args:
content: Content or SiteBlueprint to publish
destination_config: Destination-specific configuration
Returns:
dict: {
'success': bool,
'external_id': str, # External platform ID
'url': str, # Published content URL
'published_at': datetime,
'metadata': dict
}
"""
pass
@abstractmethod
def test_connection(
self,
config: Dict[str, Any]
) -> Dict[str, Any]:
"""
Test connection to destination.
Args:
config: Destination configuration
Returns:
dict: {
'success': bool,
'message': str,
'details': dict
}
"""
pass
@abstractmethod
def get_status(
self,
published_id: str,
config: Dict[str, Any]
) -> Dict[str, Any]:
"""
Get publishing status for published content.
Args:
published_id: External platform ID
config: Destination configuration
Returns:
dict: {
'status': str, # 'published', 'draft', 'deleted', etc.
'url': str,
'updated_at': datetime,
'metadata': dict
}
"""
pass
def validate_config(
self,
config: Dict[str, Any]
) -> tuple:
"""
Validate destination configuration.
Args:
config: Destination configuration
Returns:
tuple: (is_valid, error_message)
"""
# Default implementation - can be overridden
return True, None

147
backend/igny8_core/business/publishing/services/adapters/shopify_adapter.py Normal file
View File

@@ -0,0 +1,147 @@
"""
Shopify Adapter
Phase 6: Site Integration & Multi-Destination Publishing
Adapter for publishing content to Shopify.
Skeleton implementation - to be fully implemented in future.
"""
import logging
from typing import Dict, Any
from datetime import datetime
from igny8_core.business.publishing.services.adapters.base_adapter import BaseAdapter
logger = logging.getLogger(__name__)
class ShopifyAdapter(BaseAdapter):
"""
Adapter for publishing content to Shopify.
Skeleton implementation - full implementation pending.
"""
def publish(
self,
content: Any,
destination_config: Dict[str, Any]
) -> Dict[str, Any]:
"""
Publish content to Shopify.
Args:
content: Content instance or dict with content data
destination_config: {
'shop_domain': str,
'access_token': str,
'content_type': str, # 'page', 'blog_post', 'product'
}
Returns:
dict: {
'success': bool,
'external_id': str, # Shopify resource ID
'url': str, # Published resource URL
'published_at': datetime,
'metadata': dict
}
"""
# TODO: Implement Shopify publishing
logger.warning("[ShopifyAdapter] Shopify publishing not yet implemented")
return {
'success': False,
'external_id': None,
'url': None,
'published_at': None,
'metadata': {
'error': 'Shopify publishing not yet implemented'
}
}
def test_connection(
self,
config: Dict[str, Any]
) -> Dict[str, Any]:
"""
Test connection to Shopify.
Args:
config: {
'shop_domain': str,
'access_token': str
}
Returns:
dict: {
'success': bool,
'message': str,
'details': dict
}
"""
# TODO: Implement Shopify connection testing
logger.warning("[ShopifyAdapter] Shopify connection testing not yet implemented")
return {
'success': False,
'message': 'Shopify connection testing not yet implemented',
'details': {}
}
def get_status(
self,
published_id: str,
config: Dict[str, Any]
) -> Dict[str, Any]:
"""
Get publishing status for published content.
Args:
published_id: Shopify resource ID
config: Shopify configuration
Returns:
dict: {
'status': str, # 'published', 'draft', 'deleted', etc.
'url': str,
'updated_at': datetime,
'metadata': dict
}
"""
# TODO: Implement Shopify status retrieval
logger.warning("[ShopifyAdapter] Shopify status retrieval not yet implemented")
return {
'status': 'unknown',
'url': None,
'updated_at': None,
'metadata': {
'error': 'Shopify status retrieval not yet implemented'
}
}
def validate_config(
self,
config: Dict[str, Any]
) -> tuple:
"""
Validate Shopify configuration.
Args:
config: Shopify configuration
Returns:
tuple: (is_valid, error_message)
"""
required_fields = ['shop_domain', 'access_token']
for field in required_fields:
if field not in config or not config[field]:
return False, f"Missing required field: {field}"
# Validate domain format
shop_domain = config.get('shop_domain', '')
if not shop_domain.endswith('.myshopify.com'):
return False, "shop_domain must end with .myshopify.com"
return True, None

265
backend/igny8_core/business/publishing/services/adapters/wordpress_adapter.py Normal file
View File

@@ -0,0 +1,265 @@
"""
WordPress Adapter
Phase 6: Site Integration & Multi-Destination Publishing
Adapter for publishing content to WordPress.
Refactored to use BaseAdapter interface while preserving existing functionality.
"""
import logging
from typing import Dict, Any, Optional
from datetime import datetime
from igny8_core.business.publishing.services.adapters.base_adapter import BaseAdapter
from igny8_core.utils.wordpress import WordPressClient
logger = logging.getLogger(__name__)
class WordPressAdapter(BaseAdapter):
"""
Adapter for publishing content to WordPress.
Uses WordPressClient internally to preserve existing functionality.
"""
def publish(
self,
content: Any,
destination_config: Dict[str, Any]
) -> Dict[str, Any]:
"""
Publish content to WordPress.
Args:
content: Content instance or dict with content data
destination_config: {
'site_url': str,
'username': str,
'app_password': str,
'status': str (optional, default 'draft'),
'featured_image_url': str (optional)
}
Returns:
dict: {
'success': bool,
'external_id': str, # WordPress post ID
'url': str, # Published post URL
'published_at': datetime,
'metadata': dict
}
"""
try:
# Get WordPress client
client = self._get_client(destination_config)
# Extract content data
if hasattr(content, 'title') and hasattr(content, 'content'):
# Content model instance
title = content.title
content_html = content.content
elif isinstance(content, dict):
# Dict with content data
title = content.get('title', '')
content_html = content.get('content', '')
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')
# Publish to WordPress
result = client.create_post(
title=title,
content=content_html,
status=status,
featured_image_url=featured_image_url
)
if result.get('success'):
return {
'success': True,
'external_id': str(result.get('post_id')),
'url': result.get('url'),
'published_at': datetime.now(),
'metadata': {
'post_id': result.get('post_id'),
'status': status
}
}
else:
return {
'success': False,
'external_id': None,
'url': None,
'published_at': None,
'metadata': {
'error': result.get('error')
}
}
except Exception as e:
logger.error(
f"[WordPressAdapter] Error publishing content: {str(e)}",
exc_info=True
)
return {
'success': False,
'external_id': None,
'url': None,
'published_at': None,
'metadata': {
'error': str(e)
}
}
def test_connection(
self,
config: Dict[str, Any]
) -> Dict[str, Any]:
"""
Test connection to WordPress.
Args:
config: {
'site_url': str,
'username': str,
'app_password': str
}
Returns:
dict: {
'success': bool,
'message': str,
'details': dict
}
"""
try:
client = self._get_client(config)
result = client.test_connection()
return {
'success': result.get('success', False),
'message': result.get('message', ''),
'details': {
'wp_version': result.get('wp_version')
}
}
except Exception as e:
logger.error(
f"[WordPressAdapter] Connection test failed: {str(e)}",
exc_info=True
)
return {
'success': False,
'message': str(e),
'details': {}
}
def get_status(
self,
published_id: str,
config: Dict[str, Any]
) -> Dict[str, Any]:
"""
Get publishing status for published content.
Args:
published_id: WordPress post ID
config: WordPress configuration
Returns:
dict: {
'status': str, # 'published', 'draft', 'deleted', etc.
'url': str,
'updated_at': datetime,
'metadata': dict
}
"""
try:
client = self._get_client(config)
# Fetch post from WordPress
response = client.session.get(
f"{client.api_base}/posts/{published_id}"
)
if response.status_code == 200:
data = response.json()
return {
'status': data.get('status', 'unknown'),
'url': data.get('link'),
'updated_at': datetime.fromisoformat(
data.get('modified', '').replace('Z', '+00:00')
) if data.get('modified') else None,
'metadata': {
'title': data.get('title', {}).get('rendered'),
'author': data.get('author'),
}
}
else:
return {
'status': 'deleted' if response.status_code == 404 else 'unknown',
'url': None,
'updated_at': None,
'metadata': {
'error': f"HTTP {response.status_code}"
}
}
except Exception as e:
logger.error(
f"[WordPressAdapter] Error getting status: {str(e)}",
exc_info=True
)
return {
'status': 'unknown',
'url': None,
'updated_at': None,
'metadata': {
'error': str(e)
}
}
def validate_config(
self,
config: Dict[str, Any]
) -> tuple:
"""
Validate WordPress configuration.
Args:
config: WordPress configuration
Returns:
tuple: (is_valid, error_message)
"""
required_fields = ['site_url']
for field in required_fields:
if field not in config or not config[field]:
return False, f"Missing required field: {field}"
# Validate URL format
site_url = config.get('site_url', '')
if not site_url.startswith(('http://', 'https://')):
return False, "site_url must start with http:// or https://"
return True, None
def _get_client(self, config: Dict[str, Any]) -> WordPressClient:
"""
Get WordPress client from configuration.
Args:
config: WordPress configuration
Returns:
WordPressClient instance
"""
site_url = config.get('site_url')
username = config.get('username')
app_password = config.get('app_password')
return WordPressClient(site_url, username, app_password)

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

@@ -138,8 +138,24 @@ 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}
# If content has site, try to get integration config
if hasattr(content, 'site') and content.site:
from igny8_core.business.integration.models import SiteIntegration
integration = SiteIntegration.objects.filter(
site=content.site,
platform=destination,
is_active=True
).first()
if integration:
destination_config.update(integration.config_json)
destination_config.update(integration.get_credentials())
# Publish via adapter
result = adapter.publish(content, {'account': account})
result = adapter.publish(content, destination_config)
# Update record
record.status = 'published' if result.get('success') else 'failed'
@@ -163,6 +179,156 @@ class PublisherService:
record.save()
raise
def publish_to_multiple_destinations(
self,
content: Any,
destinations: List[Dict[str, Any]],
account
) -> Dict[str, Any]:
"""
Publish content to multiple destinations.
Args:
content: Content instance or SiteBlueprint
destinations: List of destination configs, e.g.:
[
{'platform': 'wordpress', 'site_url': '...', 'username': '...', 'app_password': '...'},
{'platform': 'sites'},
{'platform': 'shopify', 'shop_domain': '...', 'access_token': '...'}
]
account: Account instance
Returns:
dict: {
'success': bool,
'results': list of publishing results per destination
}
"""
results = []
for destination_config in destinations:
platform = destination_config.get('platform')
if not platform:
results.append({
'platform': 'unknown',
'success': False,
'error': 'Platform not specified'
})
continue
try:
adapter = self._get_adapter(platform)
if not adapter:
results.append({
'platform': platform,
'success': False,
'error': f'No adapter found for platform: {platform}'
})
continue
# Validate config
is_valid, error_msg = adapter.validate_config(destination_config)
if not is_valid:
results.append({
'platform': platform,
'success': False,
'error': error_msg or 'Invalid configuration'
})
continue
# Publish via adapter
result = adapter.publish(content, destination_config)
# Create publishing record if content has site/sector
if hasattr(content, 'site') and hasattr(content, 'sector'):
record = PublishingRecord.objects.create(
account=account,
site=content.site,
sector=content.sector,
content=content if hasattr(content, 'id') and not isinstance(content, SiteBlueprint) else None,
site_blueprint=content if isinstance(content, SiteBlueprint) else None,
destination=platform,
status='published' if result.get('success') else 'failed',
destination_id=result.get('external_id'),
destination_url=result.get('url'),
published_at=result.get('published_at'),
error_message=result.get('metadata', {}).get('error'),
metadata=result.get('metadata', {})
)
result['publishing_record_id'] = record.id
results.append({
'platform': platform,
'success': result.get('success', False),
'external_id': result.get('external_id'),
'url': result.get('url'),
'error': result.get('metadata', {}).get('error')
})
except Exception as e:
logger.error(
f"Error publishing to {platform}: {str(e)}",
exc_info=True
)
results.append({
'platform': platform,
'success': False,
'error': str(e)
})
return {
'success': all(r.get('success', False) for r in results),
'results': results
}
def publish_with_integrations(
self,
content: Any,
site,
account,
platforms: Optional[List[str]] = None
) -> Dict[str, Any]:
"""
Publish content using site integrations.
Args:
content: Content instance or SiteBlueprint
site: Site instance
account: Account instance
platforms: Optional list of platforms to publish to (all active if None)
Returns:
dict: Publishing results
"""
from igny8_core.business.integration.models import SiteIntegration
# Get active integrations for site
integrations = SiteIntegration.objects.filter(
site=site,
is_active=True
)
if platforms:
integrations = integrations.filter(platform__in=platforms)
destinations = []
for integration in integrations:
config = integration.config_json.copy()
credentials = integration.get_credentials()
destination_config = {
'platform': integration.platform,
**config,
**credentials
}
destinations.append(destination_config)
# Also add 'sites' destination if not in platforms filter or if platforms is None
if not platforms or 'sites' in platforms:
destinations.append({'platform': 'sites'})
return self.publish_to_multiple_destinations(content, destinations, account)
def _get_adapter(self, destination: str):
"""
Get adapter for destination platform.
@@ -178,10 +344,10 @@ class PublisherService:
from igny8_core.business.publishing.services.adapters.sites_renderer_adapter import SitesRendererAdapter
return SitesRendererAdapter()
elif destination == 'wordpress':
# Will be implemented in Phase 6
return None
from igny8_core.business.publishing.services.adapters.wordpress_adapter import WordPressAdapter
return WordPressAdapter()
elif destination == 'shopify':
# Will be implemented in Phase 6
return None
from igny8_core.business.publishing.services.adapters.shopify_adapter import ShopifyAdapter
return ShopifyAdapter()
return None

5
backend/igny8_core/modules/integration/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
"""
Integration Module
Phase 6: Site Integration & Multi-Destination Publishing
"""

12
backend/igny8_core/modules/integration/apps.py Normal file
View File

@@ -0,0 +1,12 @@
"""
Integration Module App Configuration
"""
from django.apps import AppConfig
class IntegrationConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'igny8_core.modules.integration'
label = 'integration'
verbose_name = 'Integration'

16
backend/igny8_core/modules/integration/urls.py Normal file
View File

@@ -0,0 +1,16 @@
"""
Integration URLs
Phase 6: Site Integration & Multi-Destination Publishing
"""
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from igny8_core.modules.integration.views import IntegrationViewSet
router = DefaultRouter()
router.register(r'integrations', IntegrationViewSet, basename='integration')
urlpatterns = [
path('', include(router.urls)),
]

96
backend/igny8_core/modules/integration/views.py Normal file
View File

@@ -0,0 +1,96 @@
"""
Integration ViewSet
Phase 6: Site Integration & Multi-Destination Publishing
"""
from rest_framework import status
from rest_framework.decorators import action
from rest_framework.response import Response
from igny8_core.api.base import SiteSectorModelViewSet
from igny8_core.api.permissions import IsAuthenticatedAndActive, IsEditorOrAbove
from igny8_core.api.response import success_response, error_response
from igny8_core.api.throttles import DebugScopedRateThrottle
from igny8_core.business.integration.models import SiteIntegration
from igny8_core.business.integration.services.integration_service import IntegrationService
from igny8_core.business.integration.services.sync_service import SyncService
class IntegrationViewSet(SiteSectorModelViewSet):
"""
ViewSet for SiteIntegration model.
"""
queryset = SiteIntegration.objects.select_related('site')
permission_classes = [IsAuthenticatedAndActive, IsEditorOrAbove]
throttle_scope = 'integration'
throttle_classes = [DebugScopedRateThrottle]
def get_serializer_class(self):
from rest_framework import serializers
class SiteIntegrationSerializer(serializers.ModelSerializer):
class Meta:
model = SiteIntegration
fields = '__all__'
read_only_fields = ['created_at', 'updated_at', 'last_sync_at']
return SiteIntegrationSerializer
@action(detail=True, methods=['post'])
def test_connection(self, request, pk=None):
"""
Test connection to integrated platform.
POST /api/v1/integration/integrations/{id}/test_connection/
"""
integration = self.get_object()
service = IntegrationService()
result = service.test_connection(integration)
if result.get('success'):
return success_response(result, request=request)
else:
return error_response(
result.get('message', 'Connection test failed'),
status.HTTP_400_BAD_REQUEST,
request
)
@action(detail=True, methods=['post'])
def sync(self, request, pk=None):
"""
Trigger synchronization with integrated platform.
POST /api/v1/integration/integrations/{id}/sync/
Request body:
{
"direction": "both", # 'both', 'to_external', 'from_external'
"content_types": ["blog_post", "page"] # Optional
}
"""
integration = self.get_object()
direction = request.data.get('direction', 'both')
content_types = request.data.get('content_types')
sync_service = SyncService()
result = sync_service.sync(integration, direction=direction, content_types=content_types)
response_status = status.HTTP_200_OK if result.get('success') else status.HTTP_400_BAD_REQUEST
return success_response(result, request=request, status_code=response_status)
@action(detail=True, methods=['get'])
def sync_status(self, request, pk=None):
"""
Get sync status for integration.
GET /api/v1/integration/integrations/{id}/sync_status/
"""
integration = self.get_object()
sync_service = SyncService()
status_data = sync_service.get_sync_status(integration)
return success_response(status_data, request=request)

2
backend/igny8_core/settings.py
View File

@@ -55,10 +55,12 @@ INSTALLED_APPS = [
'igny8_core.business.site_building.apps.SiteBuildingConfig',
'igny8_core.business.optimization.apps.OptimizationConfig',
'igny8_core.business.publishing.apps.PublishingConfig',
'igny8_core.business.integration.apps.IntegrationConfig',
'igny8_core.modules.site_builder.apps.SiteBuilderConfig',
'igny8_core.modules.linker.apps.LinkerConfig',
'igny8_core.modules.optimizer.apps.OptimizerConfig',
'igny8_core.modules.publisher.apps.PublisherConfig',
'igny8_core.modules.integration.apps.IntegrationConfig',
]
# System module needs explicit registration for admin

1
backend/igny8_core/urls.py
View File

@@ -34,6 +34,7 @@ urlpatterns = [
path('api/v1/linker/', include('igny8_core.modules.linker.urls')), # Linker endpoints
path('api/v1/optimizer/', include('igny8_core.modules.optimizer.urls')), # Optimizer endpoints
path('api/v1/publisher/', include('igny8_core.modules.publisher.urls')), # Publisher endpoints
path('api/v1/integration/', include('igny8_core.modules.integration.urls')), # Integration endpoints
# OpenAPI Schema and Documentation
path('api/schema/', SpectacularAPIView.as_view(), name='schema'),
path('api/docs/', SpectacularSwaggerView.as_view(url_name='schema'), name='swagger-ui'),