1c2c9354ba
- Registered the new AutomationConfig in the INSTALLED_APPS of settings.py. - Set the app_label for AutomationRule and ScheduledTask models to 'automation' for better organization and clarity in the database schema.
144 lines
4.8 KiB
Python
144 lines
4.8 KiB
Python
"""
|
|
Automation Models
|
|
Phase 2: Automation System
|
|
"""
|
|
from django.db import models
|
|
from django.core.validators import MinValueValidator
|
|
from igny8_core.auth.models import SiteSectorBaseModel, AccountBaseModel
|
|
import json
|
|
|
|
|
|
class AutomationRule(SiteSectorBaseModel):
|
|
"""
|
|
Automation Rule model for defining automated workflows.
|
|
|
|
Rules can be triggered by:
|
|
- schedule: Time-based triggers (cron-like)
|
|
- event: Event-based triggers (content created, keyword added, etc.)
|
|
- manual: Manual execution only
|
|
"""
|
|
|
|
TRIGGER_CHOICES = [
|
|
('schedule', 'Schedule'),
|
|
('event', 'Event'),
|
|
('manual', 'Manual'),
|
|
]
|
|
|
|
STATUS_CHOICES = [
|
|
('active', 'Active'),
|
|
('inactive', 'Inactive'),
|
|
('paused', 'Paused'),
|
|
]
|
|
|
|
name = models.CharField(max_length=255, help_text="Rule name")
|
|
description = models.TextField(blank=True, null=True, help_text="Rule description")
|
|
|
|
# Trigger configuration
|
|
trigger = models.CharField(max_length=50, choices=TRIGGER_CHOICES, default='manual')
|
|
|
|
# Schedule configuration (for schedule triggers)
|
|
# Stored as cron-like string: "0 0 * * *" (daily at midnight)
|
|
schedule = models.CharField(
|
|
max_length=100,
|
|
blank=True,
|
|
null=True,
|
|
help_text="Cron-like schedule string (e.g., '0 0 * * *' for daily at midnight)"
|
|
)
|
|
|
|
# Conditions (JSON field)
|
|
# Format: [{"field": "content.status", "operator": "equals", "value": "draft"}, ...]
|
|
conditions = models.JSONField(
|
|
default=list,
|
|
help_text="List of conditions that must be met for rule to execute"
|
|
)
|
|
|
|
# Actions (JSON field)
|
|
# Format: [{"type": "generate_content", "params": {...}}, ...]
|
|
actions = models.JSONField(
|
|
default=list,
|
|
help_text="List of actions to execute when rule triggers"
|
|
)
|
|
|
|
# Status
|
|
is_active = models.BooleanField(default=True, help_text="Whether rule is active")
|
|
status = models.CharField(max_length=50, choices=STATUS_CHOICES, default='active')
|
|
|
|
# Execution tracking
|
|
last_executed_at = models.DateTimeField(null=True, blank=True)
|
|
execution_count = models.IntegerField(default=0, validators=[MinValueValidator(0)])
|
|
|
|
# Metadata
|
|
metadata = models.JSONField(default=dict, help_text="Additional metadata")
|
|
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
updated_at = models.DateTimeField(auto_now=True)
|
|
|
|
class Meta:
|
|
app_label = 'automation'
|
|
db_table = 'igny8_automation_rules'
|
|
ordering = ['-created_at']
|
|
verbose_name = 'Automation Rule'
|
|
verbose_name_plural = 'Automation Rules'
|
|
indexes = [
|
|
models.Index(fields=['trigger', 'is_active']),
|
|
models.Index(fields=['status']),
|
|
models.Index(fields=['site', 'sector']),
|
|
models.Index(fields=['trigger', 'is_active', 'status']),
|
|
]
|
|
|
|
def __str__(self):
|
|
return f"{self.name} ({self.get_trigger_display()})"
|
|
|
|
|
|
class ScheduledTask(AccountBaseModel):
|
|
"""
|
|
Scheduled Task model for tracking scheduled automation rule executions.
|
|
"""
|
|
|
|
STATUS_CHOICES = [
|
|
('pending', 'Pending'),
|
|
('running', 'Running'),
|
|
('completed', 'Completed'),
|
|
('failed', 'Failed'),
|
|
('cancelled', 'Cancelled'),
|
|
]
|
|
|
|
automation_rule = models.ForeignKey(
|
|
AutomationRule,
|
|
on_delete=models.CASCADE,
|
|
related_name='scheduled_tasks',
|
|
help_text="The automation rule this task belongs to"
|
|
)
|
|
|
|
scheduled_at = models.DateTimeField(help_text="When the task is scheduled to run")
|
|
executed_at = models.DateTimeField(null=True, blank=True, help_text="When the task was actually executed")
|
|
|
|
status = models.CharField(max_length=50, choices=STATUS_CHOICES, default='pending')
|
|
|
|
# Execution results
|
|
result = models.JSONField(default=dict, help_text="Execution result data")
|
|
error_message = models.TextField(blank=True, null=True, help_text="Error message if execution failed")
|
|
|
|
# Metadata
|
|
metadata = models.JSONField(default=dict, help_text="Additional metadata")
|
|
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
updated_at = models.DateTimeField(auto_now=True)
|
|
|
|
class Meta:
|
|
app_label = 'automation'
|
|
db_table = 'igny8_scheduled_tasks'
|
|
ordering = ['-scheduled_at']
|
|
verbose_name = 'Scheduled Task'
|
|
verbose_name_plural = 'Scheduled Tasks'
|
|
indexes = [
|
|
models.Index(fields=['automation_rule', 'status']),
|
|
models.Index(fields=['scheduled_at', 'status']),
|
|
models.Index(fields=['account', 'status']),
|
|
models.Index(fields=['status', 'scheduled_at']),
|
|
]
|
|
|
|
def __str__(self):
|
|
return f"Scheduled task for {self.automation_rule.name} at {self.scheduled_at}"
|
|
|