""" Script model for user-created Python scripts. """ import secrets import uuid from django.conf import settings from django.db import models from .environment import Environment class Script(models.Model): """ Represents a Python script that can be executed. Scripts are associated with an environment and can be run manually or on schedule. """ id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) name = models.CharField(max_length=200) description = models.TextField(blank=True) # The actual Python code code = models.TextField(help_text="Python code to execute") # Execution settings environment = models.ForeignKey( Environment, on_delete=models.PROTECT, related_name="scripts", help_text="Python environment to use for execution", ) # Tags for categorization tags = models.ManyToManyField( "Tag", blank=True, related_name="scripts", help_text="Tags for organizing and filtering scripts", ) timeout_seconds = models.PositiveIntegerField( default=3600, # 1 hour default help_text="Maximum execution time in seconds (default: 1 hour, max: 24 hours)", ) # Status is_enabled = models.BooleanField( default=True, help_text="Whether this script can be executed", ) # Webhook webhook_token = models.CharField( max_length=64, unique=True, null=True, blank=True, db_index=True, help_text="Unique token for webhook URL (auto-generated)", ) # Notification settings class NotifyOn(models.TextChoices): NEVER = "never", "Never" FAILURE = "failure", "On Failure" SUCCESS = "success", "On Success" BOTH = "both", "On Success and Failure" notify_on = models.CharField( max_length=20, choices=NotifyOn.choices, default=NotifyOn.NEVER, help_text="When to send notifications for this script", ) notify_email = models.EmailField( blank=True, help_text="Override email for this script (uses global default if empty)", ) notify_webhook_url = models.URLField( blank=True, max_length=500, help_text="URL to POST notification webhooks to", ) notify_webhook_enabled = models.BooleanField( default=False, help_text="Enable webhook notifications for this script", ) # Retention overrides (null = use global settings) retention_days_override = models.PositiveIntegerField( null=True, blank=True, help_text="Override global retention days for this script", ) retention_count_override = models.PositiveIntegerField( null=True, blank=True, help_text="Override global retention count for this script", ) # Archive fields (soft delete) archived_at = models.DateTimeField( null=True, blank=True, help_text="When this script was archived (null = not archived)", ) archived_by = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="archived_scripts", ) # Metadata created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) created_by = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="scripts", ) class Meta: db_table = "scripts" verbose_name = "script" verbose_name_plural = "scripts" ordering = ["-updated_at"] def __str__(self): if self.is_archived: return f"{self.name} (archived)" status = "enabled" if self.is_enabled else "disabled" return f"{self.name} ({status})" @property def is_archived(self) -> bool: """Check if this script is archived.""" return self.archived_at is not None @property def can_run(self) -> bool: """Check if this script can be executed (enabled and not archived).""" return self.is_enabled and not self.is_archived @property def last_run(self): """Return the most recent run for this script.""" return self.runs.order_by("-created_at").first() @property def last_successful_run(self): """Return the most recent successful run for this script.""" return self.runs.filter(status="success").order_by("-created_at").first() @property def run_count(self) -> int: """Return the total number of runs for this script.""" return self.runs.count() @property def success_rate(self) -> float | None: """Return the success rate as a percentage, or None if no runs.""" total = self.run_count if total == 0: return None successful = self.runs.filter(status="success").count() return (successful / total) * 100 def get_code_preview(self, max_lines: int = 5) -> str: """Return a preview of the script code (first N lines).""" lines = self.code.split("\n")[:max_lines] preview = "\n".join(lines) if len(self.code.split("\n")) > max_lines: preview += "\n..." return preview @staticmethod def generate_webhook_token() -> str: """Generate a secure random webhook token (64 chars, URL-safe).""" return secrets.token_urlsafe(48) # 48 bytes = 64 chars in base64 def create_webhook_token(self) -> str: """Create and save a new webhook token for this script.""" self.webhook_token = self.generate_webhook_token() self.save(update_fields=["webhook_token", "updated_at"]) return self.webhook_token def regenerate_webhook_token(self) -> str: """Regenerate the webhook token, invalidating the old one.""" return self.create_webhook_token() def clear_webhook_token(self) -> None: """Remove the webhook token, disabling webhook access.""" self.webhook_token = None self.save(update_fields=["webhook_token", "updated_at"]) @property def has_webhook(self) -> bool: """Check if this script has a webhook token configured.""" return bool(self.webhook_token)