Add generate_page_content functionality for structured page content generation
- Introduced a new AI function `generate_page_content` to create structured content for website pages using JSON blocks. - Updated `AIEngine` to handle the new function and return appropriate messages for content generation. - Enhanced `PageGenerationService` to utilize the new AI function for generating page content based on blueprints. - Modified `prompts.py` to include detailed content generation requirements for the new function. - Updated site rendering logic to accommodate structured content blocks in various layouts.
This commit is contained in:
IGNY8 VPS (Salman)
@@ -7,6 +7,7 @@ from igny8_core.ai.functions.generate_content import GenerateContentFunction
|
||||
from igny8_core.ai.functions.generate_images import GenerateImagesFunction, generate_images_core
|
||||
from igny8_core.ai.functions.generate_image_prompts import GenerateImagePromptsFunction
|
||||
from igny8_core.ai.functions.generate_site_structure import GenerateSiteStructureFunction
|
||||
from igny8_core.ai.functions.generate_page_content import GeneratePageContentFunction
|
||||
|
||||
__all__ = [
|
||||
'AutoClusterFunction',
|
||||
@@ -16,4 +17,5 @@ __all__ = [
|
||||
'generate_images_core',
|
||||
'GenerateImagePromptsFunction',
|
||||
'GenerateSiteStructureFunction',
|
||||
'GeneratePageContentFunction',
|
||||
]
|
||||
|
||||
273
backend/igny8_core/ai/functions/generate_page_content.py
Normal file
273
backend/igny8_core/ai/functions/generate_page_content.py
Normal file
@@ -0,0 +1,273 @@
|
||||
"""
|
||||
Generate Page Content AI Function
|
||||
Site Builder specific content generation that outputs structured JSON blocks.
|
||||
|
||||
This is separate from the default writer module's GenerateContentFunction.
|
||||
It uses different prompts optimized for site builder pages and outputs
|
||||
structured blocks_json format instead of HTML.
|
||||
"""
|
||||
import logging
|
||||
import json
|
||||
from typing import Dict, List, Any
|
||||
from django.db import transaction
|
||||
from igny8_core.ai.base import BaseAIFunction
|
||||
from igny8_core.business.site_building.models import PageBlueprint
|
||||
from igny8_core.business.content.models import Tasks, Content
|
||||
from igny8_core.ai.ai_core import AICore
|
||||
from igny8_core.ai.prompts import PromptRegistry
|
||||
from igny8_core.ai.settings import get_model_config
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class GeneratePageContentFunction(BaseAIFunction):
|
||||
"""
|
||||
Generate structured page content for Site Builder pages.
|
||||
Outputs JSON blocks format optimized for site rendering.
|
||||
"""
|
||||
|
||||
def get_name(self) -> str:
|
||||
return 'generate_page_content'
|
||||
|
||||
def get_metadata(self) -> Dict:
|
||||
return {
|
||||
'display_name': 'Generate Page Content',
|
||||
'description': 'Generate structured page content with JSON blocks for Site Builder',
|
||||
'phases': {
|
||||
'INIT': 'Initializing page content generation...',
|
||||
'PREP': 'Loading page blueprint and building prompt...',
|
||||
'AI_CALL': 'Generating structured content with AI...',
|
||||
'PARSE': 'Parsing JSON blocks...',
|
||||
'SAVE': 'Saving blocks to page...',
|
||||
'DONE': 'Page content generated!'
|
||||
}
|
||||
}
|
||||
|
||||
def get_max_items(self) -> int:
|
||||
return 20 # Max pages per batch
|
||||
|
||||
def validate(self, payload: dict, account=None) -> Dict:
|
||||
"""Validate page blueprint IDs"""
|
||||
result = super().validate(payload, account)
|
||||
if not result['valid']:
|
||||
return result
|
||||
|
||||
page_ids = payload.get('ids', [])
|
||||
if page_ids:
|
||||
from igny8_core.business.site_building.models import PageBlueprint
|
||||
queryset = PageBlueprint.objects.filter(id__in=page_ids)
|
||||
if account:
|
||||
queryset = queryset.filter(account=account)
|
||||
|
||||
if queryset.count() == 0:
|
||||
return {'valid': False, 'error': 'No page blueprints found'}
|
||||
|
||||
return {'valid': True}
|
||||
|
||||
def prepare(self, payload: dict, account=None) -> List:
|
||||
"""Load page blueprints with relationships"""
|
||||
page_ids = payload.get('ids', [])
|
||||
|
||||
queryset = PageBlueprint.objects.filter(id__in=page_ids)
|
||||
if account:
|
||||
queryset = queryset.filter(account=account)
|
||||
|
||||
# Preload relationships
|
||||
pages = list(queryset.select_related(
|
||||
'site_blueprint', 'account', 'site', 'sector'
|
||||
))
|
||||
|
||||
if not pages:
|
||||
raise ValueError("No page blueprints found")
|
||||
|
||||
return pages
|
||||
|
||||
def build_prompt(self, data: Any, account=None) -> str:
|
||||
"""Build page content generation prompt optimized for Site Builder"""
|
||||
if isinstance(data, list):
|
||||
page = data[0] if data else None
|
||||
else:
|
||||
page = data
|
||||
|
||||
if not page:
|
||||
raise ValueError("No page blueprint provided")
|
||||
|
||||
account = account or page.account
|
||||
|
||||
# Build page context
|
||||
page_context = {
|
||||
'PAGE_TITLE': page.title or page.slug.replace('-', ' ').title(),
|
||||
'PAGE_SLUG': page.slug,
|
||||
'PAGE_TYPE': page.type or 'custom',
|
||||
'SITE_NAME': page.site_blueprint.name if page.site_blueprint else '',
|
||||
'SITE_DESCRIPTION': page.site_blueprint.description or '',
|
||||
}
|
||||
|
||||
# Extract existing block structure hints
|
||||
block_hints = []
|
||||
if page.blocks_json:
|
||||
for block in page.blocks_json[:5]: # First 5 blocks as hints
|
||||
if isinstance(block, dict):
|
||||
block_type = block.get('type', '')
|
||||
heading = block.get('heading') or block.get('title') or ''
|
||||
if block_type and heading:
|
||||
block_hints.append(f"- {block_type}: {heading}")
|
||||
|
||||
if block_hints:
|
||||
page_context['EXISTING_BLOCKS'] = '\n'.join(block_hints)
|
||||
else:
|
||||
page_context['EXISTING_BLOCKS'] = 'None (new page)'
|
||||
|
||||
# Get site blueprint structure hints
|
||||
structure_hints = ''
|
||||
if page.site_blueprint and page.site_blueprint.structure_json:
|
||||
structure = page.site_blueprint.structure_json
|
||||
if isinstance(structure, dict):
|
||||
layout = structure.get('layout', 'default')
|
||||
theme = structure.get('theme', {})
|
||||
structure_hints = f"Layout: {layout}\nTheme: {json.dumps(theme, indent=2)}"
|
||||
|
||||
page_context['STRUCTURE_HINTS'] = structure_hints or 'Default layout'
|
||||
|
||||
# Get prompt from registry (site-builder specific)
|
||||
prompt = PromptRegistry.get_prompt(
|
||||
function_name='generate_page_content',
|
||||
account=account,
|
||||
context=page_context
|
||||
)
|
||||
|
||||
return prompt
|
||||
|
||||
def parse_response(self, response: str, step_tracker=None) -> Dict:
|
||||
"""Parse AI response - must be JSON with blocks structure"""
|
||||
import json
|
||||
|
||||
# Try to extract JSON from response
|
||||
try:
|
||||
# Try direct JSON parse
|
||||
parsed = json.loads(response.strip())
|
||||
except json.JSONDecodeError:
|
||||
# Try to extract JSON object from text
|
||||
try:
|
||||
# Look for JSON object in response
|
||||
start = response.find('{')
|
||||
end = response.rfind('}')
|
||||
if start != -1 and end != -1 and end > start:
|
||||
json_str = response[start:end + 1]
|
||||
parsed = json.loads(json_str)
|
||||
else:
|
||||
raise ValueError("No JSON object found in response")
|
||||
except (json.JSONDecodeError, ValueError) as e:
|
||||
logger.error(f"Failed to parse page content response as JSON: {e}")
|
||||
logger.error(f"Response preview: {response[:500]}")
|
||||
raise ValueError(f"Invalid JSON response from AI: {str(e)}")
|
||||
|
||||
if not isinstance(parsed, dict):
|
||||
raise ValueError("Response must be a JSON object")
|
||||
|
||||
# Validate required fields
|
||||
if 'blocks' not in parsed and 'blocks_json' not in parsed:
|
||||
raise ValueError("Response must include 'blocks' or 'blocks_json' field")
|
||||
|
||||
# Normalize to 'blocks' key
|
||||
if 'blocks_json' in parsed:
|
||||
parsed['blocks'] = parsed.pop('blocks_json')
|
||||
|
||||
return parsed
|
||||
|
||||
def save_output(
|
||||
self,
|
||||
parsed: Any,
|
||||
original_data: Any,
|
||||
account=None,
|
||||
progress_tracker=None,
|
||||
step_tracker=None
|
||||
) -> Dict:
|
||||
"""Save blocks to PageBlueprint and create/update Content record"""
|
||||
if isinstance(original_data, list):
|
||||
page = original_data[0] if original_data else None
|
||||
else:
|
||||
page = original_data
|
||||
|
||||
if not page:
|
||||
raise ValueError("No page blueprint provided for saving")
|
||||
|
||||
if not isinstance(parsed, dict):
|
||||
raise ValueError("Parsed response must be a dict")
|
||||
|
||||
blocks = parsed.get('blocks', [])
|
||||
if not blocks:
|
||||
raise ValueError("No blocks found in parsed response")
|
||||
|
||||
# Ensure blocks is a list
|
||||
if not isinstance(blocks, list):
|
||||
blocks = [blocks]
|
||||
|
||||
with transaction.atomic():
|
||||
# Update PageBlueprint with generated blocks
|
||||
page.blocks_json = blocks
|
||||
page.status = 'ready' # Mark as ready after content generation
|
||||
page.save(update_fields=['blocks_json', 'status', 'updated_at'])
|
||||
|
||||
# Find or create associated Task
|
||||
task_title = f"[Site Builder] {page.title or page.slug.replace('-', ' ').title()}"
|
||||
task = Tasks.objects.filter(
|
||||
account=page.account,
|
||||
site=page.site,
|
||||
sector=page.sector,
|
||||
title=task_title
|
||||
).first()
|
||||
|
||||
# Create or update Content record with blocks
|
||||
if task:
|
||||
content_record, created = Content.objects.get_or_create(
|
||||
task=task,
|
||||
defaults={
|
||||
'account': page.account,
|
||||
'site': page.site,
|
||||
'sector': page.sector,
|
||||
'title': parsed.get('title') or page.title,
|
||||
'html_content': parsed.get('html_content', ''),
|
||||
'word_count': parsed.get('word_count', 0),
|
||||
'status': 'draft',
|
||||
'json_blocks': blocks, # Store blocks in json_blocks
|
||||
'metadata': {
|
||||
'page_id': page.id,
|
||||
'page_slug': page.slug,
|
||||
'page_type': page.type,
|
||||
'generated_by': 'generate_page_content'
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
if not created:
|
||||
# Update existing content
|
||||
content_record.json_blocks = blocks
|
||||
content_record.html_content = parsed.get('html_content', content_record.html_content)
|
||||
content_record.word_count = parsed.get('word_count', content_record.word_count)
|
||||
content_record.title = parsed.get('title') or content_record.title or page.title
|
||||
if not content_record.metadata:
|
||||
content_record.metadata = {}
|
||||
content_record.metadata.update({
|
||||
'page_id': page.id,
|
||||
'page_slug': page.slug,
|
||||
'page_type': page.type,
|
||||
'generated_by': 'generate_page_content'
|
||||
})
|
||||
content_record.save()
|
||||
else:
|
||||
logger.warning(f"No task found for page {page.id}, skipping Content record creation")
|
||||
content_record = None
|
||||
|
||||
logger.info(
|
||||
f"[GeneratePageContentFunction] Saved {len(blocks)} blocks to page {page.id} "
|
||||
f"(Content ID: {content_record.id if content_record else 'N/A'})"
|
||||
)
|
||||
|
||||
return {
|
||||
'count': 1,
|
||||
'pages_updated': 1,
|
||||
'blocks_count': len(blocks),
|
||||
'content_id': content_record.id if content_record else None
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user