from __future__ import annotations from enum import StrEnum from pathlib import Path from typing import Any, Literal from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator class StrictModel(BaseModel): model_config = ConfigDict(extra="forbid", populate_by_name=True) class CampaignMode(StrEnum): DRAFT = "draft" TEST = "test" SEND = "send" class FieldType(StrEnum): STRING = "string" INTEGER = "integer" DOUBLE = "double" DATE = "date" PASSWORD = "password" class TransportSecurity(StrEnum): PLAIN = "plain" TLS = "tls" STARTTLS = "starttls" class RecipientType(StrEnum): TO = "to" CC = "cc" BCC = "bcc" REPLY_TO = "reply_to" BOUNCE_TO = "bounce_to" DISPOSITION_NOTIFICATION_TO = "disposition_notification_to" class Behavior(StrEnum): BLOCK = "block" ASK = "ask" DROP = "drop" CONTINUE = "continue" WARN = "warn" class MissingAddressBehavior(StrEnum): BLOCK = "block" DROP = "drop" class InactiveEntryBehavior(StrEnum): DROP = "drop" BLOCK = "block" WARN = "warn" class SourceType(StrEnum): CSV = "csv" JSON = "json" class ZipMethod(StrEnum): ZIP_STANDARD = "zip_standard" AES = "aes" class BuildStatus(StrEnum): BUILT = "built" BUILD_FAILED = "build_failed" class SendStatus(StrEnum): DRAFT = "draft" QUEUED = "queued" class CampaignMeta(StrictModel): id: str name: str description: str | None = None mode: CampaignMode = CampaignMode.DRAFT class FieldDefinition(StrictModel): name: str type: FieldType = FieldType.STRING label: str | None = None required: bool = False can_override: bool = True class SmtpConfig(StrictModel): host: str | None = None port: int | None = Field(default=None, ge=1, le=65535) username: str | None = None password: str | None = None security: TransportSecurity = TransportSecurity.STARTTLS timeout_seconds: int = Field(default=30, ge=1) class ImapConfig(StrictModel): enabled: bool = False host: str | None = None port: int | None = Field(default=None, ge=1, le=65535) username: str | None = None password: str | None = None security: TransportSecurity = TransportSecurity.TLS sent_folder: str = "auto" timeout_seconds: int = Field(default=30, ge=1) class ServerConfig(StrictModel): smtp: SmtpConfig | None = None imap: ImapConfig | None = None class RecipientConfig(StrictModel): email: str name: str | None = None recipient_type: RecipientType = Field(default=RecipientType.TO, alias="type") @field_validator("email") @classmethod def email_should_look_like_address(cls, value: str) -> str: # JSON Schema's format=email remains the stricter validation layer. # Keep this deliberately lightweight to avoid an extra email-validator dependency. if "@" not in value: raise ValueError("email must contain '@'") return value class RecipientsConfig(StrictModel): from_: RecipientConfig | None = Field(default=None, alias="from") @field_validator("from_", mode="before") @classmethod def empty_from_object_means_unset(cls, value: Any) -> Any: if isinstance(value, dict) and not any(value.values()): return None return value allow_individual_from: bool = False to: list[RecipientConfig] = Field(default_factory=list) allow_individual_to: bool = False cc: list[RecipientConfig] = Field(default_factory=list) allow_individual_cc: bool = False bcc: list[RecipientConfig] = Field(default_factory=list) allow_individual_bcc: bool = False reply_to: list[RecipientConfig] = Field(default_factory=list) allow_individual_reply_to: bool = False bounce_to: list[RecipientConfig] = Field(default_factory=list) allow_individual_bounce_to: bool = False disposition_notification_to: list[RecipientConfig] = Field(default_factory=list) allow_individual_disposition_notification_to: bool = False class TemplateSourceConfig(StrictModel): type: Literal["files"] = "files" subject_path: str | None = None text_path: str | None = None html_path: str | None = None encoding: str = "utf-8" @model_validator(mode="after") def at_least_one_path(self) -> "TemplateSourceConfig": if not any([self.subject_path, self.text_path, self.html_path]): raise ValueError("template.source must define subject_path, text_path or html_path") return self class TemplateConfig(StrictModel): subject: str | None = None text: str | None = None html: str | None = None source: TemplateSourceConfig | None = None @model_validator(mode="after") def inline_or_source(self) -> "TemplateConfig": inline_values = any(value is not None for value in [self.subject, self.text, self.html]) if self.source and inline_values: raise ValueError("template must be either inline or source-based, not both") if self.source: return self if not self.subject: raise ValueError("inline template requires subject") return self @property def is_external(self) -> bool: return self.source is not None class ZipConfig(StrictModel): enabled: bool = False filename_template: str | None = None password_template: str | None = None method: ZipMethod = ZipMethod.AES class AttachmentBasePathConfig(StrictModel): id: str | None = None name: str path: str = "." allow_individual: bool = False unsent_warning: bool = False # Legacy UI builds briefly wrote a source value. Keep accepting it so older # drafts do not become invalid merely because the current UI no longer shows # or edits that column. source: str | None = None class AttachmentConfig(StrictModel): id: str | None = None label: str | None = None base_path_id: str | None = None # Legacy UI helper. Current attachment resolution ignores this value and # treats direct files as plain file_filter patterns without wildcards. # Keep accepting it so existing drafts with {"type": ""}, "direct" # or "pattern" remain valid. type_: str | None = Field(default=None, alias="type") base_dir: str file_filter: str include_subdirs: bool = False required: bool = True allow_multiple: bool = False @field_validator("type_", mode="before") @classmethod def empty_type_means_unset(cls, value: Any) -> Any: if value == "": return None return value # None means: inherit from validation_policy. Explicit values remain # supported for backwards compatibility and per-rule overrides. missing_behavior: Behavior | None = None ambiguous_behavior: Behavior | None = None zip: ZipConfig = Field(default_factory=ZipConfig) class AttachmentsConfig(StrictModel): base_path: str = "." base_paths: list[AttachmentBasePathConfig] = Field(default_factory=list) allow_individual: bool = False send_without_attachments: bool = True global_: list[AttachmentConfig] = Field(default_factory=list, alias="global") missing_behavior: Behavior = Behavior.ASK ambiguous_behavior: Behavior = Behavior.ASK @property def individual_base_path_values(self) -> set[str]: return {base_path.path for base_path in self.base_paths if base_path.allow_individual} class EntryConfig(StrictModel): id: str | None = None active: bool = True # Compatibility fields written by older/current WebUI recipient rows. # Address routing uses the explicit to/cc/bcc/reply_to/from fields below; # these values are retained for round-tripping but are not used for sending. name: str | None = None email: str | None = None from_: RecipientConfig | None = Field(default=None, alias="from") @field_validator("from_", mode="before") @classmethod def empty_from_object_means_unset(cls, value: Any) -> Any: if isinstance(value, dict) and not any(value.values()): return None return value to: list[RecipientConfig] = Field(default_factory=list) combine_to: bool = True cc: list[RecipientConfig] = Field(default_factory=list) combine_cc: bool = True bcc: list[RecipientConfig] = Field(default_factory=list) combine_bcc: bool = True reply_to: list[RecipientConfig] = Field(default_factory=list) combine_reply_to: bool = True bounce_to: list[RecipientConfig] = Field(default_factory=list) combine_bounce_to: bool = True disposition_notification_to: list[RecipientConfig] = Field(default_factory=list) combine_disposition_notification_to: bool = True attachments: list[AttachmentConfig] = Field(default_factory=list) combine_attachments: bool = True fields: dict[str, Any] = Field(default_factory=dict) last_sent: str | None = None class SourceConfig(StrictModel): type: SourceType path: str delimiter: str = ";" encoding: str = "utf-8" has_header: bool = True class EntriesConfig(StrictModel): inline: list[EntryConfig] | None = None source: SourceConfig | None = None mapping: dict[str, str] | None = None defaults: EntryConfig | None = None @model_validator(mode="after") def inline_or_external(self) -> "EntriesConfig": has_inline = self.inline is not None has_external_source = self.source is not None or self.mapping is not None # defaults are compatible with both inline and external entries. The # WebUI stores the current per-entry combination defaults here even for # inline campaigns, so treating defaults as an external-source marker # made valid UI drafts fail backend validation. if has_inline and has_external_source: raise ValueError("entries must be either inline or source-based, not both") if has_inline: return self if self.source is None or self.mapping is None: raise ValueError("external entries require source and mapping") return self @property def is_inline(self) -> bool: return self.inline is not None @property def is_external(self) -> bool: return self.source is not None class ValidationPolicy(StrictModel): missing_required_attachment: Behavior = Behavior.ASK missing_optional_attachment: Behavior = Behavior.WARN ambiguous_attachment_match: Behavior = Behavior.ASK ignore_empty_fields: bool = False unsent_attachment_files: Behavior = Behavior.WARN missing_email: MissingAddressBehavior = MissingAddressBehavior.BLOCK template_error: MissingAddressBehavior = MissingAddressBehavior.BLOCK inactive_entry: InactiveEntryBehavior = InactiveEntryBehavior.DROP class RateLimitConfig(StrictModel): messages_per_minute: int = Field(default=5, ge=1) concurrency: int = Field(default=1, ge=1) class ImapAppendSentConfig(StrictModel): enabled: bool = False folder: str = "auto" class RetryConfig(StrictModel): max_attempts: int = Field(default=3, ge=1) backoff_seconds: list[int] = Field(default_factory=lambda: [60, 300, 900]) @field_validator("backoff_seconds") @classmethod def backoff_values_must_be_positive(cls, values: list[int]) -> list[int]: if any(value < 1 for value in values): raise ValueError("backoff_seconds values must be >= 1") return values class DeliveryConfig(StrictModel): rate_limit: RateLimitConfig = Field(default_factory=RateLimitConfig) imap_append_sent: ImapAppendSentConfig = Field(default_factory=ImapAppendSentConfig) retry: RetryConfig = Field(default_factory=RetryConfig) class StatusTrackingConfig(StrictModel): enabled: bool = True initial_build_status: BuildStatus = BuildStatus.BUILT initial_send_status: SendStatus = SendStatus.DRAFT class CampaignConfig(StrictModel): version: Literal["1.0"] campaign: CampaignMeta fields: list[FieldDefinition] = Field(default_factory=list) global_values: dict[str, Any] = Field(default_factory=dict) server: ServerConfig = Field(default_factory=ServerConfig) recipients: RecipientsConfig = Field(default_factory=RecipientsConfig) template: TemplateConfig attachments: AttachmentsConfig = Field(default_factory=AttachmentsConfig) entries: EntriesConfig validation_policy: ValidationPolicy = Field(default_factory=ValidationPolicy) delivery: DeliveryConfig = Field(default_factory=DeliveryConfig) status_tracking: StatusTrackingConfig = Field(default_factory=StatusTrackingConfig) @model_validator(mode="after") def field_names_must_be_unique(self) -> "CampaignConfig": names = [field.name for field in self.fields] duplicates = sorted({name for name in names if names.count(name) > 1}) if duplicates: raise ValueError(f"duplicate field definitions: {', '.join(duplicates)}") return self @property def field_names(self) -> set[str]: return {field.name for field in self.fields} def resolve_relative_path(self, campaign_file: Path, raw_path: str) -> Path: path = Path(raw_path).expanduser() if path.is_absolute(): return path return (campaign_file.parent / path).resolve()