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 from govoplan_core.mail.config import ImapConfig, ImapServerConfig, SmtpConfig, SmtpServerConfig, TransportCredentials, TransportSecurity 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 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 ZipRuleMode(StrEnum): INHERIT = "inherit" INCLUDE = "include" EXCLUDE = "exclude" class ZipPasswordScope(StrEnum): LOCAL = "local" GLOBAL = "global" class ZipPasswordMode(StrEnum): NONE = "none" DIRECT = "direct" FIELD = "field" TEMPLATE = "template" 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 MailServerCredentials(StrictModel): smtp: TransportCredentials = Field(default_factory=TransportCredentials) imap: TransportCredentials = Field(default_factory=TransportCredentials) class ServerConfig(StrictModel): mail_profile_id: str | None = None inherit_smtp_credentials: bool = True inherit_imap_credentials: bool = True smtp: SmtpServerConfig | None = None imap: ImapServerConfig | None = None credentials: MailServerCredentials = Field(default_factory=MailServerCredentials) @model_validator(mode="before") @classmethod def normalize_legacy_credentials(cls, value: Any) -> Any: if not isinstance(value, dict): return value data = dict(value) credentials = data.get("credentials") if isinstance(data.get("credentials"), dict) else {} credentials = {key: dict(item) for key, item in credentials.items() if isinstance(item, dict)} for protocol in ("smtp", "imap"): transport = data.get(protocol) if not isinstance(transport, dict): continue next_transport = dict(transport) next_credentials = dict(credentials.get(protocol) or {}) for field in ("username", "password"): if field in next_transport and field not in next_credentials: next_credentials[field] = next_transport[field] next_transport.pop(field, None) next_transport.pop("enabled", None) data[protocol] = next_transport if next_credentials: credentials[protocol] = next_credentials if credentials: data["credentials"] = credentials return data def runtime_smtp_config(self) -> SmtpConfig | None: if self.smtp is None: return None payload = self.smtp.model_dump(mode="json") payload.update(self.credentials.smtp.model_dump(mode="json", exclude_none=True)) return SmtpConfig.model_validate(payload) def runtime_imap_config(self) -> ImapConfig | None: if self.imap is None: return None payload = self.imap.model_dump(mode="json") payload.update(self.credentials.imap.model_dump(mode="json", exclude_none=True)) return ImapConfig.model_validate(payload) 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_: list[RecipientConfig] = Field(default_factory=list, alias="from", max_length=1) @field_validator("from_", mode="before") @classmethod def normalize_from_list(cls, value: Any) -> Any: # Older campaign files stored From as one object. The canonical model # is now an array so every address header has the same list semantics. if value is None: return [] if isinstance(value, dict): return [] if not any(value.values()) else [value] 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 TemplateBodyMode(StrEnum): TEXT = "text" HTML = "html" BOTH = "both" class TemplateConfig(StrictModel): subject: str | None = None text: str | None = None html: str | None = None body_mode: TemplateBodyMode = TemplateBodyMode.BOTH 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 ZipArchiveConfig(StrictModel): id: str name: str = "attachments.zip" standard: bool = False password_enabled: bool = False password_field: str | None = None password_scope: ZipPasswordScope = ZipPasswordScope.LOCAL method: ZipMethod = ZipMethod.AES # Compatibility fields for campaigns created by the first single-archive # implementation. New WebUI campaigns use password_enabled/field/scope. password_mode: ZipPasswordMode | None = None password: str | None = None password_template: str | None = None @model_validator(mode="before") @classmethod def normalize_legacy_archive(cls, value: Any) -> Any: if not isinstance(value, dict): return value normalized = dict(value) if "name" not in normalized and normalized.get("filename_template"): normalized["name"] = normalized.get("filename_template") normalized.pop("filename_template", None) mode = normalized.get("password_mode") if "password_enabled" not in normalized and mode in { ZipPasswordMode.DIRECT.value, ZipPasswordMode.FIELD.value, ZipPasswordMode.TEMPLATE.value, }: normalized["password_enabled"] = True if mode == ZipPasswordMode.FIELD.value and "password_scope" not in normalized: normalized["password_scope"] = ZipPasswordScope.LOCAL.value return normalized class ZipCollectionConfig(StrictModel): enabled: bool = False archives: list[ZipArchiveConfig] = Field(default_factory=list) @model_validator(mode="before") @classmethod def normalize_single_archive_config(cls, value: Any) -> Any: """Upgrade the original single-recipient-ZIP object in memory. The campaign JSON remains portable: old filename/password settings are converted to one standard archive, while the new format is passed through unchanged. """ if not isinstance(value, dict) or "archives" in value: return value legacy = dict(value) enabled = bool(legacy.get("enabled")) meaningful = enabled or any( legacy.get(key) not in (None, "", False, "none", "inherit") for key in ("filename_template", "password_mode", "password", "password_field", "password_template") ) if not meaningful: return {"enabled": False, "archives": []} archive = { "id": "default", "name": legacy.get("filename_template") or "attachments.zip", "standard": True, "method": legacy.get("method", ZipMethod.AES.value), "password_mode": legacy.get("password_mode"), "password": legacy.get("password"), "password_field": legacy.get("password_field"), "password_template": legacy.get("password_template"), } mode = legacy.get("password_mode") if not mode and legacy.get("password_template"): mode = ZipPasswordMode.TEMPLATE.value archive["password_mode"] = mode archive["password_enabled"] = mode in { ZipPasswordMode.DIRECT.value, ZipPasswordMode.FIELD.value, ZipPasswordMode.TEMPLATE.value, } archive["password_scope"] = ZipPasswordScope.LOCAL.value return {"enabled": enabled, "archives": [archive]} @property def standard_archive(self) -> ZipArchiveConfig | None: return next((archive for archive in self.archives if archive.standard), self.archives[0] if self.archives else None) def archive_by_id(self, archive_id: str | None) -> ZipArchiveConfig | None: if not archive_id: return None return next((archive for archive in self.archives if archive.id == archive_id), None) class ZipRuleConfig(StrictModel): # New format: inherit the campaign standard, select an archive id, or use # the reserved value "exclude" to send the file outside every archive. archive_id: str = ZipRuleMode.INHERIT.value # Original per-rule fields remain accepted for imported campaign JSON. enabled: bool = False mode: ZipRuleMode = ZipRuleMode.INHERIT filename_template: str | None = None password_mode: ZipPasswordMode = ZipPasswordMode.NONE password: str | None = None password_field: str | None = None password_template: str | None = None method: ZipMethod = ZipMethod.AES @model_validator(mode="before") @classmethod def normalize_legacy_password_mode(cls, value: Any) -> Any: if isinstance(value, dict) and value.get("password_template") and "password_mode" not in value: return {**value, "password_mode": ZipPasswordMode.TEMPLATE.value} return value 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 message_filename_template: str | None = None zip_entry_name_template: str | None = None @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: ZipRuleConfig = Field(default_factory=ZipRuleConfig) class AttachmentsConfig(StrictModel): base_path: str = "." base_paths: list[AttachmentBasePathConfig] = Field(default_factory=list) allow_individual: bool = False send_without_attachments: bool = True zip: ZipCollectionConfig = Field(default_factory=ZipCollectionConfig) 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 @model_validator(mode="before") @classmethod def normalize_legacy_combine_flags(cls, value: Any) -> Any: if not isinstance(value, dict): return value normalized = dict(value) for address_field in ("from", "to", "cc", "bcc", "reply_to", "bounce_to", "disposition_notification_to"): merge_key = f"merge_{address_field}" combine_key = f"combine_{address_field}" if merge_key not in normalized and combine_key in normalized: normalized[merge_key] = normalized[combine_key] normalized.pop(combine_key, None) return normalized # 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_: list[RecipientConfig] = Field(default_factory=list, alias="from", max_length=1) merge_from: bool = False # Deprecated compatibility field; From never merges. @field_validator("from_", mode="before") @classmethod def normalize_from_list(cls, value: Any) -> Any: if value is None: return [] if isinstance(value, dict): return [] if not any(value.values()) else [value] return value to: list[RecipientConfig] = Field(default_factory=list) merge_to: bool = True cc: list[RecipientConfig] = Field(default_factory=list) merge_cc: bool = True bcc: list[RecipientConfig] = Field(default_factory=list) merge_bcc: bool = True reply_to: list[RecipientConfig] = Field(default_factory=list) merge_reply_to: bool = True bounce_to: list[RecipientConfig] = Field(default_factory=list) merge_bounce_to: bool = True disposition_notification_to: list[RecipientConfig] = Field(default_factory=list) merge_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 ImportColumnProvenance(StrictModel): column_index: int = Field(ge=0) header: str kind: str field_name: str | None = None new_field_name: str | None = None class ImportProvenance(StrictModel): id: str imported_at: str mode: Literal["append", "replace"] source_type: Literal["csv", "xlsx", "text"] filename: str | None = None sheet_name: str | None = None encoding: str | None = None delimiter: str | None = None header_rows: int = Field(default=0, ge=0) quoted: bool | None = None value_separators: str | None = None rows_total: int = Field(default=0, ge=0) valid_rows: int = Field(default=0, ge=0) invalid_rows: int = Field(default=0, ge=0) imported_rows: int = Field(default=0, ge=0) field_names_created: list[str] = Field(default_factory=list) attachment_patterns: int = Field(default=0, ge=0) mapping: list[ImportColumnProvenance] = Field(default_factory=list) 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 imports: list[ImportProvenance] = Field(default_factory=list) @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()