from __future__ import annotations from django import forms from .models import GlobalConfig, HostConfig, ScheduleConfig from .scheduler import parse_cron_expr class NewlineListField(forms.CharField): widget = forms.Textarea def __init__(self, *args, **kwargs) -> None: kwargs.setdefault("required", False) super().__init__(*args, **kwargs) def prepare_value(self, value): if isinstance(value, list): return "\n".join(str(item) for item in value) return value def to_python(self, value) -> list[str]: if not value: return [] if isinstance(value, list): return [str(item).strip() for item in value if str(item).strip()] return [line.strip() for line in str(value).splitlines() if line.strip()] class NullableNewlineListField(NewlineListField): def to_python(self, value) -> list[str] | None: parsed = super().to_python(value) return parsed or None class HostConfigForm(forms.ModelForm): includes = NewlineListField(help_text="One include path per line. Leave empty to include defaults.") excludes_add = NewlineListField(help_text="One additional exclude pattern per line.") excludes_replace = NullableNewlineListField( help_text="Optional. When set, replaces global excludes; one pattern per line." ) rsync_extra_args = NewlineListField(help_text="One extra rsync argument per line.") class Meta: model = HostConfig fields = ( "address", "enabled", "ssh_user", "ssh_port", "source_root", "includes", "excludes_add", "excludes_replace", "rsync_extra_args", "retention_daily", "retention_weekly", "retention_monthly", "retention_yearly", ) help_texts = { "ssh_user": "Leave empty to use the global SSH user.", "ssh_port": "Leave empty to use the global SSH port.", "source_root": "Leave empty to use the global default source root.", } class CreateHostConfigForm(HostConfigForm): class Meta(HostConfigForm.Meta): fields = ("host", *HostConfigForm.Meta.fields) help_texts = { **HostConfigForm.Meta.help_texts, "host": "Stable internal host name used for backup paths.", } class GlobalConfigForm(forms.ModelForm): ssh_options = NewlineListField(help_text="One SSH option per line.") rsync_args = NewlineListField(help_text="One default rsync argument per line.") rsync_extra_args = NewlineListField(help_text="One extra rsync argument per line.") excludes_default = NewlineListField(help_text="One default exclude pattern per line.") class Meta: model = GlobalConfig fields = ( "name", "backup_root", "pobsync_home", "ssh_user", "ssh_port", "ssh_options", "rsync_binary", "rsync_args", "rsync_extra_args", "rsync_timeout_seconds", "rsync_bwlimit_kbps", "default_source_root", "default_destination_subdir", "excludes_default", "retention_daily", "retention_weekly", "retention_monthly", "retention_yearly", ) help_texts = { "name": "Usually 'default'. The backup engine currently reads the default config.", "backup_root": "Directory that contains host backup folders.", "pobsync_home": "Base directory for runtime state inside the container or host.", "default_source_root": "Used by hosts without a custom source root.", "default_destination_subdir": "Optional subdirectory below each snapshot.", } class ScheduleConfigForm(forms.ModelForm): cron_expr = forms.CharField( label="Cron expression", help_text='Five-field cron expression, for example "15 2 * * *".', ) prune_max_delete = forms.IntegerField(min_value=0) class Meta: model = ScheduleConfig fields = ( "cron_expr", "user", "enabled", "prune", "prune_max_delete", "prune_protect_bases", ) def clean_cron_expr(self) -> str: cron_expr = self.cleaned_data["cron_expr"].strip() try: parse_cron_expr(cron_expr) except ValueError as exc: raise forms.ValidationError(str(exc)) from exc return cron_expr