Add Linker and Optimizer modules with API integration and frontend components
- Added Linker and Optimizer apps to `INSTALLED_APPS` in `settings.py`. - Configured API endpoints for Linker and Optimizer in `urls.py`. - Implemented `OptimizeContentFunction` for content optimization in the AI module. - Created prompts for content optimization and site structure generation. - Updated `OptimizerService` to utilize the new AI function for content optimization. - Developed frontend components including dashboards and content lists for Linker and Optimizer. - Integrated new routes and sidebar navigation for Linker and Optimizer in the frontend. - Enhanced content management with source and sync status filters in the Writer module. - Comprehensive test coverage added for new features and components.
This commit is contained in:
alorig
167
backend/igny8_core/ai/functions/optimize_content.py
Normal file
167
backend/igny8_core/ai/functions/optimize_content.py
Normal file
@@ -0,0 +1,167 @@
|
||||
"""
|
||||
Optimize Content AI Function
|
||||
Phase 4 – Linker & Optimizer
|
||||
"""
|
||||
import json
|
||||
import logging
|
||||
from typing import Any, Dict
|
||||
|
||||
from igny8_core.ai.base import BaseAIFunction
|
||||
from igny8_core.ai.prompts import PromptRegistry
|
||||
from igny8_core.business.content.models import Content
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class OptimizeContentFunction(BaseAIFunction):
|
||||
"""AI function that optimizes content for SEO, readability, and engagement."""
|
||||
|
||||
def get_name(self) -> str:
|
||||
return 'optimize_content'
|
||||
|
||||
def get_metadata(self) -> Dict:
|
||||
metadata = super().get_metadata()
|
||||
metadata.update({
|
||||
'display_name': 'Optimize Content',
|
||||
'description': 'Optimize content for SEO, readability, and engagement.',
|
||||
'phases': {
|
||||
'INIT': 'Validating content data…',
|
||||
'PREP': 'Preparing content context…',
|
||||
'AI_CALL': 'Optimizing content with AI…',
|
||||
'PARSE': 'Parsing optimized content…',
|
||||
'SAVE': 'Saving optimized content…',
|
||||
'DONE': 'Content optimized!'
|
||||
}
|
||||
})
|
||||
return metadata
|
||||
|
||||
def validate(self, payload: dict, account=None) -> Dict[str, Any]:
|
||||
if not payload.get('ids'):
|
||||
return {'valid': False, 'error': 'Content ID is required'}
|
||||
return {'valid': True}
|
||||
|
||||
def prepare(self, payload: dict, account=None) -> Dict[str, Any]:
|
||||
content_ids = payload.get('ids', [])
|
||||
queryset = Content.objects.filter(id__in=content_ids)
|
||||
if account:
|
||||
queryset = queryset.filter(account=account)
|
||||
content = queryset.select_related('account', 'site', 'sector').first()
|
||||
if not content:
|
||||
raise ValueError("Content not found")
|
||||
|
||||
# Get current scores from analyzer
|
||||
from igny8_core.business.optimization.services.analyzer import ContentAnalyzer
|
||||
analyzer = ContentAnalyzer()
|
||||
scores_before = analyzer.analyze(content)
|
||||
|
||||
return {
|
||||
'content': content,
|
||||
'scores_before': scores_before,
|
||||
'html_content': content.html_content or '',
|
||||
'meta_title': content.meta_title or '',
|
||||
'meta_description': content.meta_description or '',
|
||||
'primary_keyword': content.primary_keyword or '',
|
||||
}
|
||||
|
||||
def build_prompt(self, data: Dict[str, Any], account=None) -> str:
|
||||
content: Content = data['content']
|
||||
scores_before = data.get('scores_before', {})
|
||||
|
||||
context = {
|
||||
'CONTENT_TITLE': content.title or 'Untitled',
|
||||
'HTML_CONTENT': data.get('html_content', ''),
|
||||
'META_TITLE': data.get('meta_title', ''),
|
||||
'META_DESCRIPTION': data.get('meta_description', ''),
|
||||
'PRIMARY_KEYWORD': data.get('primary_keyword', ''),
|
||||
'WORD_COUNT': str(content.word_count or 0),
|
||||
'CURRENT_SCORES': json.dumps(scores_before, indent=2),
|
||||
'SOURCE': content.source,
|
||||
'INTERNAL_LINKS_COUNT': str(len(content.internal_links) if content.internal_links else 0),
|
||||
}
|
||||
|
||||
return PromptRegistry.get_prompt(
|
||||
'optimize_content',
|
||||
account=account or content.account,
|
||||
context=context
|
||||
)
|
||||
|
||||
def parse_response(self, response: str, step_tracker=None) -> Dict[str, Any]:
|
||||
if not response:
|
||||
raise ValueError("AI response is empty")
|
||||
|
||||
response = response.strip()
|
||||
try:
|
||||
return self._ensure_dict(json.loads(response))
|
||||
except json.JSONDecodeError:
|
||||
logger.warning("Response not valid JSON, attempting to extract JSON object")
|
||||
cleaned = self._extract_json_object(response)
|
||||
if cleaned:
|
||||
return self._ensure_dict(json.loads(cleaned))
|
||||
raise ValueError("Unable to parse AI response into JSON")
|
||||
|
||||
def save_output(
|
||||
self,
|
||||
parsed: Dict[str, Any],
|
||||
original_data: Dict[str, Any],
|
||||
account=None,
|
||||
progress_tracker=None,
|
||||
step_tracker=None
|
||||
) -> Dict[str, Any]:
|
||||
content: Content = original_data['content']
|
||||
|
||||
# Extract optimized content
|
||||
optimized_html = parsed.get('html_content') or parsed.get('content') or content.html_content
|
||||
optimized_meta_title = parsed.get('meta_title') or content.meta_title
|
||||
optimized_meta_description = parsed.get('meta_description') or content.meta_description
|
||||
|
||||
# Update content
|
||||
content.html_content = optimized_html
|
||||
if optimized_meta_title:
|
||||
content.meta_title = optimized_meta_title
|
||||
if optimized_meta_description:
|
||||
content.meta_description = optimized_meta_description
|
||||
|
||||
# Recalculate word count
|
||||
from igny8_core.business.content.services.content_generation_service import ContentGenerationService
|
||||
content_service = ContentGenerationService()
|
||||
content.word_count = content_service._count_words(optimized_html)
|
||||
|
||||
# Increment optimizer version
|
||||
content.optimizer_version += 1
|
||||
|
||||
# Get scores after optimization
|
||||
from igny8_core.business.optimization.services.analyzer import ContentAnalyzer
|
||||
analyzer = ContentAnalyzer()
|
||||
scores_after = analyzer.analyze(content)
|
||||
content.optimization_scores = scores_after
|
||||
|
||||
content.save(update_fields=[
|
||||
'html_content', 'meta_title', 'meta_description',
|
||||
'word_count', 'optimizer_version', 'optimization_scores', 'updated_at'
|
||||
])
|
||||
|
||||
return {
|
||||
'success': True,
|
||||
'content_id': content.id,
|
||||
'scores_before': original_data.get('scores_before', {}),
|
||||
'scores_after': scores_after,
|
||||
'word_count_before': original_data.get('word_count', 0),
|
||||
'word_count_after': content.word_count,
|
||||
'html_content': optimized_html,
|
||||
'meta_title': optimized_meta_title,
|
||||
'meta_description': optimized_meta_description,
|
||||
}
|
||||
|
||||
# Helper methods
|
||||
def _ensure_dict(self, data: Any) -> Dict[str, Any]:
|
||||
if isinstance(data, dict):
|
||||
return data
|
||||
raise ValueError("AI response must be a JSON object")
|
||||
|
||||
def _extract_json_object(self, text: str) -> str:
|
||||
start = text.find('{')
|
||||
end = text.rfind('}')
|
||||
if start != -1 and end != -1 and end > start:
|
||||
return text[start:end + 1]
|
||||
return ''
|
||||
|
||||
2
backend/igny8_core/ai/functions/tests/__init__.py
Normal file
2
backend/igny8_core/ai/functions/tests/__init__.py
Normal file
@@ -0,0 +1,2 @@
|
||||
# AI functions tests
|
||||
|
||||
179
backend/igny8_core/ai/functions/tests/test_optimize_content.py
Normal file
179
backend/igny8_core/ai/functions/tests/test_optimize_content.py
Normal file
@@ -0,0 +1,179 @@
|
||||
"""
|
||||
Tests for OptimizeContentFunction
|
||||
"""
|
||||
from unittest.mock import Mock, patch, MagicMock
|
||||
from django.test import TestCase
|
||||
from igny8_core.business.content.models import Content
|
||||
from igny8_core.ai.functions.optimize_content import OptimizeContentFunction
|
||||
from igny8_core.api.tests.test_integration_base import IntegrationTestBase
|
||||
|
||||
|
||||
class OptimizeContentFunctionTests(IntegrationTestBase):
|
||||
"""Tests for OptimizeContentFunction"""
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.function = OptimizeContentFunction()
|
||||
|
||||
# Create test content
|
||||
self.content = Content.objects.create(
|
||||
account=self.account,
|
||||
site=self.site,
|
||||
sector=self.sector,
|
||||
title="Test Content",
|
||||
html_content="<p>This is test content.</p>",
|
||||
meta_title="Test Title",
|
||||
meta_description="Test description",
|
||||
primary_keyword="test keyword",
|
||||
word_count=500,
|
||||
status='draft'
|
||||
)
|
||||
|
||||
def test_function_validation_phase(self):
|
||||
"""Test validation phase"""
|
||||
# Valid payload
|
||||
result = self.function.validate({'ids': [self.content.id]}, self.account)
|
||||
self.assertTrue(result['valid'])
|
||||
|
||||
# Invalid payload - missing ids
|
||||
result = self.function.validate({}, self.account)
|
||||
self.assertFalse(result['valid'])
|
||||
self.assertIn('error', result)
|
||||
|
||||
def test_function_prep_phase(self):
|
||||
"""Test prep phase"""
|
||||
payload = {'ids': [self.content.id]}
|
||||
|
||||
data = self.function.prepare(payload, self.account)
|
||||
|
||||
self.assertIn('content', data)
|
||||
self.assertIn('scores_before', data)
|
||||
self.assertIn('html_content', data)
|
||||
self.assertEqual(data['content'].id, self.content.id)
|
||||
|
||||
def test_function_prep_phase_content_not_found(self):
|
||||
"""Test prep phase with non-existent content"""
|
||||
payload = {'ids': [99999]}
|
||||
|
||||
with self.assertRaises(ValueError):
|
||||
self.function.prepare(payload, self.account)
|
||||
|
||||
@patch('igny8_core.ai.functions.optimize_content.PromptRegistry.get_prompt')
|
||||
def test_function_build_prompt(self, mock_get_prompt):
|
||||
"""Test prompt building"""
|
||||
mock_get_prompt.return_value = "Test prompt"
|
||||
|
||||
data = {
|
||||
'content': self.content,
|
||||
'html_content': '<p>Test</p>',
|
||||
'meta_title': 'Title',
|
||||
'meta_description': 'Description',
|
||||
'primary_keyword': 'keyword',
|
||||
'scores_before': {'overall_score': 50.0}
|
||||
}
|
||||
|
||||
prompt = self.function.build_prompt(data, self.account)
|
||||
|
||||
self.assertEqual(prompt, "Test prompt")
|
||||
mock_get_prompt.assert_called_once()
|
||||
# Check that context was passed
|
||||
call_args = mock_get_prompt.call_args
|
||||
self.assertIn('context', call_args.kwargs)
|
||||
|
||||
def test_function_parse_response_valid_json(self):
|
||||
"""Test parsing valid JSON response"""
|
||||
response = '{"html_content": "<p>Optimized</p>", "meta_title": "New Title"}'
|
||||
|
||||
parsed = self.function.parse_response(response)
|
||||
|
||||
self.assertIn('html_content', parsed)
|
||||
self.assertEqual(parsed['html_content'], "<p>Optimized</p>")
|
||||
self.assertEqual(parsed['meta_title'], "New Title")
|
||||
|
||||
def test_function_parse_response_invalid_json(self):
|
||||
"""Test parsing invalid JSON response"""
|
||||
response = "This is not JSON"
|
||||
|
||||
with self.assertRaises(ValueError):
|
||||
self.function.parse_response(response)
|
||||
|
||||
def test_function_parse_response_extracts_json_object(self):
|
||||
"""Test that JSON object is extracted from text"""
|
||||
response = 'Some text {"html_content": "<p>Optimized</p>"} more text'
|
||||
|
||||
parsed = self.function.parse_response(response)
|
||||
|
||||
self.assertIn('html_content', parsed)
|
||||
self.assertEqual(parsed['html_content'], "<p>Optimized</p>")
|
||||
|
||||
@patch('igny8_core.business.optimization.services.analyzer.ContentAnalyzer.analyze')
|
||||
@patch('igny8_core.business.content.services.content_generation_service.ContentGenerationService._count_words')
|
||||
def test_function_save_phase(self, mock_count_words, mock_analyze):
|
||||
"""Test save phase updates content"""
|
||||
mock_count_words.return_value = 600
|
||||
mock_analyze.return_value = {
|
||||
'seo_score': 75.0,
|
||||
'readability_score': 80.0,
|
||||
'engagement_score': 70.0,
|
||||
'overall_score': 75.0
|
||||
}
|
||||
|
||||
parsed = {
|
||||
'html_content': '<p>Optimized content.</p>',
|
||||
'meta_title': 'Optimized Title',
|
||||
'meta_description': 'Optimized Description'
|
||||
}
|
||||
|
||||
original_data = {
|
||||
'content': self.content,
|
||||
'scores_before': {'overall_score': 50.0},
|
||||
'word_count': 500
|
||||
}
|
||||
|
||||
result = self.function.save_output(parsed, original_data, self.account)
|
||||
|
||||
self.assertTrue(result['success'])
|
||||
self.assertEqual(result['content_id'], self.content.id)
|
||||
|
||||
# Refresh content from DB
|
||||
self.content.refresh_from_db()
|
||||
self.assertEqual(self.content.html_content, '<p>Optimized content.</p>')
|
||||
self.assertEqual(self.content.optimizer_version, 1)
|
||||
self.assertIsNotNone(self.content.optimization_scores)
|
||||
|
||||
def test_function_handles_invalid_content_id(self):
|
||||
"""Test that function handles invalid content ID"""
|
||||
payload = {'ids': [99999]}
|
||||
|
||||
with self.assertRaises(ValueError):
|
||||
self.function.prepare(payload, self.account)
|
||||
|
||||
def test_function_respects_account_isolation(self):
|
||||
"""Test that function respects account isolation"""
|
||||
from igny8_core.auth.models import Account
|
||||
other_account = Account.objects.create(
|
||||
name="Other Account",
|
||||
slug="other",
|
||||
plan=self.plan,
|
||||
owner=self.user
|
||||
)
|
||||
|
||||
payload = {'ids': [self.content.id]}
|
||||
|
||||
# Should not find content from different account
|
||||
with self.assertRaises(ValueError):
|
||||
self.function.prepare(payload, other_account)
|
||||
|
||||
def test_get_name(self):
|
||||
"""Test get_name method"""
|
||||
self.assertEqual(self.function.get_name(), 'optimize_content')
|
||||
|
||||
def test_get_metadata(self):
|
||||
"""Test get_metadata method"""
|
||||
metadata = self.function.get_metadata()
|
||||
|
||||
self.assertIn('display_name', metadata)
|
||||
self.assertIn('description', metadata)
|
||||
self.assertIn('phases', metadata)
|
||||
self.assertEqual(metadata['display_name'], 'Optimize Content')
|
||||
|
||||
Reference in New Issue
Block a user