Files
igny8/backend/igny8_core/business/content/models.py
2026-01-03 20:08:16 +00:00

824 lines
29 KiB
Python

from django.db import models
from django.core.validators import MinValueValidator
from igny8_core.auth.models import SiteSectorBaseModel
from igny8_core.common.soft_delete import SoftDeletableModel, SoftDeleteManager
class Tasks(SoftDeletableModel, SiteSectorBaseModel):
"""Tasks model for content generation queue"""
STATUS_CHOICES = [
('queued', 'Queued'),
('completed', 'Completed'),
]
CONTENT_TYPE_CHOICES = [
('post', 'Post'),
('page', 'Page'),
('product', 'Product'),
('taxonomy', 'Taxonomy'),
]
CONTENT_STRUCTURE_CHOICES = [
# Post structures
('article', 'Article'),
('guide', 'Guide'),
('comparison', 'Comparison'),
('review', 'Review'),
('listicle', 'Listicle'),
# Page structures
('landing_page', 'Landing Page'),
('business_page', 'Business Page'),
('service_page', 'Service Page'),
('general', 'General'),
('cluster_hub', 'Cluster Hub'),
# Product structures
('product_page', 'Product Page'),
# Taxonomy structures
('category_archive', 'Category Archive'),
('tag_archive', 'Tag Archive'),
('attribute_archive', 'Attribute Archive'),
]
title = models.CharField(max_length=255, db_index=True)
description = models.TextField(blank=True, null=True)
cluster = models.ForeignKey(
'planner.Clusters',
on_delete=models.SET_NULL,
null=True,
blank=False,
related_name='tasks',
limit_choices_to={'sector': models.F('sector')},
help_text="Parent cluster (required)"
)
idea = models.ForeignKey(
'planner.ContentIdeas',
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name='tasks',
help_text="Optional content idea reference",
db_column='idea_id'
)
content_type = models.CharField(
max_length=100,
db_index=True,
help_text="Content type: post, page, product, taxonomy",
choices=CONTENT_TYPE_CHOICES,
default='post',
blank=True,
null=True
)
content_structure = models.CharField(
max_length=100,
db_index=True,
help_text="Content structure: article, guide, comparison, review, listicle, landing_page, etc.",
choices=CONTENT_STRUCTURE_CHOICES,
default='article',
blank=True,
null=True
)
taxonomy_term = models.ForeignKey(
'ContentTaxonomy',
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name='tasks',
help_text="Optional taxonomy term assignment",
db_column='taxonomy_id'
)
keywords = models.TextField(
blank=True,
null=True,
help_text="Comma-separated keywords for this task"
)
word_count = models.IntegerField(
default=1000,
validators=[MinValueValidator(100)],
help_text="Target word count for content generation"
)
status = models.CharField(max_length=50, choices=STATUS_CHOICES, default='queued')
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
app_label = 'writer'
db_table = 'igny8_tasks'
ordering = ['-created_at']
verbose_name = 'Task'
verbose_name_plural = 'Tasks'
indexes = [
models.Index(fields=['title']),
models.Index(fields=['status']),
models.Index(fields=['cluster']),
models.Index(fields=['content_type']),
models.Index(fields=['content_structure']),
models.Index(fields=['site', 'sector']),
]
objects = SoftDeleteManager()
all_objects = models.Manager()
def __str__(self):
return self.title
def soft_delete(self, user=None, reason=None, retention_days=None):
"""
Override soft_delete to cascade to related models.
This ensures Images and ContentClusterMap are also deleted when a Task is deleted.
"""
import logging
logger = logging.getLogger(__name__)
# Soft-delete related Images (which are also SoftDeletable)
related_images = self.images.filter(is_deleted=False)
images_count = related_images.count()
for image in related_images:
image.soft_delete(user=user, reason=f"Parent task deleted: {reason or 'No reason'}")
# Hard-delete ContentClusterMap (not soft-deletable)
cluster_maps_count = self.cluster_mappings.count()
self.cluster_mappings.all().delete()
# Hard-delete ContentAttribute (not soft-deletable)
attributes_count = self.attribute_mappings.count()
self.attribute_mappings.all().delete()
logger.info(
f"[Tasks.soft_delete] Task {self.id} '{self.title}' cascade delete: "
f"{images_count} images, {cluster_maps_count} cluster maps, {attributes_count} attributes"
)
# Call parent soft_delete
super().soft_delete(user=user, reason=reason, retention_days=retention_days)
class ContentTaxonomyRelation(models.Model):
"""Through model for Content-Taxonomy many-to-many relationship"""
content = models.ForeignKey('Content', on_delete=models.CASCADE, db_column='content_id')
taxonomy = models.ForeignKey('ContentTaxonomy', on_delete=models.CASCADE, db_column='taxonomy_id')
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
app_label = 'writer'
db_table = 'igny8_content_taxonomy_relations'
unique_together = [['content', 'taxonomy']]
class Content(SoftDeletableModel, SiteSectorBaseModel):
"""
Content model for AI-generated or WordPress-imported content.
Final architecture: simplified content management.
"""
CONTENT_TYPE_CHOICES = [
('post', 'Post'),
('page', 'Page'),
('product', 'Product'),
('taxonomy', 'Taxonomy'),
]
CONTENT_STRUCTURE_CHOICES = [
# Post structures
('article', 'Article'),
('guide', 'Guide'),
('comparison', 'Comparison'),
('review', 'Review'),
('listicle', 'Listicle'),
# Page structures
('landing_page', 'Landing Page'),
('business_page', 'Business Page'),
('service_page', 'Service Page'),
('general', 'General'),
('cluster_hub', 'Cluster Hub'),
# Product structures
('product_page', 'Product Page'),
# Taxonomy structures
('category_archive', 'Category Archive'),
('tag_archive', 'Tag Archive'),
('attribute_archive', 'Attribute Archive'),
]
# Core content fields
title = models.CharField(max_length=255, db_index=True)
content_html = models.TextField(help_text="Final HTML content")
word_count = models.IntegerField(
default=0,
help_text="Actual word count of content (calculated from HTML)"
)
# SEO fields
meta_title = models.CharField(max_length=255, blank=True, null=True, help_text="SEO meta title")
meta_description = models.TextField(blank=True, null=True, help_text="SEO meta description")
primary_keyword = models.CharField(max_length=255, blank=True, null=True, help_text="Primary SEO keyword")
secondary_keywords = models.JSONField(default=list, blank=True, help_text="Secondary SEO keywords")
cluster = models.ForeignKey(
'planner.Clusters',
on_delete=models.SET_NULL,
null=True,
blank=False,
related_name='contents',
help_text="Parent cluster (required)"
)
content_type = models.CharField(
max_length=50,
choices=CONTENT_TYPE_CHOICES,
default='post',
db_index=True,
help_text="Content type: post, page, product, taxonomy"
)
content_structure = models.CharField(
max_length=50,
choices=CONTENT_STRUCTURE_CHOICES,
default='article',
db_index=True,
help_text="Content structure/format based on content type"
)
# Taxonomy relationships
taxonomy_terms = models.ManyToManyField(
'ContentTaxonomy',
through='ContentTaxonomyRelation',
blank=True,
related_name='contents',
help_text="Associated taxonomy terms (categories, tags, attributes)"
)
# External platform fields (WordPress integration)
external_id = models.CharField(max_length=255, blank=True, null=True, db_index=True, help_text="WordPress/external platform post ID")
external_url = models.URLField(blank=True, null=True, help_text="WordPress/external platform URL")
external_type = models.CharField(max_length=100, blank=True, null=True, help_text="WordPress post type (post, page, product, etc.)")
external_metadata = models.JSONField(blank=True, null=True, default=dict, help_text="External platform metadata (WordPress term IDs, etc.)")
sync_status = models.CharField(max_length=50, blank=True, null=True, help_text="Sync status with WordPress")
# Source tracking
SOURCE_CHOICES = [
('igny8', 'IGNY8 Generated'),
('wordpress', 'WordPress Imported'),
]
source = models.CharField(
max_length=50,
choices=SOURCE_CHOICES,
default='igny8',
db_index=True,
help_text="Content source"
)
# Status tracking
STATUS_CHOICES = [
('draft', 'Draft'),
('review', 'Review'),
('approved', 'Approved'), # Ready for publishing to external site
('published', 'Published'), # Actually published on external site
]
status = models.CharField(
max_length=50,
choices=STATUS_CHOICES,
default='draft',
db_index=True,
help_text="Content status"
)
# Publishing scheduler fields
SITE_STATUS_CHOICES = [
('not_published', 'Not Published'),
('scheduled', 'Scheduled'),
('publishing', 'Publishing'),
('published', 'Published'),
('failed', 'Failed'),
]
site_status = models.CharField(
max_length=50,
choices=SITE_STATUS_CHOICES,
default='not_published',
db_index=True,
help_text="External site publishing status"
)
scheduled_publish_at = models.DateTimeField(
null=True,
blank=True,
db_index=True,
help_text="Scheduled time for publishing to external site"
)
site_status_updated_at = models.DateTimeField(
null=True,
blank=True,
help_text="Last time site_status was changed"
)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
app_label = 'writer'
db_table = 'igny8_content'
ordering = ['-created_at']
verbose_name = 'Content'
verbose_name_plural = 'Contents'
indexes = [
models.Index(fields=['title']),
models.Index(fields=['cluster']),
models.Index(fields=['content_type']),
models.Index(fields=['content_structure']),
models.Index(fields=['source']),
models.Index(fields=['status']),
models.Index(fields=['external_id']),
models.Index(fields=['site', 'sector']),
]
objects = SoftDeleteManager()
all_objects = models.Manager()
def __str__(self):
return self.title or f"Content {self.id}"
def save(self, *args, **kwargs):
"""Override save to auto-calculate word_count from content_html"""
is_new = self.pk is None
old_word_count = 0
# Get old word count if updating
if not is_new and self.content_html:
try:
old_instance = Content.objects.get(pk=self.pk)
old_word_count = old_instance.word_count or 0
except Content.DoesNotExist:
pass
# Auto-calculate word count if content_html has changed
if self.content_html:
from igny8_core.utils.word_counter import calculate_word_count
calculated_count = calculate_word_count(self.content_html)
# Only update if different to avoid unnecessary saves
if self.word_count != calculated_count:
self.word_count = calculated_count
super().save(*args, **kwargs)
# Increment usage for new content or if word count increased
if self.content_html and self.word_count:
# Only count newly generated words
new_words = self.word_count - old_word_count if not is_new else self.word_count
if new_words > 0:
from igny8_core.business.billing.services.limit_service import LimitService
try:
# Get account from site
account = self.site.account if self.site else None
if account:
LimitService.increment_usage(
account=account,
limit_type='content_words',
amount=new_words,
metadata={
'content_id': self.id,
'content_title': self.title,
'site_id': self.site.id if self.site else None,
}
)
except Exception as e:
import logging
logger = logging.getLogger(__name__)
logger.error(f"Error incrementing word usage for content {self.id}: {str(e)}")
def soft_delete(self, user=None, reason=None, retention_days=None):
"""
Override soft_delete to cascade to related models.
This ensures Images, ContentClusterMap, ContentAttribute are also deleted.
"""
import logging
logger = logging.getLogger(__name__)
# Soft-delete related Images (which are also SoftDeletable)
related_images = self.images.filter(is_deleted=False)
images_count = related_images.count()
for image in related_images:
image.soft_delete(user=user, reason=f"Parent content deleted: {reason or 'No reason'}")
# Hard-delete ContentClusterMap (not soft-deletable)
cluster_maps_count = self.cluster_mappings.count()
self.cluster_mappings.all().delete()
# Hard-delete ContentAttribute (not soft-deletable)
attributes_count = self.attributes.count()
self.attributes.all().delete()
# Hard-delete ContentTaxonomyRelation (through model for many-to-many)
taxonomy_relations_count = ContentTaxonomyRelation.objects.filter(content=self).count()
ContentTaxonomyRelation.objects.filter(content=self).delete()
logger.info(
f"[Content.soft_delete] Content {self.id} '{self.title}' cascade delete: "
f"{images_count} images, {cluster_maps_count} cluster maps, "
f"{attributes_count} attributes, {taxonomy_relations_count} taxonomy relations"
)
# Call parent soft_delete
super().soft_delete(user=user, reason=reason, retention_days=retention_days)
def hard_delete(self, using=None, keep_parents=False):
"""
Override hard_delete to cascade to related models.
Django CASCADE should handle this, but we explicitly clean up for safety.
"""
import logging
logger = logging.getLogger(__name__)
# Hard-delete related Images (including soft-deleted ones)
images_count = Images.all_objects.filter(content=self).count()
Images.all_objects.filter(content=self).delete()
logger.info(
f"[Content.hard_delete] Content {self.id} '{self.title}' hard delete: "
f"{images_count} images removed"
)
# Call parent hard_delete (Django CASCADE will handle the rest)
return super().hard_delete(using=using, keep_parents=keep_parents)
class ContentTaxonomy(SiteSectorBaseModel):
"""
Simplified taxonomy model for AI-generated categories and tags.
Directly linked to Content via many-to-many relationship.
"""
TAXONOMY_TYPE_CHOICES = [
('category', 'Category'),
('tag', 'Tag'),
]
name = models.CharField(max_length=255, db_index=True, help_text="Term name")
slug = models.SlugField(max_length=255, db_index=True, help_text="URL slug")
taxonomy_type = models.CharField(
max_length=50,
choices=TAXONOMY_TYPE_CHOICES,
db_index=True,
help_text="Type of taxonomy (category or tag)"
)
# WordPress/external platform sync fields
external_taxonomy = models.CharField(
max_length=100,
blank=True,
default='',
help_text="WordPress taxonomy slug (category, post_tag)"
)
external_id = models.IntegerField(
null=True,
blank=True,
db_index=True,
help_text="WordPress term_id for sync"
)
sync_status = models.CharField(
max_length=50,
blank=True,
default='',
help_text="Sync status with external platform"
)
description = models.TextField(
blank=True,
default='',
help_text="Taxonomy term description"
)
count = models.IntegerField(
default=0,
help_text="Number of times this term is used"
)
metadata = models.JSONField(
default=dict,
blank=True,
help_text="Additional metadata (AI generation details, etc.)"
)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
app_label = 'writer'
db_table = 'igny8_content_taxonomy_terms'
verbose_name = 'Content Taxonomy'
verbose_name_plural = 'Content Taxonomies'
unique_together = [
['site', 'slug', 'taxonomy_type'],
['site', 'external_id', 'external_taxonomy'],
]
indexes = [
models.Index(fields=['name']),
models.Index(fields=['slug']),
models.Index(fields=['taxonomy_type']),
models.Index(fields=['external_id', 'external_taxonomy']),
models.Index(fields=['site', 'taxonomy_type']),
models.Index(fields=['site', 'sector']),
]
def __str__(self):
return f"{self.name} ({self.get_taxonomy_type_display()})"
class Images(SoftDeletableModel, SiteSectorBaseModel):
"""Images model for content-related images (featured, desktop, mobile, in-article)"""
IMAGE_TYPE_CHOICES = [
('featured', 'Featured Image'),
('desktop', 'Desktop Image'),
('mobile', 'Mobile Image'),
('in_article', 'In-Article Image'),
]
content = models.ForeignKey(
Content,
on_delete=models.CASCADE,
related_name='images',
null=True,
blank=True,
help_text="The content this image belongs to (preferred)"
)
task = models.ForeignKey(
Tasks,
on_delete=models.CASCADE,
related_name='images',
null=True,
blank=True,
help_text="The task this image belongs to (legacy, use content instead)"
)
image_type = models.CharField(max_length=50, choices=IMAGE_TYPE_CHOICES, default='featured')
image_url = models.CharField(max_length=500, blank=True, null=True, help_text="URL of the generated/stored image")
image_path = models.CharField(max_length=500, blank=True, null=True, help_text="Local path if stored locally")
prompt = models.TextField(blank=True, null=True, help_text="Image generation prompt used")
caption = models.TextField(blank=True, null=True, help_text="Image caption (40-60 words) to display with the image")
status = models.CharField(max_length=50, default='pending', help_text="Status: pending, generated, failed")
position = models.IntegerField(default=0, help_text="Position for in-article images ordering")
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
app_label = 'writer'
db_table = 'igny8_images'
ordering = ['content', 'position', '-created_at']
verbose_name = 'Image'
verbose_name_plural = 'Images'
indexes = [
models.Index(fields=['content', 'image_type']),
models.Index(fields=['task', 'image_type']),
models.Index(fields=['status']),
models.Index(fields=['content', 'position']),
models.Index(fields=['task', 'position']),
]
# Ensure unique position per content+image_type combination
constraints = [
models.UniqueConstraint(
fields=['content', 'image_type', 'position'],
name='unique_content_image_type_position',
condition=models.Q(is_deleted=False)
),
]
objects = SoftDeleteManager()
all_objects = models.Manager()
@property
def aspect_ratio(self):
"""
Determine aspect ratio based on position for layout rendering.
Position 0, 2: square (1:1)
Position 1, 3: landscape (16:9 or similar)
Featured: always landscape
"""
if self.image_type == 'featured':
return 'landscape'
elif self.image_type == 'in_article':
# Even positions are square, odd positions are landscape
return 'square' if (self.position or 0) % 2 == 0 else 'landscape'
return 'square' # Default
def save(self, *args, **kwargs):
"""Track image usage when creating new images"""
is_new = self.pk is None
# Automatically set account, site, and sector from content or task
if self.content:
self.account = self.content.account
self.site = self.content.site
self.sector = self.content.sector
elif self.task:
self.account = self.task.account
self.site = self.task.site
self.sector = self.task.sector
super().save(*args, **kwargs)
# Increment usage for new images
if is_new:
from igny8_core.business.billing.services.limit_service import LimitService
try:
account = self.account
if account:
# Track image prompt usage
if self.prompt:
LimitService.increment_usage(
account=account,
limit_type='image_prompts',
amount=1,
metadata={
'image_id': self.id,
'image_type': self.image_type,
'content_id': self.content.id if self.content else None,
}
)
# Track basic image usage (for now, all images are counted as basic)
# TODO: Implement premium image tracking when premium models are used
LimitService.increment_usage(
account=account,
limit_type='images_basic',
amount=1,
metadata={
'image_id': self.id,
'image_type': self.image_type,
'content_id': self.content.id if self.content else None,
}
)
except Exception as e:
import logging
logger = logging.getLogger(__name__)
logger.error(f"Error incrementing image usage for image {self.id}: {str(e)}")
def __str__(self):
content_title = self.content.title if self.content else None
task_title = self.task.title if self.task else None
title = content_title or task_title or 'Unknown'
return f"{title} - {self.image_type}"
class ContentClusterMap(SiteSectorBaseModel):
"""Associates generated content with planner clusters + roles."""
ROLE_CHOICES = [
('hub', 'Hub Page'),
('supporting', 'Supporting Page'),
('attribute', 'Attribute Page'),
]
SOURCE_CHOICES = [
('blueprint', 'Blueprint'),
('manual', 'Manual'),
('import', 'Import'),
]
content = models.ForeignKey(
Content,
on_delete=models.CASCADE,
related_name='cluster_mappings',
null=True,
blank=True,
)
task = models.ForeignKey(
Tasks,
on_delete=models.CASCADE,
related_name='cluster_mappings',
null=True,
blank=True,
)
cluster = models.ForeignKey(
'planner.Clusters',
on_delete=models.CASCADE,
related_name='content_mappings',
)
role = models.CharField(max_length=50, choices=ROLE_CHOICES, default='hub')
source = models.CharField(max_length=50, choices=SOURCE_CHOICES, default='blueprint')
metadata = models.JSONField(default=dict, blank=True)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
app_label = 'writer'
db_table = 'igny8_content_cluster_map'
unique_together = [['content', 'cluster', 'role']]
indexes = [
models.Index(fields=['cluster', 'role']),
models.Index(fields=['content', 'role']),
models.Index(fields=['task', 'role']),
]
def save(self, *args, **kwargs):
provider = self.content or self.task
if provider:
self.account = provider.account
self.site = provider.site
self.sector = provider.sector
super().save(*args, **kwargs)
def __str__(self):
return f"{self.cluster.name} ({self.get_role_display()})"
class ContentAttribute(SiteSectorBaseModel):
"""
Unified attribute storage for products, services, and semantic facets.
Replaces ContentAttributeMap with enhanced WP sync support.
"""
ATTRIBUTE_TYPE_CHOICES = [
('product_spec', 'Product Specification'),
('service_modifier', 'Service Modifier'),
('semantic_facet', 'Semantic Facet'),
]
SOURCE_CHOICES = [
('blueprint', 'Blueprint'),
('manual', 'Manual'),
('import', 'Import'),
('wordpress', 'WordPress'),
]
content = models.ForeignKey(
Content,
on_delete=models.CASCADE,
related_name='attributes',
null=True,
blank=True,
)
task = models.ForeignKey(
Tasks,
on_delete=models.CASCADE,
related_name='attribute_mappings',
null=True,
blank=True,
)
cluster = models.ForeignKey(
'planner.Clusters',
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name='attributes',
help_text="Optional cluster association for semantic attributes"
)
attribute_type = models.CharField(
max_length=50,
choices=ATTRIBUTE_TYPE_CHOICES,
default='product_spec',
db_index=True,
help_text="Type of attribute"
)
name = models.CharField(max_length=120, help_text="Attribute name (e.g., Color, Material)")
value = models.CharField(max_length=255, blank=True, null=True, help_text="Attribute value (e.g., Blue, Cotton)")
# WordPress/WooCommerce sync fields
external_id = models.IntegerField(null=True, blank=True, help_text="WP attribute term ID")
external_attribute_name = models.CharField(
max_length=100,
blank=True,
help_text="WP attribute slug (e.g., pa_color, pa_size)"
)
source = models.CharField(max_length=50, choices=SOURCE_CHOICES, default='manual')
metadata = models.JSONField(default=dict, blank=True, help_text="Additional metadata")
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
app_label = 'writer'
db_table = 'igny8_content_attributes'
verbose_name = 'Content Attribute'
verbose_name_plural = 'Content Attributes'
indexes = [
models.Index(fields=['name']),
models.Index(fields=['attribute_type']),
models.Index(fields=['content', 'name']),
models.Index(fields=['content', 'attribute_type']),
models.Index(fields=['cluster', 'attribute_type']),
models.Index(fields=['external_id']),
]
def save(self, *args, **kwargs):
provider = self.content or self.task
if provider:
self.account = provider.account
self.site = provider.site
self.sector = provider.sector
super().save(*args, **kwargs)
def __str__(self):
return f"{self.name}: {self.value}"
# Backward compatibility alias
ContentAttributeMap = ContentAttribute
class ImagePrompts(Images):
"""
Proxy model for Images to provide a separate admin interface focused on prompts.
This allows a dedicated "Image Prompts" view in the admin sidebar.
"""
class Meta:
proxy = True
verbose_name = 'Image Prompt'
verbose_name_plural = 'Image Prompts'
app_label = 'writer'