Files
igny8/backend/igny8_core/business/content/models.py
IGNY8 VPS (Salman) c84bb9bc14 backedn
2025-11-22 01:13:25 +00:00

707 lines
23 KiB
Python

from django.db import models
from django.core.validators import MinValueValidator
from igny8_core.auth.models import SiteSectorBaseModel
class Tasks(SiteSectorBaseModel):
"""Tasks model for content generation queue"""
STATUS_CHOICES = [
('queued', 'Queued'),
('in_progress', 'In Progress'),
('completed', 'Completed'),
('failed', 'Failed'),
]
ENTITY_TYPE_CHOICES = [
('post', 'Post'),
('page', 'Page'),
('product', 'Product'),
('service', 'Service'),
('taxonomy_term', 'Taxonomy Term'),
]
CLUSTER_ROLE_CHOICES = [
('hub', 'Hub'),
('supporting', 'Supporting'),
('attribute', 'Attribute'),
]
title = models.CharField(max_length=255, db_index=True)
description = models.TextField(blank=True, null=True)
keywords = models.CharField(max_length=500, blank=True) # Comma-separated keywords (legacy)
cluster = models.ForeignKey(
'planner.Clusters',
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name='tasks',
limit_choices_to={'sector': models.F('sector')}
)
keyword_objects = models.ManyToManyField(
'planner.Keywords',
blank=True,
related_name='tasks',
help_text="Individual keywords linked to this task"
)
idea = models.ForeignKey(
'planner.ContentIdeas',
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name='tasks'
)
status = models.CharField(max_length=50, choices=STATUS_CHOICES, default='queued')
entity_type = models.CharField(
max_length=50,
choices=ENTITY_TYPE_CHOICES,
default='post',
db_index=True,
help_text="Type of content entity"
)
taxonomy = models.ForeignKey(
'site_building.SiteBlueprintTaxonomy',
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name='tasks',
help_text="Taxonomy association when derived from blueprint planning"
)
cluster_role = models.CharField(
max_length=50,
choices=CLUSTER_ROLE_CHOICES,
default='hub',
help_text="Role within the cluster-driven sitemap"
)
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=['entity_type']),
models.Index(fields=['cluster_role']),
models.Index(fields=['site', 'sector']),
]
def __str__(self):
return self.title
class Content(SiteSectorBaseModel):
"""
Content model for storing final AI-generated article content.
Separated from Task for content versioning and storage optimization.
"""
task = models.OneToOneField(
Tasks,
on_delete=models.CASCADE,
null=True,
blank=True,
related_name='content_record',
help_text="The task this content belongs to"
)
html_content = models.TextField(help_text="Final AI-generated HTML content")
word_count = models.IntegerField(default=0, validators=[MinValueValidator(0)])
metadata = models.JSONField(default=dict, help_text="Additional metadata (SEO, structure, etc.)")
title = models.CharField(max_length=255, blank=True, null=True)
meta_title = models.CharField(max_length=255, blank=True, null=True)
meta_description = models.TextField(blank=True, null=True)
primary_keyword = models.CharField(max_length=255, blank=True, null=True)
secondary_keywords = models.JSONField(default=list, blank=True, help_text="List of secondary keywords")
STATUS_CHOICES = [
('draft', 'Draft'),
('review', 'Review'),
('publish', 'Publish'),
]
status = models.CharField(max_length=50, choices=STATUS_CHOICES, default='draft', help_text="Content workflow status (draft, review, publish)")
generated_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
# Phase 4: Source tracking
SOURCE_CHOICES = [
('igny8', 'IGNY8 Generated'),
('wordpress', 'WordPress Synced'),
('shopify', 'Shopify Synced'),
('custom', 'Custom API Synced'),
]
source = models.CharField(
max_length=50,
choices=SOURCE_CHOICES,
default='igny8',
db_index=True,
help_text="Source of the content"
)
SYNC_STATUS_CHOICES = [
('native', 'Native IGNY8 Content'),
('imported', 'Imported from External'),
('synced', 'Synced from External'),
]
sync_status = models.CharField(
max_length=50,
choices=SYNC_STATUS_CHOICES,
default='native',
db_index=True,
help_text="Sync status of the content"
)
# External reference fields
external_id = models.CharField(max_length=255, blank=True, null=True, help_text="External platform ID")
external_url = models.URLField(blank=True, null=True, help_text="External platform URL")
sync_metadata = models.JSONField(default=dict, blank=True, help_text="Platform-specific sync metadata")
# Phase 4: Linking fields
internal_links = models.JSONField(default=list, blank=True, help_text="Internal links added by linker")
linker_version = models.IntegerField(default=0, help_text="Version of linker processing")
# Phase 4: Optimization fields
optimizer_version = models.IntegerField(default=0, help_text="Version of optimizer processing")
optimization_scores = models.JSONField(default=dict, blank=True, help_text="Optimization scores (SEO, readability, engagement)")
# Phase 8: Universal Content Types
ENTITY_TYPE_CHOICES = [
('post', 'Blog Post'),
('page', 'Page'),
('product', 'Product'),
('service', 'Service Page'),
('taxonomy_term', 'Taxonomy Term Page'),
# Legacy choices for backward compatibility
('blog_post', 'Blog Post (Legacy)'),
('article', 'Article (Legacy)'),
('taxonomy', 'Taxonomy Page (Legacy)'),
]
entity_type = models.CharField(
max_length=50,
choices=ENTITY_TYPE_CHOICES,
default='post',
db_index=True,
help_text="Type of content entity"
)
# Phase 9: Content format (for posts)
CONTENT_FORMAT_CHOICES = [
('article', 'Article'),
('listicle', 'Listicle'),
('guide', 'How-To Guide'),
('comparison', 'Comparison'),
('review', 'Review'),
('roundup', 'Roundup'),
]
content_format = models.CharField(
max_length=50,
choices=CONTENT_FORMAT_CHOICES,
blank=True,
null=True,
db_index=True,
help_text="Content format (only for entity_type=post)"
)
# Phase 9: Cluster role
CLUSTER_ROLE_CHOICES = [
('hub', 'Hub Page'),
('supporting', 'Supporting Content'),
('attribute', 'Attribute Page'),
]
cluster_role = models.CharField(
max_length=50,
choices=CLUSTER_ROLE_CHOICES,
default='supporting',
blank=True,
null=True,
db_index=True,
help_text="Role within cluster strategy"
)
# Phase 9: WordPress post type
external_type = models.CharField(
max_length=100,
blank=True,
help_text="WordPress post type (post, page, product, service)"
)
# Phase 8: Structured content blocks
json_blocks = models.JSONField(
default=list,
blank=True,
help_text="Structured content blocks (for products, services, taxonomies)"
)
# Phase 8: Content structure data
structure_data = models.JSONField(
default=dict,
blank=True,
help_text="Content structure data (metadata, schema, etc.)"
)
# Phase 9: Taxonomy relationships
taxonomies = models.ManyToManyField(
'ContentTaxonomy',
blank=True,
related_name='contents',
through='ContentTaxonomyRelation',
help_text="Associated taxonomy terms (categories, tags, attributes)"
)
# Phase 9: Direct cluster relationship
cluster = models.ForeignKey(
'planner.Clusters',
on_delete=models.SET_NULL,
null=True,
blank=True,
related_name='contents',
help_text="Primary semantic cluster"
)
class Meta:
app_label = 'writer'
db_table = 'igny8_content'
ordering = ['-generated_at']
verbose_name = 'Content'
verbose_name_plural = 'Contents'
indexes = [
models.Index(fields=['task']),
models.Index(fields=['generated_at']),
models.Index(fields=['source']),
models.Index(fields=['sync_status']),
models.Index(fields=['source', 'sync_status']),
models.Index(fields=['entity_type']),
models.Index(fields=['content_format']),
models.Index(fields=['cluster_role']),
models.Index(fields=['cluster']),
models.Index(fields=['external_type']),
models.Index(fields=['site', 'entity_type']),
]
def save(self, *args, **kwargs):
"""Automatically set account, site, and sector from task"""
if self.task_id: # Check task_id instead of accessing task to avoid RelatedObjectDoesNotExist
try:
self.account = self.task.account
self.site = self.task.site
self.sector = self.task.sector
except self.task.RelatedObjectDoesNotExist:
pass # Task doesn't exist, skip
super().save(*args, **kwargs)
def __str__(self):
return f"Content for {self.task.title}"
class ContentTaxonomy(SiteSectorBaseModel):
"""
Universal taxonomy model for categories, tags, and product attributes.
Syncs with WordPress taxonomies and stores terms.
"""
TAXONOMY_TYPE_CHOICES = [
('category', 'Category'),
('tag', 'Tag'),
('product_cat', 'Product Category'),
('product_tag', 'Product Tag'),
('product_attr', 'Product Attribute'),
('service_cat', 'Service Category'),
]
SYNC_STATUS_CHOICES = [
('native', 'Native IGNY8'),
('imported', 'Imported from External'),
('synced', 'Synced with External'),
]
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"
)
description = models.TextField(blank=True, help_text="Term description")
parent = models.ForeignKey(
'self',
null=True,
blank=True,
on_delete=models.CASCADE,
related_name='children',
help_text="Parent term for hierarchical taxonomies"
)
# WordPress/WooCommerce sync fields
external_id = models.IntegerField(
null=True,
blank=True,
db_index=True,
help_text="WordPress term ID"
)
external_taxonomy = models.CharField(
max_length=100,
blank=True,
help_text="WP taxonomy name (category, post_tag, product_cat, pa_color)"
)
sync_status = models.CharField(
max_length=50,
choices=SYNC_STATUS_CHOICES,
default='native',
db_index=True,
help_text="Sync status with external system"
)
# WordPress metadata
count = models.IntegerField(default=0, help_text="Post/product count from WordPress")
metadata = models.JSONField(default=dict, blank=True, help_text="Additional metadata")
# Cluster mapping
clusters = models.ManyToManyField(
'planner.Clusters',
blank=True,
related_name='taxonomy_terms',
help_text="Semantic clusters this term maps to"
)
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=['sync_status']),
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 ContentTaxonomyRelation(models.Model):
"""
Through model for Content-Taxonomy M2M relationship.
Simplified without SiteSectorBaseModel to avoid tenant_id issues.
"""
content = models.ForeignKey(
Content,
on_delete=models.CASCADE,
related_name='taxonomy_relations'
)
taxonomy = models.ForeignKey(
ContentTaxonomy,
on_delete=models.CASCADE,
related_name='content_relations'
)
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']]
indexes = [
models.Index(fields=['content']),
models.Index(fields=['taxonomy']),
]
def __str__(self):
return f"{self.content}{self.taxonomy}"
class Images(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")
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']),
]
def save(self, *args, **kwargs):
"""Automatically set account, site, and sector from content or task"""
# Prefer content over 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)
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 ContentTaxonomyMap(SiteSectorBaseModel):
"""Maps content entities to blueprint taxonomies for syncing/publishing."""
SOURCE_CHOICES = [
('blueprint', 'Blueprint'),
('manual', 'Manual'),
('import', 'Import'),
]
content = models.ForeignKey(
Content,
on_delete=models.CASCADE,
related_name='taxonomy_mappings',
null=True,
blank=True,
)
task = models.ForeignKey(
Tasks,
on_delete=models.CASCADE,
related_name='taxonomy_mappings',
null=True,
blank=True,
)
taxonomy = models.ForeignKey(
'site_building.SiteBlueprintTaxonomy',
on_delete=models.CASCADE,
related_name='content_mappings',
)
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_taxonomy_map'
unique_together = [['content', 'taxonomy']]
indexes = [
models.Index(fields=['taxonomy']),
models.Index(fields=['content', 'taxonomy']),
models.Index(fields=['task', 'taxonomy']),
]
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.taxonomy.name}"
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