148 lines
4.8 KiB
Python
148 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
from enum import StrEnum
|
|
from typing import Any
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
|
|
|
|
|
|
class StrictModel(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", populate_by_name=True)
|
|
|
|
|
|
class TransportSecurity(StrEnum):
|
|
PLAIN = "plain"
|
|
TLS = "tls"
|
|
STARTTLS = "starttls"
|
|
|
|
|
|
class TransportCredentials(StrictModel):
|
|
username: str | None = None
|
|
password: str | None = None
|
|
|
|
|
|
class SmtpServerConfig(StrictModel):
|
|
host: str | None = None
|
|
port: int | None = Field(default=None, ge=1, le=65535)
|
|
security: TransportSecurity = TransportSecurity.STARTTLS
|
|
timeout_seconds: int = Field(default=30, ge=1)
|
|
|
|
@model_validator(mode="after")
|
|
def apply_default_port(self) -> "SmtpServerConfig":
|
|
if self.port is None:
|
|
self.port = default_smtp_port(self.security)
|
|
return self
|
|
|
|
|
|
class ImapFolderMappings(StrictModel):
|
|
"""Profile-level names for the standard IMAP mailbox roles."""
|
|
|
|
inbox: str | None = None
|
|
sent: str | None = None
|
|
drafts: str | None = None
|
|
trash: str | None = None
|
|
archive: str | None = None
|
|
junk: str | None = None
|
|
|
|
@field_validator("*", mode="before")
|
|
@classmethod
|
|
def normalize_folder_name(cls, value: Any) -> Any:
|
|
if value is None:
|
|
return None
|
|
normalized = str(value).strip()
|
|
return normalized or None
|
|
|
|
|
|
class ImapServerConfig(StrictModel):
|
|
host: str | None = None
|
|
port: int | None = Field(default=None, ge=1, le=65535)
|
|
security: TransportSecurity = TransportSecurity.TLS
|
|
sent_folder: str = "auto"
|
|
folder_mappings: ImapFolderMappings | None = None
|
|
timeout_seconds: int = Field(default=30, ge=1)
|
|
|
|
@model_validator(mode="before")
|
|
@classmethod
|
|
def discard_legacy_enabled(cls, value: Any) -> Any:
|
|
if not isinstance(value, dict):
|
|
return value
|
|
data = dict(value)
|
|
data.pop("enabled", None)
|
|
mappings_value = data.get("folder_mappings")
|
|
mappings = (
|
|
mappings_value.model_dump(exclude_none=True)
|
|
if isinstance(mappings_value, ImapFolderMappings)
|
|
else dict(mappings_value)
|
|
if isinstance(mappings_value, dict)
|
|
else {}
|
|
)
|
|
mapped_sent = str(mappings.get("sent") or "").strip()
|
|
legacy_sent = str(data.get("sent_folder") or "").strip()
|
|
if mapped_sent:
|
|
# The typed mapping is canonical when both new and legacy callers
|
|
# provide a Sent value. Keep the legacy field synchronized for
|
|
# existing Campaign append consumers.
|
|
data["sent_folder"] = mapped_sent
|
|
elif legacy_sent and legacy_sent != "auto":
|
|
data["sent_folder"] = legacy_sent
|
|
mappings["sent"] = legacy_sent
|
|
if mappings:
|
|
data["folder_mappings"] = mappings
|
|
return data
|
|
|
|
@model_validator(mode="after")
|
|
def apply_default_port(self) -> "ImapServerConfig":
|
|
if self.port is None:
|
|
self.port = default_imap_port(self.security)
|
|
return self
|
|
|
|
|
|
class SmtpConfig(SmtpServerConfig):
|
|
username: str | None = None
|
|
password: str | None = None
|
|
|
|
|
|
class ImapConfig(ImapServerConfig):
|
|
username: str | None = None
|
|
password: str | None = None
|
|
|
|
|
|
def normalize_split_transport_credentials(value: object) -> object:
|
|
"""Move legacy transport username/password fields into credentials."""
|
|
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 default_smtp_port(security: TransportSecurity | str | None) -> int:
|
|
if security == TransportSecurity.TLS or security == "tls":
|
|
return 465
|
|
if security == TransportSecurity.PLAIN or security == "plain":
|
|
return 25
|
|
return 587
|
|
|
|
|
|
def default_imap_port(security: TransportSecurity | str | None) -> int:
|
|
if security == TransportSecurity.TLS or security == "tls":
|
|
return 993
|
|
return 143
|