from __future__ import annotations from dataclasses import dataclass, field from enum import StrEnum from pathlib import Path from .fields import Field, FieldConfiguration, FieldContents from .recipients import Recipient, RecipientList, RecipientType from .template import MailTemplate class TransportSecurity(StrEnum): PLAIN = "plain" TLS = "tls" STARTTLS = "starttls" @property def standard_port(self) -> int: return 465 if self == TransportSecurity.TLS else 587 @dataclass class MailServerSettings: server: str = "" port: int | None = None username: str = "" password: str = "" transport_security: TransportSecurity = TransportSecurity.PLAIN def use_plain(self) -> "MailServerSettings": self.transport_security = TransportSecurity.PLAIN self.port = self.port or self.transport_security.standard_port return self def use_tls(self) -> "MailServerSettings": self.transport_security = TransportSecurity.TLS self.port = self.port or self.transport_security.standard_port return self def use_starttls(self) -> "MailServerSettings": self.transport_security = TransportSecurity.STARTTLS self.port = self.port or self.transport_security.standard_port return self def resolved_port(self) -> int: return self.port or self.transport_security.standard_port @dataclass(frozen=True, slots=True) class MailAttachmentConfig: base_dir: Path file_filter: str = "*" include_subdirs: bool = False @dataclass class MailEntry: field_config: FieldConfiguration is_active: bool = True from_recipient: Recipient | None = None to: RecipientList = field(default_factory=RecipientList) cc: RecipientList = field(default_factory=RecipientList) bcc: RecipientList = field(default_factory=RecipientList) combine_to: bool = True combine_cc: bool = True combine_bcc: bool = True attachment_configs: list[MailAttachmentConfig] = field(default_factory=list) combine_attachments: bool = True field_contents: FieldContents = field(init=False) def __post_init__(self) -> None: self.field_contents = FieldContents(self.field_config) def add_to(self, recipient: Recipient) -> "MailEntry": self.to.add_recipient(recipient) return self def add_cc(self, recipient: Recipient) -> "MailEntry": self.cc.add_recipient(recipient) return self def add_bcc(self, recipient: Recipient) -> "MailEntry": self.bcc.add_recipient(recipient) return self def no_combine_to(self) -> "MailEntry": self.combine_to = False return self def combine_to_recipients(self) -> "MailEntry": self.combine_to = True return self def no_combine_attachments(self) -> "MailEntry": self.combine_attachments = False return self def combine_attachments_with_global(self) -> "MailEntry": self.combine_attachments = True return self def add_mail_attachment_config(self, config: MailAttachmentConfig) -> "MailEntry": self.attachment_configs.append(config) return self def set_field_content_for_name(self, name: str, value: Field | object) -> "MailEntry": if not self.field_contents.set_field_content_for_name(name, value): raise KeyError(f"unknown field: {name}") return self def get_field_content_from_name(self, name: str) -> Field: return self.field_contents.get_field_content_from_name(name) @dataclass class MailCampaign: mail_server_settings: MailServerSettings | None = None global_from: Recipient | None = None global_to: RecipientList = field(default_factory=RecipientList) global_cc: RecipientList = field(default_factory=RecipientList) global_bcc: RecipientList = field(default_factory=RecipientList) individual_from: bool = False individual_to: bool = False individual_cc: bool = False individual_bcc: bool = False base_attachment_path: Path = Path(".") global_attachment_configs: list[MailAttachmentConfig] = field(default_factory=list) individual_attachments: bool = False send_without_attachments: bool = True field_config: FieldConfiguration = field(default_factory=FieldConfiguration) field_contents: FieldContents = field(init=False) subject_template: MailTemplate = field(default_factory=MailTemplate) mail_template: MailTemplate = field(default_factory=MailTemplate) mail_entries: list[MailEntry] = field(default_factory=list) def __post_init__(self) -> None: self.field_contents = FieldContents(self.field_config) @classmethod def with_server_settings(cls, settings: MailServerSettings) -> "MailCampaign": return cls(mail_server_settings=settings) def add_field(self, name: str, field_type) -> "MailCampaign": from .fields import FieldDescription self.field_config.add_field_at_end(FieldDescription(name, field_type)) self.field_contents.ensure_field(self.field_config.get_field_description(name)) # type: ignore[arg-type] for entry in self.mail_entries: entry.field_contents.ensure_field(self.field_config.get_field_description(name)) # type: ignore[arg-type] return self def set_from(self, recipient: Recipient) -> "MailCampaign": self.global_from = recipient return self def add_to(self, recipient: Recipient) -> "MailCampaign": self.global_to.add_recipient(recipient) return self def allow_individual_to(self) -> "MailCampaign": self.individual_to = True return self def disallow_individual_to(self) -> "MailCampaign": self.individual_to = False return self def allow_individual_attachments(self) -> "MailCampaign": self.individual_attachments = True return self def disallow_individual_attachments(self) -> "MailCampaign": self.individual_attachments = False return self def dont_send_without_attachments(self) -> "MailCampaign": self.send_without_attachments = False return self def send_without_attachments_allowed(self) -> "MailCampaign": self.send_without_attachments = True return self def add_new_mail_entry(self) -> MailEntry: entry = MailEntry(self.field_config) self.mail_entries.append(entry) return entry def set_field_content_for_name(self, name: str, value: Field | object) -> "MailCampaign": if not self.field_contents.set_field_content_for_name(name, value): raise KeyError(f"unknown field: {name}") return self def get_field_content_from_name(self, name: str) -> Field: return self.field_contents.get_field_content_from_name(name) def all_recipients_for(self, entry: MailEntry) -> list[Recipient]: recipients: list[Recipient] = [] if not self.individual_to or entry.combine_to: recipients.extend(self.global_to.recipients) if not self.individual_cc or entry.combine_cc: recipients.extend(self.global_cc.recipients) if not self.individual_bcc or entry.combine_bcc: recipients.extend(self.global_bcc.recipients) if self.individual_to: recipients.extend(entry.to.recipients) if self.individual_cc: recipients.extend(entry.cc.recipients) if self.individual_bcc: recipients.extend(entry.bcc.recipients) return recipients