from __future__ import annotations from django.db import models from django.contrib.auth import get_user_model from django.utils import timezone # Remove eager evaluation to avoid accessing settings before configured # User = get_user_model() def get_user_model_cached(): return get_user_model() class RemoteHost(models.Model): AUTH_METHOD_CHOICES = [ ('ssh_key', 'SSH Key'), ('agent', 'SSH Agent'), ('password', 'Password'), # not stored; prompt in future ] name = models.CharField(max_length=100) hostname = models.CharField(max_length=255) port = models.PositiveIntegerField(default=22) username = models.CharField(max_length=100) auth_method = models.CharField(max_length=20, choices=AUTH_METHOD_CHOICES, default='ssh_key') key_path = models.CharField(max_length=500, blank=True) strict_host_key_checking = models.BooleanField(default=True) notes = models.TextField(blank=True) created_at = models.DateTimeField(auto_now_add=True) class Meta: ordering = ['name'] def __str__(self): return f"{self.name} ({self.username}@{self.hostname})" class CommandLog(models.Model): STATUS_CHOICES = [ ('running', 'Running'), ('ok', 'Success'), ('failed', 'Failed'), ('canceled', 'Canceled'), ('error', 'Error'), ] RUN_TYPE_CHOICES = [ ('single','Single'), ('batch','Batch'), ] host = models.ForeignKey(RemoteHost, on_delete=models.CASCADE, related_name='command_logs') created_by = models.ForeignKey('auth.User', null=True, blank=True, on_delete=models.SET_NULL) command = models.TextField() args_json = models.JSONField(null=True, blank=True) run_type = models.CharField(max_length=10, choices=RUN_TYPE_CHOICES, default='single') failed_step = models.IntegerField(null=True, blank=True) started_at = models.DateTimeField(default=timezone.now) finished_at = models.DateTimeField(null=True, blank=True) status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='running') exit_code = models.IntegerField(null=True, blank=True) output_tail = models.TextField(blank=True) class Meta: ordering = ['-started_at'] def mark_finished(self, status: str, exit_code: int | None, tail: str | None, failed_step: int | None = None): self.status = status self.exit_code = exit_code self.failed_step = failed_step self.finished_at = timezone.now() if tail is not None: self.output_tail = tail self.save(update_fields=['status', 'exit_code', 'failed_step', 'finished_at', 'output_tail']) def duration(self): # pragma: no cover simple helper end = self.finished_at or timezone.now() return (end - self.started_at).total_seconds() class BatchScript(models.Model): name = models.CharField(max_length=100, unique=True) description = models.CharField(max_length=255, blank=True) script = models.TextField(help_text="One command per line. Lines starting with # are ignored.") created_by = models.ForeignKey('auth.User', null=True, blank=True, on_delete=models.SET_NULL) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: ordering = ['name'] def __str__(self): # pragma: no cover simple return self.name def clean(self): # normalize script formatting if self.script: normalized = self.script.replace('\r\n', '\n').replace('\r', '\n') lines = [ln.rstrip() for ln in normalized.split('\n')] # remove trailing empty lines while lines and not lines[-1].strip(): lines.pop() self.script = '\n'.join(lines).strip('\n') def step_lines(self): # returns list of effective lines excluding comments/blank if not self.script: return [] return [l for l in self.script.split('\n') if l.strip() and not l.strip().startswith('#')] def step_count(self): # pragma: no cover trivial return len(self.step_lines()) class CommandTask(models.Model): name = models.CharField(max_length=100, unique=True, help_text="Key used internally") label = models.CharField(max_length=150) command = models.TextField() description = models.CharField(max_length=255, blank=True) created_by = models.ForeignKey('auth.User', null=True, blank=True, on_delete=models.SET_NULL) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: ordering = ['name'] def __str__(self): # pragma: no cover return self.label or self.name