from __future__ import annotations import csv from dataclasses import dataclass from enum import StrEnum from pathlib import Path from typing import Iterable from pydantic import BaseModel, ConfigDict, Field from .addressing import effective_address_lists from .field_values import ignored_entry_field_overrides from .models import ( AttachmentConfig, CampaignConfig, DeliveryChannelPolicy, EntryConfig, FieldType, PostboxTargetConfig, SourceType, ZipArchiveConfig, ZipPasswordMode, ZipPasswordScope, ZipRuleMode, effective_delivery_channel_policy, effective_postbox_targets, ) from ..attachments.resolver import resolve_campaign_attachments class Severity(StrEnum): INFO = "info" WARNING = "warning" ERROR = "error" class SemanticIssue(BaseModel): model_config = ConfigDict(extra="forbid") severity: Severity code: str message: str path: str | None = None class SemanticReport(BaseModel): model_config = ConfigDict(extra="forbid") campaign_id: str campaign_name: str issues: list[SemanticIssue] = Field(default_factory=list) entries_mode: str entries_count: int | None = None attachments_base_path: str rate_limit: str imap_append_enabled: bool @property def error_count(self) -> int: return sum(1 for issue in self.issues if issue.severity == Severity.ERROR) @property def warning_count(self) -> int: return sum(1 for issue in self.issues if issue.severity == Severity.WARNING) @property def ok(self) -> bool: return self.error_count == 0 @dataclass(frozen=True, slots=True) class _EntriesValidation: mode: str count: int | None issues: list[SemanticIssue] def _issue(severity: Severity, code: str, message: str, path: str | None = None) -> SemanticIssue: return SemanticIssue(severity=severity, code=code, message=message, path=path) def _resolve(campaign_file: Path, raw_path: str) -> Path: path = Path(raw_path).expanduser() if path.is_absolute(): return path return (campaign_file.parent / path).resolve() def _mapping_target_field_name(target: str) -> str | None: if target.startswith("fields."): return target.split(".", 1)[1] return None def _mapping_target_known(target: str, field_names: set[str]) -> bool: direct_targets = { "id", "active", "last_sent", "merge_from", "merge_to", "merge_cc", "merge_bcc", "merge_reply_to", "merge_bounce_to", "merge_disposition_notification_to", "merge_postbox_targets", "channel_policy", "combine_to", "combine_cc", "combine_bcc", "combine_reply_to", "combine_bounce_to", "combine_disposition_notification_to", "combine_attachments", } if target in direct_targets: return True if target.startswith("fields."): name = target.split(".", 1)[1] return not field_names or name in field_names if target.startswith("from."): return target in {"from.email", "from.name", "from.type"} for prefix in ["to", "cc", "bcc", "reply_to", "bounce_to", "disposition_notification_to"]: if target.startswith(prefix + "."): parts = target.split(".") return len(parts) == 3 and parts[1].isdigit() and parts[2] in {"email", "name", "type"} if target.startswith("attachments."): parts = target.split(".") # attachments.0.zip.filename_template etc. if len(parts) >= 3 and parts[1].isdigit(): if parts[2] in { "id", "label", "base_dir", "file_filter", "include_subdirs", "required", "allow_multiple", "missing_behavior", "ambiguous_behavior", }: return len(parts) == 3 if parts[2] == "zip" and len(parts) == 4: return parts[3] in {"enabled", "mode", "filename_template", "password_mode", "password", "password_field", "password_template", "method"} return False def _csv_header(path: Path, delimiter: str, encoding: str) -> list[str] | None: with path.open("r", encoding=encoding, newline="") as handle: reader = csv.reader(handle, delimiter=delimiter) try: return next(reader) except StopIteration: return [] def _iter_template_source_paths(config: CampaignConfig) -> Iterable[tuple[str, str]]: if not config.template.source: return [] source = config.template.source paths: list[tuple[str, str]] = [] if source.subject_path: paths.append(("/template/source/subject_path", source.subject_path)) if source.text_path: paths.append(("/template/source/text_path", source.text_path)) if source.html_path: paths.append(("/template/source/html_path", source.html_path)) return paths def _attachment_base_path_report_value(config: CampaignConfig) -> str: if config.attachments.base_paths: return ", ".join(f"{base_path.name}: {base_path.path}" for base_path in config.attachments.base_paths) return config.attachments.base_path def _iter_attachment_rules(config: CampaignConfig) -> Iterable[tuple[str, AttachmentConfig, bool]]: for index, attachment_config in enumerate(config.attachments.global_): yield f"/attachments/global/{index}", attachment_config, False inline_entries = config.entries.inline or [] if config.entries.is_inline else [] for entry_index, entry in enumerate(inline_entries): if not entry.active: continue for attachment_index, attachment_config in enumerate(entry.attachments): yield f"/entries/inline/{entry_index}/attachments/{attachment_index}", attachment_config, True if config.entries.defaults: for attachment_index, attachment_config in enumerate(config.entries.defaults.attachments): yield f"/entries/defaults/attachments/{attachment_index}", attachment_config, True def _attachment_path_issues(config: CampaignConfig) -> list[SemanticIssue]: issues: list[SemanticIssue] = [] configured_paths = {base_path.path for base_path in config.attachments.base_paths} individual_paths = config.attachments.individual_base_path_values if config.attachments.base_paths: for index, base_path in enumerate(config.attachments.base_paths): if not base_path.name.strip(): issues.append(_issue(Severity.WARNING, "attachment_base_path_missing_name", "attachment base path has no display name", f"/attachments/base_paths/{index}/name")) if not base_path.path.strip(): issues.append(_issue(Severity.ERROR, "attachment_base_path_missing_path", "attachment base path has no path", f"/attachments/base_paths/{index}/path")) elif not config.attachments.base_path: issues.append(_issue(Severity.INFO, "missing_attachment_base_path", "Attachment base path is not configured yet.", "/attachments/base_path")) if configured_paths: for path, attachment_config, is_individual in _iter_attachment_rules(config): if attachment_config.base_dir and attachment_config.base_dir not in configured_paths: issues.append(_issue( Severity.WARNING, "unknown_attachment_base_path", f"attachment rule refers to base path {attachment_config.base_dir!r}, but it is not listed in attachments.base_paths", f"{path}/base_dir", )) if is_individual and individual_paths and attachment_config.base_dir not in individual_paths: issues.append(_issue( Severity.WARNING, "individual_attachment_base_path_not_allowed", f"individual attachment rule uses base path {attachment_config.base_dir!r}, but that base path does not allow individual attachments", f"{path}/base_dir", )) return issues def _zip_configuration_issues(config: CampaignConfig) -> list[SemanticIssue]: collection = config.attachments.zip if not collection.enabled: return [] if not collection.archives: return [_issue(Severity.ERROR, "zip_archive_missing", "Attachment zipping is enabled, but no ZIP archive is configured", "/attachments/zip/archives")] issues, archive_ids, standard_count = _zip_archive_catalog_issues(config) if standard_count != 1: issues.append(_issue(Severity.ERROR, "zip_standard_archive_invalid", "Exactly one ZIP archive must be selected as the campaign standard", "/attachments/zip/archives")) issues.extend(_zip_rule_selection_issues(config, archive_ids)) return issues def _zip_archive_catalog_issues(config: CampaignConfig) -> tuple[list[SemanticIssue], set[str], int]: issues: list[SemanticIssue] = [] archive_ids: set[str] = set() archive_names: set[str] = set() standard_count = 0 field_definitions = {field.name: field for field in config.fields} for index, archive in enumerate(config.attachments.zip.archives): path = f"/attachments/zip/archives/{index}" issues.extend(_zip_archive_identity_issues(archive, path, archive_ids, archive_names)) if archive.standard: standard_count += 1 issues.extend(_zip_password_issues(config, archive, path, field_definitions)) return issues, archive_ids, standard_count def _zip_archive_identity_issues( archive: ZipArchiveConfig, path: str, archive_ids: set[str], archive_names: set[str], ) -> list[SemanticIssue]: issues: list[SemanticIssue] = [] if not archive.id.strip(): issues.append(_issue(Severity.ERROR, "zip_archive_id_missing", "ZIP archive has no identifier", f"{path}/id")) elif archive.id in archive_ids: issues.append(_issue(Severity.ERROR, "zip_archive_id_duplicate", f"ZIP archive id {archive.id!r} is used more than once", f"{path}/id")) archive_ids.add(archive.id) normalized_archive_name = _normalized_zip_archive_name(archive.name) if not normalized_archive_name: issues.append(_issue(Severity.ERROR, "zip_archive_name_missing", "ZIP archive has no filename", f"{path}/name")) elif normalized_archive_name in archive_names: issues.append(_issue(Severity.ERROR, "zip_archive_name_duplicate", f"ZIP archive filename {archive.name!r} is used more than once; archive filenames must be unique", f"{path}/name")) archive_names.add(normalized_archive_name) return issues def _zip_password_issues( config: CampaignConfig, archive: ZipArchiveConfig, path: str, field_definitions: dict[str, object], ) -> list[SemanticIssue]: if not archive.password_enabled: return [] if archive.password_mode == ZipPasswordMode.DIRECT: if not (archive.password or ""): return [_issue(Severity.ERROR, "zip_password_missing", "A legacy fixed ZIP password is enabled, but no password is configured", f"{path}/password")] return [] if archive.password_mode == ZipPasswordMode.TEMPLATE: if not (archive.password_template or ""): return [_issue(Severity.ERROR, "zip_password_template_missing", "A legacy ZIP password template is enabled, but no template is configured", f"{path}/password_template")] return [] return _zip_password_field_issues(config, archive, path, field_definitions) def _zip_password_field_issues( config: CampaignConfig, archive: ZipArchiveConfig, path: str, field_definitions: dict[str, object], ) -> list[SemanticIssue]: field_name = (archive.password_field or "").strip() field = field_definitions.get(field_name) if not field_name: return [_issue(Severity.ERROR, "zip_password_field_missing", f"ZIP archive {archive.name!r} has password protection enabled, but no field is selected", f"{path}/password_field")] if field is None: return [_issue(Severity.ERROR, "zip_password_field_unknown", f"ZIP password field {field_name!r} is not declared in campaign fields", f"{path}/password_field")] if getattr(field, "type", None) != FieldType.PASSWORD: return [_issue(Severity.WARNING, "zip_password_field_not_password_type", f"ZIP password field {field_name!r} is not configured with field type 'password'", f"{path}/password_field")] if archive.password_scope == ZipPasswordScope.GLOBAL and config.global_values.get(field_name) in (None, ""): return [_issue(Severity.ERROR, "zip_global_password_value_missing", f"Global ZIP password field {field_name!r} has no campaign-wide value", f"/global_values/{field_name}")] return [] def _zip_rule_selection_issues(config: CampaignConfig, archive_ids: set[str]) -> list[SemanticIssue]: issues: list[SemanticIssue] = [] for path, rule, _is_individual in _iter_attachment_rules(config): selection = (rule.zip.archive_id or ZipRuleMode.INHERIT.value).strip() if selection not in {"", ZipRuleMode.INHERIT.value, ZipRuleMode.INCLUDE.value, ZipRuleMode.EXCLUDE.value} and selection not in archive_ids: issues.append(_issue(Severity.ERROR, "zip_archive_unknown", f"Attachment rule selects unknown ZIP archive {selection!r}", f"{path}/zip/archive_id")) return issues def _normalized_zip_archive_name(value: str) -> str: normalized = value.strip().casefold() if not normalized: return "" return normalized if normalized.endswith(".zip") else f"{normalized}.zip" def _ignored_override_issues(config: CampaignConfig, entry: EntryConfig, path_prefix: str) -> list[SemanticIssue]: return [ _issue( Severity.WARNING, "field_override_not_allowed", f"recipient value for field {field_name!r} will be ignored because the field does not allow overrides", f"{path_prefix}/fields/{field_name}", ) for field_name in ignored_entry_field_overrides(config, entry) ] def _global_value_issues(config: CampaignConfig, declared_names: set[str]) -> list[SemanticIssue]: return [ _issue( Severity.WARNING, "unknown_global_value", f"global_values contains {key!r}, but it is not declared in fields", f"/global_values/{key}", ) for key in config.global_values if declared_names and key not in declared_names ] def _active_delivery_entries(config: CampaignConfig) -> list[EntryConfig]: if config.entries.is_inline: return [ entry for entry in (config.entries.inline or []) if entry.active ] return [config.entries.defaults or EntryConfig()] def _delivery_policies(config: CampaignConfig) -> set[DeliveryChannelPolicy]: return { effective_delivery_channel_policy(config, entry) for entry in _active_delivery_entries(config) } def _postbox_target_field_issues( config: CampaignConfig, target: PostboxTargetConfig, path: str, ) -> list[SemanticIssue]: definitions = {field.name: field for field in config.fields} checks = ( ( target.organization_unit_field, FieldType.ORGANIZATION_UNIT, "organization unit", "organization_unit_field", ), ( target.function_field, FieldType.ORGANIZATION_FUNCTION, "organization function", "function_field", ), (target.context_field, None, "context", "context_field"), ) issues: list[SemanticIssue] = [] for field_name, expected_type, label, key in checks: if not field_name: continue definition = definitions.get(field_name) if definition is None: issues.append( _issue( Severity.ERROR, "postbox_target_field_missing", f"Postbox {label} field {field_name!r} is not declared.", f"{path}/{key}", ) ) elif expected_type is not None and definition.type != expected_type: issues.append( _issue( Severity.WARNING, "postbox_target_field_type", ( f"Postbox {label} field {field_name!r} should use " f"field type {expected_type.value!r}." ), f"{path}/{key}", ) ) return issues def _postbox_delivery_issues( config: CampaignConfig, *, postbox_available: bool, ) -> list[SemanticIssue]: issues: list[SemanticIssue] = [] policies = _delivery_policies(config) if not any(policy.uses_postbox for policy in policies): return issues if not postbox_available: issues.append( _issue( Severity.ERROR, "postbox_unavailable", ( "This campaign uses Postbox delivery, but the Postbox " "module and its delivery directory are not active." ), "/delivery/channel_policy", ) ) for entry_index, entry in enumerate(_active_delivery_entries(config)): policy = effective_delivery_channel_policy(config, entry) if not policy.uses_postbox: continue targets = effective_postbox_targets(config, entry) if not targets: issues.append( _issue( Severity.ERROR, "postbox_target_missing", "Postbox delivery requires at least one target.", f"/entries/inline/{entry_index}/postbox_targets", ) ) continue seen_ids: set[str] = set() for target_index, target in enumerate(targets): target_path = ( f"/entries/inline/{entry_index}/postbox_targets/" f"{target_index}" ) if target.id in seen_ids: issues.append( _issue( Severity.WARNING, "postbox_target_id_duplicate", f"Postbox target id {target.id!r} is repeated.", f"{target_path}/id", ) ) seen_ids.add(target.id) issues.extend( _postbox_target_field_issues(config, target, target_path) ) return issues def _delivery_issues( config: CampaignConfig, *, postbox_available: bool, ) -> list[SemanticIssue]: issues: list[SemanticIssue] = [] policies = _delivery_policies(config) uses_mail = any(policy.uses_mail for policy in policies) profile_id = (config.server.mail_profile_id or "").strip() if ( ( config.campaign.mode == "send" and uses_mail or config.delivery.imap_append_sent.enabled ) and not profile_id ): issues.append( _issue( Severity.ERROR, "missing_mail_profile", "Select an authorized Mail-module profile; campaigns cannot store SMTP/IMAP settings or credentials.", "/server/mail_profile_id", ) ) capabilities = config.server.profile_capabilities if ( config.campaign.mode == "send" and uses_mail and profile_id and not capabilities.smtp_available ): issues.append( _issue( Severity.ERROR, "mail_profile_without_smtp", "campaign mode is 'send', but the selected Mail profile has no SMTP configuration", "/server/mail_profile_id", ) ) if config.delivery.imap_append_sent.enabled and profile_id and not capabilities.imap_available: issues.append( _issue( Severity.ERROR, "mail_profile_without_imap", "IMAP append is enabled, but the selected Mail profile has no IMAP configuration", "/server/mail_profile_id", ) ) issues.extend( _postbox_delivery_issues( config, postbox_available=postbox_available, ) ) return issues def _sender_issues(config: CampaignConfig) -> list[SemanticIssue]: """Require Campaign-owned sender data before a send-mode build.""" if ( config.campaign.mode != "send" or not any(policy.uses_mail for policy in _delivery_policies(config)) ): return [] if config.entries.is_inline: return [ _issue( Severity.ERROR, "missing_sender", "No effective From address is configured; Campaign must resolve the sender before building.", f"/entries/inline/{index}/from", ) for index, entry in enumerate(config.entries.inline or []) if ( entry.active and effective_delivery_channel_policy(config, entry).uses_mail and not effective_address_lists(config, entry)["from"] ) ] if config.recipients.from_: return [] defaults = config.entries.defaults mapping_can_supply_sender = bool( config.recipients.allow_individual_from and ( (defaults is not None and defaults.from_) or "from.email" in (config.entries.mapping or {}) ) ) if mapping_can_supply_sender: return [] return [ _issue( Severity.ERROR, "missing_sender", "Configure recipients.from, or allow and map an individual From address, before building a send-mode campaign.", "/recipients/from", ) ] def _entries_validation( config: CampaignConfig, *, campaign_path: Path, check_files: bool, field_names: set[str], field_definitions: dict[str, object], ) -> _EntriesValidation: if config.entries.is_inline: inline_entries = config.entries.inline or [] issues = _inline_entries_issues(config, inline_entries) return _EntriesValidation(mode="inline", count=len(inline_entries), issues=issues) issues = _external_entries_issues( config, campaign_path=campaign_path, check_files=check_files, field_names=field_names, field_definitions=field_definitions, ) source_type = config.entries.source.type.value if config.entries.source else "unknown" return _EntriesValidation(mode=f"external:{source_type}", count=None, issues=issues) def _inline_entries_issues(config: CampaignConfig, inline_entries: list[EntryConfig]) -> list[SemanticIssue]: issues: list[SemanticIssue] = [] if not inline_entries: issues.append(_issue(Severity.WARNING, "no_inline_entries", "entries.inline is empty", "/entries/inline")) for index, entry in enumerate(inline_entries): if entry.active: issues.extend(_ignored_override_issues(config, entry, f"/entries/inline/{index}")) return issues def _external_entries_issues( config: CampaignConfig, *, campaign_path: Path, check_files: bool, field_names: set[str], field_definitions: dict[str, object], ) -> list[SemanticIssue]: issues: list[SemanticIssue] = [] mapping = config.entries.mapping or {} if not mapping: issues.append(_issue(Severity.ERROR, "empty_mapping", "external entries require a non-empty mapping", "/entries/mapping")) issues.extend(_mapping_issues(mapping, field_names, field_definitions)) if config.entries.defaults: issues.extend(_ignored_override_issues(config, config.entries.defaults, "/entries/defaults")) if check_files and config.entries.source: issues.extend(_entries_source_file_issues(config, campaign_path, mapping)) return issues def _mapping_issues( mapping: dict[str, str], field_names: set[str], field_definitions: dict[str, object], ) -> list[SemanticIssue]: issues: list[SemanticIssue] = [] for target in mapping: if not _mapping_target_known(target, field_names): issues.append( _issue( Severity.WARNING, "unknown_mapping_target", f"mapping target {target!r} is not recognized by the current campaign model", f"/entries/mapping/{target}", ) ) field_name = _mapping_target_field_name(target) field = field_definitions.get(field_name or "") if field_name and field is not None and not getattr(field, "can_override", True): issues.append( _issue( Severity.WARNING, "mapping_target_not_overridable", f"mapping target {target!r} points to a field that does not allow recipient overrides; mapped values will be ignored", f"/entries/mapping/{target}", ) ) return issues def _entries_source_file_issues(config: CampaignConfig, campaign_path: Path, mapping: dict[str, str]) -> list[SemanticIssue]: source = config.entries.source if source is None: return [] source_path = _resolve(campaign_path, source.path) if not source_path.exists(): return [ _issue( Severity.ERROR, "entries_source_not_found", f"entries source file does not exist: {source_path}", "/entries/source/path", ) ] if source.type != SourceType.CSV or not source.has_header: return [] try: header = _csv_header(source_path, source.delimiter, source.encoding) except OSError as exc: return [_issue(Severity.ERROR, "entries_source_read_error", str(exc), "/entries/source/path")] header_set = set(header or []) missing_columns = sorted({source_name for source_name in mapping.values() if source_name not in header_set}) if not missing_columns: return [] return [ _issue( Severity.ERROR, "mapping_columns_missing", "CSV mapping refers to missing columns: " + ", ".join(missing_columns), "/entries/mapping", ) ] def _file_check_issues(config: CampaignConfig, campaign_path: Path) -> list[SemanticIssue]: issues: list[SemanticIssue] = [] issues.extend(_attachment_file_check_issues(config, campaign_path)) issues.extend(_attachment_resolution_check_issues(config, campaign_path)) issues.extend(_template_source_file_issues(config, campaign_path)) return issues def _attachment_file_check_issues(config: CampaignConfig, campaign_path: Path) -> list[SemanticIssue]: if config.attachments.base_paths: return [ _issue( Severity.WARNING, "attachments_base_path_not_found", f"attachment base path {base_path_config.name!r} does not exist: {_resolve(campaign_path, base_path_config.path)}", f"/attachments/base_paths/{index}/path", ) for index, base_path_config in enumerate(config.attachments.base_paths) if not _resolve(campaign_path, base_path_config.path).exists() ] attachments_base_path = _resolve(campaign_path, config.attachments.base_path) if attachments_base_path.exists(): return [] return [ _issue( Severity.WARNING, "attachments_base_path_not_found", f"attachments.base_path does not exist: {attachments_base_path}", "/attachments/base_path", ) ] def _attachment_resolution_check_issues(config: CampaignConfig, campaign_path: Path) -> list[SemanticIssue]: try: report = resolve_campaign_attachments(config, campaign_file=campaign_path) except Exception as exc: return [ _issue( Severity.ERROR, "attachment_resolution_failed", f"attachment rules could not be resolved: {exc}", "/attachments", ) ] issues: list[SemanticIssue] = [] for entry in report.entries: entry_path = f"/entries/{entry.entry_id or entry.entry_index}" for issue in entry.issues: if any(issue is attachment_issue for attachment in entry.attachments for attachment_issue in attachment.issues): continue issues.append(_issue(Severity(issue.severity.value), issue.code, issue.message, entry_path)) for attachment in entry.attachments: attachment_path = ( f"/attachments/global/{attachment.index}" if attachment.scope.value == "global" else f"{entry_path}/attachments/{attachment.index}" ) for issue in attachment.issues: issues.append(_issue(Severity(issue.severity.value), issue.code, issue.message, attachment_path)) return issues def _template_source_file_issues(config: CampaignConfig, campaign_path: Path) -> list[SemanticIssue]: issues: list[SemanticIssue] = [] for schema_path, raw_path in _iter_template_source_paths(config): path = _resolve(campaign_path, raw_path) if not path.exists(): issues.append( _issue( Severity.ERROR, "template_source_not_found", f"template source file does not exist: {path}", schema_path, ) ) return issues def validate_campaign_config( config: CampaignConfig, *, campaign_file: str | Path | None = None, check_files: bool = False, postbox_available: bool = False, ) -> SemanticReport: campaign_path = Path(campaign_file).resolve() if campaign_file else Path.cwd() / "campaign.json" issues: list[SemanticIssue] = [] field_names = config.field_names field_definitions = {field.name: field for field in config.fields} declared_names = set(field_definitions) issues.extend(_global_value_issues(config, declared_names)) issues.extend(_attachment_path_issues(config)) issues.extend(_zip_configuration_issues(config)) issues.extend( _delivery_issues( config, postbox_available=postbox_available, ) ) issues.extend(_sender_issues(config)) entries = _entries_validation( config, campaign_path=campaign_path, check_files=check_files, field_names=field_names, field_definitions=field_definitions, ) issues.extend(entries.issues) if check_files: issues.extend(_file_check_issues(config, campaign_path)) report = SemanticReport( campaign_id=config.campaign.id, campaign_name=config.campaign.name, issues=issues, entries_mode=entries.mode, entries_count=entries.count, attachments_base_path=_attachment_base_path_report_value(config), rate_limit=f"{config.delivery.rate_limit.messages_per_minute}/min, concurrency {config.delivery.rate_limit.concurrency}", imap_append_enabled=config.delivery.imap_append_sent.enabled, ) return report