old automation cleanup adn status feilds of planner udpate
This commit is contained in:
IGNY8 VPS (Salman)
@@ -1,4 +0,0 @@
|
||||
"""
|
||||
Automation business logic - AutomationRule, ScheduledTask models and services
|
||||
"""
|
||||
|
||||
@@ -1,143 +0,0 @@
|
||||
"""
|
||||
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}"
|
||||
|
||||
@@ -1,4 +0,0 @@
|
||||
"""
|
||||
Automation services
|
||||
"""
|
||||
|
||||
@@ -1,101 +0,0 @@
|
||||
"""
|
||||
Action Executor
|
||||
Executes rule actions
|
||||
"""
|
||||
import logging
|
||||
from igny8_core.business.planning.services.clustering_service import ClusteringService
|
||||
from igny8_core.business.planning.services.ideas_service import IdeasService
|
||||
from igny8_core.business.content.services.content_generation_service import ContentGenerationService
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class ActionExecutor:
|
||||
"""Executes rule actions"""
|
||||
|
||||
def __init__(self):
|
||||
self.clustering_service = ClusteringService()
|
||||
self.ideas_service = IdeasService()
|
||||
self.content_service = ContentGenerationService()
|
||||
|
||||
def execute(self, action, context, rule):
|
||||
"""
|
||||
Execute a single action.
|
||||
|
||||
Args:
|
||||
action: Action dict with 'type' and 'params'
|
||||
context: Context dict
|
||||
rule: AutomationRule instance
|
||||
|
||||
Returns:
|
||||
dict: Action execution result
|
||||
"""
|
||||
action_type = action.get('type')
|
||||
params = action.get('params', {})
|
||||
|
||||
if action_type == 'cluster_keywords':
|
||||
return self._execute_cluster_keywords(params, rule)
|
||||
elif action_type == 'generate_ideas':
|
||||
return self._execute_generate_ideas(params, rule)
|
||||
elif action_type == 'generate_content':
|
||||
return self._execute_generate_content(params, rule)
|
||||
else:
|
||||
logger.warning(f"Unknown action type: {action_type}")
|
||||
return {
|
||||
'success': False,
|
||||
'error': f'Unknown action type: {action_type}'
|
||||
}
|
||||
|
||||
def _execute_cluster_keywords(self, params, rule):
|
||||
"""Execute cluster keywords action"""
|
||||
keyword_ids = params.get('keyword_ids', [])
|
||||
sector_id = params.get('sector_id') or (rule.sector.id if rule.sector else None)
|
||||
|
||||
try:
|
||||
result = self.clustering_service.cluster_keywords(
|
||||
keyword_ids=keyword_ids,
|
||||
account=rule.account,
|
||||
sector_id=sector_id
|
||||
)
|
||||
return result
|
||||
except Exception as e:
|
||||
logger.error(f"Error clustering keywords: {str(e)}", exc_info=True)
|
||||
return {
|
||||
'success': False,
|
||||
'error': str(e)
|
||||
}
|
||||
|
||||
def _execute_generate_ideas(self, params, rule):
|
||||
"""Execute generate ideas action"""
|
||||
cluster_ids = params.get('cluster_ids', [])
|
||||
|
||||
try:
|
||||
result = self.ideas_service.generate_ideas(
|
||||
cluster_ids=cluster_ids,
|
||||
account=rule.account
|
||||
)
|
||||
return result
|
||||
except Exception as e:
|
||||
logger.error(f"Error generating ideas: {str(e)}", exc_info=True)
|
||||
return {
|
||||
'success': False,
|
||||
'error': str(e)
|
||||
}
|
||||
|
||||
def _execute_generate_content(self, params, rule):
|
||||
"""Execute generate content action"""
|
||||
task_ids = params.get('task_ids', [])
|
||||
|
||||
try:
|
||||
result = self.content_service.generate_content(
|
||||
task_ids=task_ids,
|
||||
account=rule.account
|
||||
)
|
||||
return result
|
||||
except Exception as e:
|
||||
logger.error(f"Error generating content: {str(e)}", exc_info=True)
|
||||
return {
|
||||
'success': False,
|
||||
'error': str(e)
|
||||
}
|
||||
|
||||
@@ -1,141 +0,0 @@
|
||||
"""
|
||||
Automation Service
|
||||
Main service for executing automation rules
|
||||
"""
|
||||
import logging
|
||||
from django.utils import timezone
|
||||
from igny8_core.business.automation.models import AutomationRule, ScheduledTask
|
||||
from igny8_core.business.automation.services.rule_engine import RuleEngine
|
||||
from igny8_core.business.billing.services.credit_service import CreditService
|
||||
from igny8_core.business.billing.exceptions import InsufficientCreditsError
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class AutomationService:
|
||||
"""Service for executing automation rules"""
|
||||
|
||||
def __init__(self):
|
||||
self.rule_engine = RuleEngine()
|
||||
self.credit_service = CreditService()
|
||||
|
||||
def execute_rule(self, rule, context=None):
|
||||
"""
|
||||
Execute an automation rule.
|
||||
|
||||
Args:
|
||||
rule: AutomationRule instance
|
||||
context: Optional context dict for condition evaluation
|
||||
|
||||
Returns:
|
||||
dict: Execution result with status and data
|
||||
"""
|
||||
if not rule.is_active or rule.status != 'active':
|
||||
return {
|
||||
'status': 'skipped',
|
||||
'reason': 'Rule is inactive',
|
||||
'rule_id': rule.id
|
||||
}
|
||||
|
||||
# Check credits (estimate based on actions)
|
||||
estimated_credits = self._estimate_credits(rule)
|
||||
try:
|
||||
self.credit_service.check_credits_legacy(rule.account, estimated_credits)
|
||||
except InsufficientCreditsError as e:
|
||||
logger.warning(f"Rule {rule.id} skipped: {str(e)}")
|
||||
return {
|
||||
'status': 'skipped',
|
||||
'reason': f'Insufficient credits: {str(e)}',
|
||||
'rule_id': rule.id
|
||||
}
|
||||
|
||||
# Execute via rule engine
|
||||
try:
|
||||
result = self.rule_engine.execute(rule, context or {})
|
||||
|
||||
# Update rule tracking
|
||||
rule.last_executed_at = timezone.now()
|
||||
rule.execution_count += 1
|
||||
rule.save(update_fields=['last_executed_at', 'execution_count'])
|
||||
|
||||
return {
|
||||
'status': 'completed',
|
||||
'rule_id': rule.id,
|
||||
'result': result
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"Error executing rule {rule.id}: {str(e)}", exc_info=True)
|
||||
return {
|
||||
'status': 'failed',
|
||||
'reason': str(e),
|
||||
'rule_id': rule.id
|
||||
}
|
||||
|
||||
def _estimate_credits(self, rule):
|
||||
"""Estimate credits needed for rule execution"""
|
||||
# Simple estimation based on action types
|
||||
estimated = 0
|
||||
for action in rule.actions:
|
||||
action_type = action.get('type', '')
|
||||
if 'cluster' in action_type:
|
||||
estimated += 10
|
||||
elif 'idea' in action_type:
|
||||
estimated += 15
|
||||
elif 'content' in action_type:
|
||||
estimated += 50 # Conservative estimate
|
||||
else:
|
||||
estimated += 5 # Default
|
||||
return max(estimated, 10) # Minimum 10 credits
|
||||
|
||||
def execute_scheduled_rules(self):
|
||||
"""
|
||||
Execute all scheduled rules that are due.
|
||||
Called by Celery Beat task.
|
||||
|
||||
Returns:
|
||||
dict: Summary of executions
|
||||
"""
|
||||
from django.utils import timezone
|
||||
now = timezone.now()
|
||||
|
||||
# Get active scheduled rules
|
||||
rules = AutomationRule.objects.filter(
|
||||
trigger='schedule',
|
||||
is_active=True,
|
||||
status='active'
|
||||
)
|
||||
|
||||
executed = 0
|
||||
skipped = 0
|
||||
failed = 0
|
||||
|
||||
for rule in rules:
|
||||
# Check if rule should execute based on schedule
|
||||
if self._should_execute_schedule(rule, now):
|
||||
result = self.execute_rule(rule)
|
||||
if result['status'] == 'completed':
|
||||
executed += 1
|
||||
elif result['status'] == 'skipped':
|
||||
skipped += 1
|
||||
else:
|
||||
failed += 1
|
||||
|
||||
return {
|
||||
'executed': executed,
|
||||
'skipped': skipped,
|
||||
'failed': failed,
|
||||
'total': len(rules)
|
||||
}
|
||||
|
||||
def _should_execute_schedule(self, rule, now):
|
||||
"""
|
||||
Check if a scheduled rule should execute now.
|
||||
Simple implementation - can be enhanced with proper cron parsing.
|
||||
"""
|
||||
if not rule.schedule:
|
||||
return False
|
||||
|
||||
# For now, simple check - can be enhanced with cron parser
|
||||
# This is a placeholder - proper implementation would parse cron string
|
||||
return True # Simplified for now
|
||||
|
||||
@@ -1,104 +0,0 @@
|
||||
"""
|
||||
Condition Evaluator
|
||||
Evaluates rule conditions
|
||||
"""
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class ConditionEvaluator:
|
||||
"""Evaluates rule conditions"""
|
||||
|
||||
OPERATORS = {
|
||||
'equals': lambda a, b: a == b,
|
||||
'not_equals': lambda a, b: a != b,
|
||||
'greater_than': lambda a, b: a > b,
|
||||
'greater_than_or_equal': lambda a, b: a >= b,
|
||||
'less_than': lambda a, b: a < b,
|
||||
'less_than_or_equal': lambda a, b: a <= b,
|
||||
'in': lambda a, b: a in b,
|
||||
'contains': lambda a, b: b in a if isinstance(a, str) else a in b,
|
||||
'is_empty': lambda a, b: not a or (isinstance(a, str) and not a.strip()),
|
||||
'is_not_empty': lambda a, b: a and (not isinstance(a, str) or a.strip()),
|
||||
}
|
||||
|
||||
def evaluate(self, conditions, context):
|
||||
"""
|
||||
Evaluate a list of conditions.
|
||||
|
||||
Args:
|
||||
conditions: List of condition dicts
|
||||
context: Context dict for field resolution
|
||||
|
||||
Returns:
|
||||
bool: True if all conditions are met
|
||||
"""
|
||||
if not conditions:
|
||||
return True
|
||||
|
||||
for condition in conditions:
|
||||
if not self._evaluate_condition(condition, context):
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
def _evaluate_condition(self, condition, context):
|
||||
"""
|
||||
Evaluate a single condition.
|
||||
|
||||
Condition format:
|
||||
{
|
||||
"field": "content.status",
|
||||
"operator": "equals",
|
||||
"value": "draft"
|
||||
}
|
||||
"""
|
||||
field_path = condition.get('field')
|
||||
operator = condition.get('operator', 'equals')
|
||||
expected_value = condition.get('value')
|
||||
|
||||
if not field_path:
|
||||
logger.warning("Condition missing 'field'")
|
||||
return False
|
||||
|
||||
# Resolve field value from context
|
||||
actual_value = self._resolve_field(field_path, context)
|
||||
|
||||
# Get operator function
|
||||
op_func = self.OPERATORS.get(operator)
|
||||
if not op_func:
|
||||
logger.warning(f"Unknown operator: {operator}")
|
||||
return False
|
||||
|
||||
# Evaluate
|
||||
try:
|
||||
return op_func(actual_value, expected_value)
|
||||
except Exception as e:
|
||||
logger.error(f"Error evaluating condition: {str(e)}", exc_info=True)
|
||||
return False
|
||||
|
||||
def _resolve_field(self, field_path, context):
|
||||
"""
|
||||
Resolve a field path from context.
|
||||
|
||||
Examples:
|
||||
- "content.status" -> context['content']['status']
|
||||
- "count" -> context['count']
|
||||
"""
|
||||
parts = field_path.split('.')
|
||||
value = context
|
||||
|
||||
for part in parts:
|
||||
if isinstance(value, dict):
|
||||
value = value.get(part)
|
||||
elif hasattr(value, part):
|
||||
value = getattr(value, part)
|
||||
else:
|
||||
return None
|
||||
|
||||
if value is None:
|
||||
return None
|
||||
|
||||
return value
|
||||
|
||||
@@ -1,61 +0,0 @@
|
||||
"""
|
||||
Rule Engine
|
||||
Orchestrates rule execution
|
||||
"""
|
||||
import logging
|
||||
from igny8_core.business.automation.services.condition_evaluator import ConditionEvaluator
|
||||
from igny8_core.business.automation.services.action_executor import ActionExecutor
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class RuleEngine:
|
||||
"""Orchestrates rule execution"""
|
||||
|
||||
def __init__(self):
|
||||
self.condition_evaluator = ConditionEvaluator()
|
||||
self.action_executor = ActionExecutor()
|
||||
|
||||
def execute(self, rule, context):
|
||||
"""
|
||||
Execute a rule by evaluating conditions and executing actions.
|
||||
|
||||
Args:
|
||||
rule: AutomationRule instance
|
||||
context: Context dict for evaluation
|
||||
|
||||
Returns:
|
||||
dict: Execution results
|
||||
"""
|
||||
# Evaluate conditions
|
||||
if rule.conditions:
|
||||
conditions_met = self.condition_evaluator.evaluate(rule.conditions, context)
|
||||
if not conditions_met:
|
||||
return {
|
||||
'success': False,
|
||||
'reason': 'Conditions not met'
|
||||
}
|
||||
|
||||
# Execute actions
|
||||
action_results = []
|
||||
for action in rule.actions:
|
||||
try:
|
||||
result = self.action_executor.execute(action, context, rule)
|
||||
action_results.append({
|
||||
'action': action,
|
||||
'success': True,
|
||||
'result': result
|
||||
})
|
||||
except Exception as e:
|
||||
logger.error(f"Action execution failed: {str(e)}", exc_info=True)
|
||||
action_results.append({
|
||||
'action': action,
|
||||
'success': False,
|
||||
'error': str(e)
|
||||
})
|
||||
|
||||
return {
|
||||
'success': True,
|
||||
'actions': action_results
|
||||
}
|
||||
|
||||
@@ -1,28 +0,0 @@
|
||||
"""
|
||||
Automation Celery Tasks
|
||||
"""
|
||||
from celery import shared_task
|
||||
import logging
|
||||
from igny8_core.business.automation.services.automation_service import AutomationService
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@shared_task(name='igny8_core.business.automation.tasks.execute_scheduled_automation_rules')
|
||||
def execute_scheduled_automation_rules():
|
||||
"""
|
||||
Execute all scheduled automation rules.
|
||||
Called by Celery Beat.
|
||||
"""
|
||||
try:
|
||||
service = AutomationService()
|
||||
result = service.execute_scheduled_rules()
|
||||
logger.info(f"Executed scheduled automation rules: {result}")
|
||||
return result
|
||||
except Exception as e:
|
||||
logger.error(f"Error executing scheduled automation rules: {str(e)}", exc_info=True)
|
||||
return {
|
||||
'success': False,
|
||||
'error': str(e)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user