GovOPlaN/govoplan-campaign
1
0
Fork 0
Code Issues 46 Pull Requests Actions Packages Projects Releases Wiki Activity

Release v0.1.2

Browse Source
This commit is contained in:
Albrecht Degering
2026-06-25 19:58:20 +02:00
parent 39ad3500e2
commit 23318c709a
98 changed files with 3432 additions and 2339 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
src/govoplan_campaign/backend/sending/execution.py
View File

@@ -10,7 +10,7 @@ from sqlalchemy.orm import Session
from govoplan_campaign.backend.db.models import Campaign, CampaignJob, CampaignVersion, JobValidationStatus
from govoplan_campaign.backend.campaign.models import DeliveryConfig, ImapConfig, SmtpConfig
from govoplan_mail.backend.mail_profiles import MailProfileError, apply_campaign_credentials, effective_profile_credentials_inherited, ensure_mail_profile_allowed_for_campaign, imap_config_from_profile, mail_profile_id_from_campaign_json, smtp_config_from_profile
from govoplan_campaign.backend.integrations import MailProfileError, mail_integration
SNAPSHOT_VERSION = "3"
@@ -83,8 +83,12 @@ def _redacted_transport_config(config: SmtpConfig | ImapConfig | None) -> SmtpCo
def _transport_password_from_campaign_json(raw_json: dict[str, Any] | None, name: str) -> str | None:
server = raw_json.get("server") if isinstance(raw_json, dict) else None
config = server.get(name) if isinstance(server, dict) else None
credentials = server.get("credentials") if isinstance(server, dict) and isinstance(server.get("credentials"), dict) else None
config = credentials.get(name) if isinstance(credentials, dict) and isinstance(credentials.get(name), dict) else None
password = config.get("password") if isinstance(config, dict) else None
if password is None:
legacy_config = server.get(name) if isinstance(server, dict) else None
password = legacy_config.get("password") if isinstance(legacy_config, dict) else None
if password is None:
return None
return str(password)
@@ -96,14 +100,15 @@ def _server_from_campaign_json(raw_json: dict[str, Any] | None) -> dict[str, Any
def _profile_for_version(session: Session, version: CampaignVersion):
profile_id = mail_profile_id_from_campaign_json(version.raw_json if isinstance(version.raw_json, dict) else {})
mail = mail_integration()
profile_id = mail.mail_profile_id_from_campaign_json(version.raw_json if isinstance(version.raw_json, dict) else {})
if not profile_id:
return None
campaign = session.get(Campaign, version.campaign_id)
if campaign is None:
raise ExecutionSnapshotError("Campaign not found for mail-server profile resolution")
try:
return ensure_mail_profile_allowed_for_campaign(session, tenant_id=campaign.tenant_id, campaign_id=campaign.id, profile_id=profile_id, require_active=True)
return mail.ensure_mail_profile_allowed_for_campaign(session, tenant_id=campaign.tenant_id, campaign_id=campaign.id, profile_id=profile_id, require_active=True)
except MailProfileError as exc:
raise ExecutionSnapshotError(str(exc)) from exc
@@ -117,10 +122,11 @@ def runtime_smtp_config(session: Session, version: CampaignVersion, snapshot: Ex
campaign = session.get(Campaign, version.campaign_id)
if campaign is None:
raise ExecutionSnapshotError("Campaign not found for mail-server profile resolution")
profile_payload = smtp_config_from_profile(profile).model_dump(mode="json")
if effective_profile_credentials_inherited(session, tenant_id=campaign.tenant_id, campaign_id=campaign.id, server=server, protocol="smtp"):
mail = mail_integration()
profile_payload = mail.smtp_config_from_profile(profile).model_dump(mode="json")
if mail.effective_profile_credentials_inherited(session, tenant_id=campaign.tenant_id, campaign_id=campaign.id, server=server, protocol="smtp"):
return SmtpConfig.model_validate(profile_payload)
return SmtpConfig.model_validate(apply_campaign_credentials(profile_payload, server, "smtp"))
return SmtpConfig.model_validate(mail.apply_campaign_credentials(profile_payload, server, "smtp"))
payload["password"] = _transport_password_from_campaign_json(version.raw_json, "smtp")
return SmtpConfig.model_validate(payload)
@@ -131,29 +137,31 @@ def runtime_imap_config(session: Session, version: CampaignVersion, snapshot: Ex
profile = _profile_for_version(session, version)
if profile is None:
return None
imap = imap_config_from_profile(profile)
mail = mail_integration()
imap = mail.imap_config_from_profile(profile)
if imap is None:
return None
campaign = session.get(Campaign, version.campaign_id)
if campaign is None:
raise ExecutionSnapshotError("Campaign not found for mail-server profile resolution")
imap_payload = imap.model_dump(mode="json")
if effective_profile_credentials_inherited(session, tenant_id=campaign.tenant_id, campaign_id=campaign.id, server=server, protocol="imap"):
if mail.effective_profile_credentials_inherited(session, tenant_id=campaign.tenant_id, campaign_id=campaign.id, server=server, protocol="imap"):
return ImapConfig.model_validate(imap_payload)
return ImapConfig.model_validate(apply_campaign_credentials(imap_payload, server, "imap"))
return ImapConfig.model_validate(mail.apply_campaign_credentials(imap_payload, server, "imap"))
payload = snapshot.imap.model_dump(mode="json")
if not payload.get("password"):
profile = _profile_for_version(session, version)
if profile is not None:
imap = imap_config_from_profile(profile)
mail = mail_integration()
imap = mail.imap_config_from_profile(profile)
if imap is not None:
campaign = session.get(Campaign, version.campaign_id)
if campaign is None:
raise ExecutionSnapshotError("Campaign not found for mail-server profile resolution")
imap_payload = imap.model_dump(mode="json")
if effective_profile_credentials_inherited(session, tenant_id=campaign.tenant_id, campaign_id=campaign.id, server=server, protocol="imap"):
if mail.effective_profile_credentials_inherited(session, tenant_id=campaign.tenant_id, campaign_id=campaign.id, server=server, protocol="imap"):
return ImapConfig.model_validate(imap_payload)
return ImapConfig.model_validate(apply_campaign_credentials(imap_payload, server, "imap"))
return ImapConfig.model_validate(mail.apply_campaign_credentials(imap_payload, server, "imap"))
payload["password"] = _transport_password_from_campaign_json(version.raw_json, "imap")
return ImapConfig.model_validate(payload)
@@ -250,7 +258,8 @@ def ensure_execution_snapshot(session: Session, version: CampaignVersion) -> Exe
from govoplan_campaign.backend.persistence.campaigns import load_version_config
_, _, config = load_version_config(session, version.id)
if not config.server.smtp:
runtime_smtp = config.server.runtime_smtp_config()
if not runtime_smtp:
raise ExecutionSnapshotError("Campaign has no SMTP configuration")
jobs = (
session.query(CampaignJob)
@@ -262,8 +271,8 @@ def ensure_execution_snapshot(session: Session, version: CampaignVersion) -> Exe
raise ExecutionSnapshotError("Campaign version has no built jobs; rebuild it before delivery")
payload, digest = create_execution_snapshot(
version,
smtp=config.server.smtp,
imap=config.server.imap,
smtp=runtime_smtp,
imap=config.server.runtime_imap_config(),
delivery=config.delivery,
jobs=jobs,
build_summary=version.build_summary if isinstance(version.build_summary, dict) else {},

72
src/govoplan_campaign/backend/sending/jobs.py
View File

@@ -28,11 +28,15 @@ from govoplan_campaign.backend.db.models import (
SendAttempt,
)
from govoplan_campaign.backend.sending.execution import ExecutionSnapshot, ExecutionSnapshotError, ensure_execution_snapshot, runtime_imap_config, runtime_smtp_config
from govoplan_mail.backend.mail_profiles import MailProfileError, assert_mail_policy_allows_send
from govoplan_mail.backend.sending.rate_limit import wait_for_rate_limit
from govoplan_mail.backend.sending.smtp import SmtpConfigurationError, SmtpSendError, send_email_bytes
from govoplan_mail.backend.sending.imap import ImapAppendError, ImapConfigurationError, append_message_to_sent
from govoplan_files.backend.storage.services import mark_job_attachment_uses_sent
from govoplan_campaign.backend.integrations import (
ImapAppendError,
ImapConfigurationError,
MailProfileError,
SmtpConfigurationError,
SmtpSendError,
files_integration,
mail_integration,
)
class QueueingError(RuntimeError):
@@ -709,7 +713,7 @@ def reconcile_job_outcome(
job.claim_token = None
job.last_error = note or "Operator confirmed SMTP acceptance after an uncertain outcome."
job.imap_status = JobImapStatus.PENDING.value if snapshot.delivery.imap_append_sent.enabled else JobImapStatus.NOT_REQUESTED.value
mark_job_attachment_uses_sent(session, job)
files_integration().mark_job_attachment_uses_sent(session, job)
attempt_status = "reconciled_smtp_accepted"
elif decision == "not_sent":
job.send_status = JobSendStatus.FAILED_TEMPORARY.value
@@ -1054,7 +1058,7 @@ def send_campaign_job(
job = session.get(CampaignJob, job.id)
assert job is not None
wait_for_rate_limit(
mail_integration().wait_for_rate_limit(
key=f"tenant:{job.tenant_id}:campaign:{job.campaign_id}",
messages_per_minute=snapshot.delivery.rate_limit.messages_per_minute,
enabled=use_rate_limit,
@@ -1062,7 +1066,7 @@ def send_campaign_job(
attempt = _record_attempt_start(session, job, claim_token)
try:
smtp_config = runtime_smtp_config(session, version, snapshot)
assert_mail_policy_allows_send(
mail_integration().assert_mail_policy_allows_send(
session,
tenant_id=job.tenant_id,
campaign_id=job.campaign_id,
@@ -1071,7 +1075,7 @@ def send_campaign_job(
from_header=_from_header_from_job(job, snapshot),
recipients=envelope_recipients,
)
result = send_email_bytes(
result = mail_integration().send_email_bytes(
message_bytes,
smtp_config=smtp_config,
envelope_from=envelope_from,
@@ -1098,7 +1102,7 @@ def send_campaign_job(
else:
job.imap_status = JobImapStatus.NOT_REQUESTED.value
job.last_error = refused_warning
mark_job_attachment_uses_sent(session, job)
files_integration().mark_job_attachment_uses_sent(session, job)
session.add(attempt)
session.add(job)
_update_campaign_after_job(session, job.campaign_id, job.campaign_version_id)
@@ -1208,9 +1212,9 @@ def append_sent_for_job(session: Session, *, job_id: str, dry_run: bool = False)
session.commit()
return AppendSentResult(job_id=job.id, status="not_requested", attempt_number=0, dry_run=dry_run)
imap_config = runtime_imap_config(session, version, snapshot)
if not imap_config or not imap_config.enabled:
if not imap_config:
job.imap_status = JobImapStatus.SKIPPED.value
job.last_error = "IMAP append requested, but the execution snapshot has no enabled IMAP configuration"
job.last_error = "IMAP append requested, but the execution snapshot has no IMAP configuration"
session.add(job)
session.commit()
return AppendSentResult(job_id=job.id, status="skipped", attempt_number=0, dry_run=dry_run, message=job.last_error)
@@ -1230,14 +1234,14 @@ def append_sent_for_job(session: Session, *, job_id: str, dry_run: bool = False)
attempt = _record_imap_attempt_start(session, job)
try:
assert_mail_policy_allows_send(
mail_integration().assert_mail_policy_allows_send(
session,
tenant_id=job.tenant_id,
campaign_id=job.campaign_id,
smtp=snapshot.smtp,
imap=imap_config,
)
result = append_message_to_sent(
result = mail_integration().append_message_to_sent(
message_bytes,
imap_config=imap_config,
folder=None if folder == "auto" else folder,
@@ -1246,7 +1250,7 @@ def append_sent_for_job(session: Session, *, job_id: str, dry_run: bool = False)
attempt.folder = result.folder
job.imap_status = JobImapStatus.APPENDED.value
job.last_error = None
mark_job_attachment_uses_sent(session, job)
files_integration().mark_job_attachment_uses_sent(session, job)
session.add(attempt)
session.add(job)
session.commit()
@@ -1263,7 +1267,15 @@ def append_sent_for_job(session: Session, *, job_id: str, dry_run: bool = False)
raise
def enqueue_pending_imap_appends(session: Session, *, tenant_id: str, campaign_id: str, enqueue_celery: bool = True, dry_run: bool = False) -> dict[str, Any]:
def enqueue_pending_imap_appends(
session: Session,
*,
tenant_id: str,
campaign_id: str,
enqueue_celery: bool = True,
run_inline: bool = False,
dry_run: bool = False,
) -> dict[str, Any]:
campaign = _get_campaign_for_tenant(session, campaign_id=campaign_id, tenant_id=tenant_id)
jobs = (
session.query(CampaignJob)
@@ -1276,15 +1288,39 @@ def enqueue_pending_imap_appends(session: Session, *, tenant_id: str, campaign_i
.order_by(CampaignJob.entry_index.asc())
.all()
)
should_enqueue = _should_enqueue_celery(enqueue_celery) and not dry_run
if should_enqueue:
should_enqueue = _should_enqueue_celery(enqueue_celery) and not dry_run and not run_inline
results: list[dict[str, Any]] = []
appended_count = 0
failed_count = 0
skipped_count = 0
if run_inline or dry_run:
for job in jobs:
try:
result = append_sent_for_job(session, job_id=job.id, dry_run=dry_run)
payload = result.as_dict()
results.append(payload)
if result.status == JobImapStatus.APPENDED.value:
appended_count += 1
elif result.status in {"skipped", "not_requested", "not_sent", "already_appended", "dry_run"}:
skipped_count += 1
except Exception as exc: # keep processing later jobs and expose per-job details
failed_count += 1
results.append({"job_id": job.id, "status": "failed", "message": str(exc)})
elif should_enqueue:
for job in jobs:
_celery_enqueue_append_sent_job(job.id)
return {
"campaign_id": campaign.id,
"pending_count": len(jobs),
"enqueued_count": len(jobs) if should_enqueue else 0,
"processed_count": len(results) if run_inline and not dry_run else 0,
"appended_count": appended_count,
"failed_count": failed_count,
"skipped_count": skipped_count,
"dry_run": dry_run,
"run_inline": run_inline,
"results": results,
}
def next_retry_delay(snapshot: ExecutionSnapshot, attempt_count: int) -> int: