from __future__ import annotations import csv import io import logging import math from collections import Counter from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any from sqlalchemy import func from sqlalchemy.orm import Session from govoplan_core.core.policy import ( CAPABILITY_POLICY_PRIVACY_RETENTION, PrivacyRetentionService, ) from govoplan_core.privacy.schemas import PrivacyRetentionPolicyItem from govoplan_core.settings import settings as core_settings from govoplan_campaign.backend.db.models import ( Campaign, CampaignIssue, CampaignJob, CampaignMessageAction, CampaignMessageActionAttempt, CampaignVersion, ImapAppendAttempt, PostboxDeliveryAttempt, SendAttempt, ) from govoplan_campaign.backend.integrations import calendar_integration, postbox_integration from govoplan_campaign.backend.runtime import capability from govoplan_campaign.backend.sending.execution import ExecutionSnapshot from govoplan_campaign.backend.response_security import ( public_campaign_payload, public_delivery_result_message, public_source_filename, ) from govoplan_campaign.backend.services.job_queries import ( _calendar_invitation_state_from_provenance, _calendar_invitations_for_jobs, ) class CampaignReportError(RuntimeError): pass logger = logging.getLogger(__name__) CAMPAIGN_JSON_JOB_LIMIT = 5_000 CAMPAIGN_CSV_JOB_LIMIT = 100_000 def _utcnow_iso() -> str: return datetime.now(timezone.utc).isoformat() def _counter(values: list[str | None]) -> dict[str, int]: return dict(Counter(value or "unknown" for value in values)) def _get_campaign(session: Session, *, tenant_id: str, campaign_id: str) -> Campaign: campaign = session.query(Campaign).filter(Campaign.tenant_id == tenant_id, Campaign.id == campaign_id).one_or_none() if not campaign: raise CampaignReportError(f"Campaign not found or not accessible: {campaign_id}") return campaign def _selected_version( session: Session, campaign: Campaign, version_id: str | None = None, ) -> CampaignVersion | None: wanted = version_id or campaign.current_version_id if not wanted: return None version = session.get(CampaignVersion, wanted) if not version or version.campaign_id != campaign.id: raise CampaignReportError(f"Campaign version not found or not part of campaign: {wanted}") return version def _version_info( version: CampaignVersion | None, *, include_diagnostics: bool = False, ) -> dict[str, Any] | None: if not version: return None return { "id": version.id, "version_number": version.version_number, "schema_version": version.schema_version, "source_filename": public_source_filename(version.source_filename), "created_at": version.created_at.isoformat() if version.created_at else None, "validation_summary": public_campaign_payload( version.validation_summary, include_diagnostics=include_diagnostics, ), "build_summary": public_campaign_payload( version.build_summary, include_diagnostics=include_diagnostics, ), "execution_snapshot_hash": version.execution_snapshot_hash, "execution_snapshot_at": version.execution_snapshot_at.isoformat() if version.execution_snapshot_at else None, "delivery_mode": version.delivery_mode, "delivery_mode_selected_at": ( version.delivery_mode_selected_at.isoformat() if version.delivery_mode_selected_at else None ), } def _load_delivery_info( version: CampaignVersion | None, jobs: list[CampaignJob] | None = None, *, pending_job_count: int | None = None, include_diagnostics: bool = False, ) -> dict[str, Any]: """Read deterministic delivery settings from the immutable execution snapshot.""" default = { "rate_limit": {"messages_per_minute": None, "concurrency": None}, "imap_append_sent": {"enabled": None, "folder": None}, "retry": {"max_attempts": None, "backoff_seconds": []}, "execution_snapshot_hash": None, "execution_snapshot_at": None, "snapshot_version": None, "built_at": None, "job_count": 0, "queueable_job_count": 0, "estimated_remaining_send_seconds": None, "estimated_remaining_send_human": None, "delivery_mode": getattr(version, "delivery_mode", None) if version else None, "delivery_mode_selected_at": ( getattr(version, "delivery_mode_selected_at", None).isoformat() if version and getattr(version, "delivery_mode_selected_at", None) else None ), } if include_diagnostics: default.update( { "build_token": None, # nosec B105 - absent report value, not a credential. "job_manifest_sha256": None, "effective_policy_sha256": None, "smtp_transport_revision": None, "imap_transport_revision": None, "background_workers_enabled": bool(core_settings.celery_enabled), } ) if not version or not isinstance(version.execution_snapshot, dict): default["load_error"] = "No execution snapshot exists; rebuild the current locked version before delivery." return default try: snapshot = ExecutionSnapshot.model_validate(version.execution_snapshot) except Exception: # pragma: no cover - reporting should remain available logger.warning("Campaign execution snapshot could not be loaded for a report") default["load_error"] = "Execution snapshot is invalid; rebuild the selected version before delivery." return default messages_per_minute = snapshot.delivery.rate_limit.messages_per_minute if pending_job_count is None: pending_job_count = sum( 1 for job in jobs or [] if job.send_status in {"queued", "claimed", "sending"} ) estimated_seconds = None if messages_per_minute and pending_job_count: estimated_seconds = int( math.ceil((pending_job_count / messages_per_minute) * 60) ) result = { "rate_limit": { "messages_per_minute": messages_per_minute, "concurrency": snapshot.delivery.rate_limit.concurrency, }, "imap_append_sent": { "enabled": snapshot.delivery.imap_append_sent.enabled, "folder": snapshot.delivery.imap_append_sent.folder, }, "retry": { "max_attempts": snapshot.delivery.retry.max_attempts, "backoff_seconds": snapshot.delivery.retry.backoff_seconds, }, "execution_snapshot_hash": version.execution_snapshot_hash, "execution_snapshot_at": version.execution_snapshot_at.isoformat() if version.execution_snapshot_at else None, "snapshot_version": snapshot.snapshot_version, "built_at": snapshot.built_at, "job_count": snapshot.job_count, "queueable_job_count": snapshot.queueable_job_count, "estimated_remaining_send_seconds": estimated_seconds, "estimated_remaining_send_human": _human_duration(estimated_seconds), "delivery_mode": version.delivery_mode, "delivery_mode_selected_at": ( version.delivery_mode_selected_at.isoformat() if version.delivery_mode_selected_at else None ), } if include_diagnostics: result.update( { "build_token": snapshot.build_token, "job_manifest_sha256": snapshot.job_manifest_sha256, "effective_policy_sha256": snapshot.effective_policy_sha256, "smtp_transport_revision": snapshot.smtp_transport_revision, "imap_transport_revision": snapshot.imap_transport_revision, "background_workers_enabled": bool(core_settings.celery_enabled), } ) return result def _human_duration(seconds: int | None) -> str | None: if seconds is None: return None if seconds < 60: return f"{seconds}s" minutes, sec = divmod(seconds, 60) if minutes < 60: return f"{minutes}m {sec}s" if sec else f"{minutes}m" hours, minute = divmod(minutes, 60) return f"{hours}h {minute}m" if minute else f"{hours}h" def _issue_summary_from_jobs(jobs: list[CampaignJob]) -> dict[str, Any]: severity_counter: Counter[str] = Counter() code_counter: Counter[str] = Counter() behavior_counter: Counter[str] = Counter() total = 0 for job in jobs: for issue in job.issues_snapshot or []: if not isinstance(issue, dict): continue total += 1 severity_counter[issue.get("severity") or "unknown"] += 1 code_counter[issue.get("code") or "unknown"] += 1 if issue.get("behavior"): behavior_counter[issue["behavior"]] += 1 return { "total": total, "by_severity": dict(severity_counter), "by_code": dict(code_counter), "by_behavior": dict(behavior_counter), } def _attachment_summary(jobs: list[CampaignJob]) -> dict[str, Any]: status_counter: Counter[str] = Counter() behavior_counter: Counter[str] = Counter() total_configs = 0 total_matched_files = 0 zip_enabled = 0 missing = 0 ambiguous = 0 for job in jobs: for attachment in job.resolved_attachments or []: if not isinstance(attachment, dict): continue total_configs += 1 status = attachment.get("status") or "unknown" status_counter[status] += 1 if attachment.get("behavior"): behavior_counter[attachment["behavior"]] += 1 matches = attachment.get("matches") or [] if isinstance(matches, list): total_matched_files += len(matches) if attachment.get("zip_enabled"): zip_enabled += 1 if status == "missing": missing += 1 if status == "ambiguous": ambiguous += 1 return { "total_attachment_configs": total_configs, "total_matched_files": total_matched_files, "zip_enabled_configs": zip_enabled, "missing_configs": missing, "ambiguous_configs": ambiguous, "by_status": dict(status_counter), "by_behavior": dict(behavior_counter), } @dataclass(slots=True) class _JobReportAggregate: total: int = 0 pending: int = 0 queueable: int = 0 queueable_unattempted: int = 0 retryable: int = 0 cancellable: int = 0 needs_attention: int = 0 build: Counter[str] = field(default_factory=Counter) validation: Counter[str] = field(default_factory=Counter) queue: Counter[str] = field(default_factory=Counter) send: Counter[str] = field(default_factory=Counter) postbox: Counter[str] = field(default_factory=Counter) print: Counter[str] = field(default_factory=Counter) imap: Counter[str] = field(default_factory=Counter) issue_total: int = 0 issue_severity: Counter[str] = field(default_factory=Counter) issue_code: Counter[str] = field(default_factory=Counter) issue_behavior: Counter[str] = field(default_factory=Counter) attachment_total: int = 0 attachment_matches: int = 0 attachment_zip_enabled: int = 0 attachment_missing: int = 0 attachment_ambiguous: int = 0 attachment_status: Counter[str] = field(default_factory=Counter) attachment_behavior: Counter[str] = field(default_factory=Counter) eml_retained: int = 0 eml_expired: int = 0 eml_not_generated: int = 0 recent_failures: list[CampaignJob] = field(default_factory=list) def add( self, job: CampaignJob, *, retry_max_attempts: int | None, include_recent_failures: bool, ) -> None: self.total += 1 self._add_statuses(job) self._add_delivery_counts(job, retry_max_attempts=retry_max_attempts) self._add_issues(job) self._add_attachments(job) self._add_eml_evidence(job) self._add_recent_failure(job, include_recent_failures=include_recent_failures) def _add_statuses(self, job: CampaignJob) -> None: self.build[job.build_status or "unknown"] += 1 self.validation[job.validation_status or "unknown"] += 1 self.queue[job.queue_status or "unknown"] += 1 self.send[job.send_status or "unknown"] += 1 self.postbox[job.postbox_status or "unknown"] += 1 self.print[getattr(job, "print_status", None) or "unknown"] += 1 self.imap[job.imap_status or "unknown"] += 1 def _add_delivery_counts(self, job: CampaignJob, *, retry_max_attempts: int | None) -> None: if job.send_status in {"queued", "claimed", "sending"}: self.pending += 1 if _job_is_queueable(job): self.queueable += 1 if _job_is_queueable_unattempted(job): self.queueable_unattempted += 1 if _job_is_retryable(job, retry_max_attempts=retry_max_attempts): self.retryable += 1 if _job_is_cancellable(job): self.cancellable += 1 if _job_needs_attention(job): self.needs_attention += 1 def _add_recent_failure(self, job: CampaignJob, *, include_recent_failures: bool) -> None: if not include_recent_failures or not _job_is_recent_failure(job): return self.recent_failures.append(job) self.recent_failures.sort(key=lambda item: item.updated_at or item.created_at, reverse=True) del self.recent_failures[20:] def _add_issues(self, job: CampaignJob) -> None: for issue in job.issues_snapshot or []: if not isinstance(issue, dict): continue self.issue_total += 1 self.issue_severity[issue.get("severity") or "unknown"] += 1 self.issue_code[issue.get("code") or "unknown"] += 1 if issue.get("behavior"): self.issue_behavior[str(issue["behavior"])] += 1 def _add_attachments(self, job: CampaignJob) -> None: for attachment in job.resolved_attachments or []: if not isinstance(attachment, dict): continue self.attachment_total += 1 status_value = str(attachment.get("status") or "unknown") self.attachment_status[status_value] += 1 if attachment.get("behavior"): self.attachment_behavior[str(attachment["behavior"])] += 1 matches = attachment.get("matches") if isinstance(matches, list): self.attachment_matches += len(matches) if attachment.get("zip_enabled"): self.attachment_zip_enabled += 1 if status_value == "missing": self.attachment_missing += 1 if status_value == "ambiguous": self.attachment_ambiguous += 1 def _add_eml_evidence(self, job: CampaignJob) -> None: if not job.eml_sha256: self.eml_not_generated += 1 elif job.eml_local_path or job.eml_storage_key: self.eml_retained += 1 else: self.eml_expired += 1 NON_CANCELLABLE_SEND_STATUSES = { "skipped", "smtp_accepted", "postbox_accepted", "print_accepted", "delivered", "partially_accepted", "sent", "outcome_unknown", "claimed", "sending", "cancelled", } def _job_is_queueable(job: CampaignJob) -> bool: return job.validation_status in {"ready", "warning"} and job.build_status == "built" def _job_is_queueable_unattempted(job: CampaignJob) -> bool: return ( job.attempt_count == 0 and job.postbox_attempt_count == 0 and getattr(job, "print_attempt_count", 0) == 0 and job.send_status in {"not_queued", "cancelled"} and _job_is_queueable(job) ) def _job_is_retryable(job: CampaignJob, *, retry_max_attempts: int | None) -> bool: attempts_available = retry_max_attempts is None or job.attempt_count < retry_max_attempts return job.send_status in {"failed_temporary", "partially_accepted"} and attempts_available def _job_is_cancellable(job: CampaignJob) -> bool: return job.send_status not in NON_CANCELLABLE_SEND_STATUSES def _job_needs_attention(job: CampaignJob) -> bool: return ( job.validation_status in {"needs_review", "blocked"} or job.send_status in { "failed_temporary", "failed_permanent", "partially_accepted", "outcome_unknown", "claimed", "sending", } or job.postbox_status in { "partially_accepted", "rejected_temporary", "rejected_permanent", "outcome_unknown", } or getattr(job, "print_status", "not_requested") == "failed" or job.imap_status == "failed" ) def _job_is_recent_failure(job: CampaignJob) -> bool: return bool( job.last_error or str(job.send_status).startswith("failed") or job.send_status == "partially_accepted" or job.postbox_status in { "partially_accepted", "rejected_temporary", "rejected_permanent", "outcome_unknown", } or getattr(job, "print_status", "not_requested") == "failed" or job.imap_status == "failed" ) def _recent_failures(jobs: list[CampaignJob], *, limit: int = 20) -> list[dict[str, Any]]: failed = [job for job in jobs if _job_is_recent_failure(job)] failed.sort(key=lambda job: job.updated_at or job.created_at, reverse=True) return [ { "job_id": job.id, "entry_index": job.entry_index, "entry_id": job.entry_id, "recipient_email": job.recipient_email, "validation_status": job.validation_status, "send_status": job.send_status, "postbox_status": job.postbox_status, "imap_status": job.imap_status, "attempt_count": job.attempt_count, "postbox_attempt_count": job.postbox_attempt_count, "last_error": public_delivery_result_message( last_error=job.last_error, send_status=job.send_status, imap_status=job.imap_status, postbox_status=job.postbox_status, ), "updated_at": job.updated_at.isoformat() if job.updated_at else None, } for job in failed[:limit] ] def _review_decisions_by_job( version: CampaignVersion | None, ) -> dict[str, dict[str, Any]]: if version is None: return {} editor_state = ( getattr(version, "editor_state", None) if isinstance(getattr(version, "editor_state", None), dict) else {} ) review_state = editor_state.get("review_send") build_summary = ( getattr(version, "build_summary", None) if isinstance(getattr(version, "build_summary", None), dict) else {} ) build_token = str( build_summary.get("build_token") or build_summary.get("built_at") or "" ) if ( not isinstance(review_state, dict) or review_state.get("inspection_complete") is not True or not build_token or str(review_state.get("build_token") or "") != build_token ): return {} decisions: dict[str, dict[str, Any]] = {} for item in review_state.get("issue_decisions") or []: if not isinstance(item, dict): continue job_id = str(item.get("job_id") or "").strip() if job_id and job_id not in decisions: decisions[job_id] = item return decisions def _public_review_decision( decision: dict[str, Any] | None, *, include_diagnostics: bool, ) -> dict[str, Any] | None: if not decision: return None result = { "decision": decision.get("decision"), "reason": decision.get("reason"), "actor_user_id": decision.get("actor_user_id"), "decided_at": decision.get("decided_at"), "issue_codes": list(decision.get("issue_codes") or []), } if include_diagnostics: result.update( { "review_key": decision.get("review_key"), "message_sha256": decision.get("message_sha256"), "issue_fingerprint": decision.get("issue_fingerprint"), } ) return result def _review_decision_summary( decisions: dict[str, dict[str, Any]], ) -> dict[str, Any]: issue_codes: Counter[str] = Counter() for decision in decisions.values(): issue_codes.update( str(code) for code in decision.get("issue_codes") or [] if code ) return { "exception_decision_count": len(decisions), "by_issue_code": dict(issue_codes), } def _job_row( job: CampaignJob, *, include_diagnostics: bool = False, review_decision: dict[str, Any] | None = None, calendar_invitation: dict[str, object] | None = None, ) -> dict[str, Any]: invitation = calendar_invitation or _calendar_invitation_state_from_provenance(job) row = { "job_id": job.id, "entry_index": job.entry_index, "entry_id": job.entry_id, "recipient_email": job.recipient_email, "subject": job.subject, "build_status": job.build_status, "validation_status": job.validation_status, "queue_status": job.queue_status, "send_status": job.send_status, "delivery_channel_policy": getattr( job, "delivery_channel_policy", "mail", ), "postbox_status": getattr( job, "postbox_status", "not_requested", ), "print_status": getattr(job, "print_status", "not_requested"), "imap_status": job.imap_status, "attempt_count": job.attempt_count, "postbox_attempt_count": getattr(job, "postbox_attempt_count", 0), "print_attempt_count": getattr(job, "print_attempt_count", 0), "postbox_target_count": len( getattr(job, "resolved_postbox_targets", None) or [] ), "print_output": public_campaign_payload( getattr(job, "resolved_print_output", None) ), "delivery_provenance": public_campaign_payload( getattr(job, "delivery_provenance", None) or {} ), "queued_at": job.queued_at.isoformat() if job.queued_at else None, "outcome_unknown_at": job.outcome_unknown_at.isoformat() if job.outcome_unknown_at else None, "sent_at": job.sent_at.isoformat() if job.sent_at else None, "last_error": public_delivery_result_message( last_error=job.last_error, send_status=job.send_status, imap_status=job.imap_status, postbox_status=getattr(job, "postbox_status", "not_requested"), ), "eml_size_bytes": job.eml_size_bytes, "eml_sha256": job.eml_sha256, "issues_count": len(job.issues_snapshot or []), "attachment_config_count": len(job.resolved_attachments or []), "matched_file_count": sum(len(item.get("matches") or []) for item in (job.resolved_attachments or []) if isinstance(item, dict)), "review_decision": _public_review_decision( review_decision, include_diagnostics=include_diagnostics, ), "calendar_invitation": invitation, "calendar_rsvp_status": ( invitation.get("rsvp_status") if invitation is not None else None ), } if include_diagnostics: row.update( { "claimed_at": job.claimed_at.isoformat() if job.claimed_at else None, "smtp_started_at": job.smtp_started_at.isoformat() if job.smtp_started_at else None, } ) return row def _address_summary(value: Any) -> str: if value is None: return "" if isinstance(value, str): return value if isinstance(value, dict): email = str(value.get("email") or "").strip() name = str(value.get("name") or "").strip() if name and email: return f"{name} <{email}>" return email or name if isinstance(value, list): return "; ".join(item for item in (_address_summary(item) for item in value) if item) return str(value) def _latest_by_job_id(attempts: list[Any]) -> dict[str, Any]: latest: dict[str, Any] = {} for attempt in attempts: current = latest.get(attempt.job_id) if current is None or (attempt.attempt_number or 0) >= (current.attempt_number or 0): latest[attempt.job_id] = attempt return latest def _attachment_names(attachments: list[dict[str, Any]] | None) -> str: names: list[str] = [] for attachment in attachments or []: if not isinstance(attachment, dict): continue for key in ("message_filenames", "zip_entry_names"): values = attachment.get(key) if isinstance(values, list): names.extend(str(value) for value in values if value) zip_filename = attachment.get("zip_filename") if zip_filename: names.append(str(zip_filename)) matches = attachment.get("matches") if isinstance(matches, list): for match in matches: if isinstance(match, dict): filename = match.get("filename") or match.get("name") or match.get("display_name") if filename: names.append(str(filename)) elif match: names.append(str(match).rsplit("/", 1)[-1]) seen: set[str] = set() deduplicated = [] for name in names: if name and name not in seen: seen.add(name) deduplicated.append(name) return "; ".join(deduplicated) def _job_evidence_row( job: CampaignJob, *, latest_smtp: SendAttempt | None = None, latest_imap: ImapAppendAttempt | None = None, latest_message_action: CampaignMessageAction | None = None, include_diagnostics: bool = False, review_decision: dict[str, Any] | None = None, calendar_invitation: dict[str, object] | None = None, ) -> dict[str, Any]: row = _job_row( job, include_diagnostics=include_diagnostics, review_decision=review_decision, calendar_invitation=calendar_invitation, ) public_decision = row.pop("review_decision") or {} invitation = row.pop("calendar_invitation") or {} recipients = job.resolved_recipients or {} row.update({ "campaign_id": job.campaign_id, "campaign_version_id": job.campaign_version_id, "message_id_header": job.message_id_header, **_job_evidence_addresses(recipients), "postbox_targets": _job_postbox_target_summary(job), **_job_print_and_route_evidence(job), "attachment_names": _attachment_names(job.resolved_attachments), **_job_attempt_evidence(latest_smtp=latest_smtp, latest_imap=latest_imap), "latest_message_action_kind": ( latest_message_action.kind if latest_message_action else None ), "latest_message_action_status": ( latest_message_action.status if latest_message_action else None ), "latest_message_action_id": ( latest_message_action.id if latest_message_action else None ), "latest_message_action_at": ( _iso_timestamp(latest_message_action.completed_at) if latest_message_action else None ), "review_decision": public_decision.get("decision"), "review_reason": public_decision.get("reason"), "review_actor_user_id": public_decision.get("actor_user_id"), "review_decided_at": public_decision.get("decided_at"), "review_issue_codes": "; ".join( str(code) for code in public_decision.get("issue_codes") or [] ), "calendar_invitation_state": invitation.get("state"), "calendar_rsvp_status": invitation.get("rsvp_status"), "calendar_event_id": invitation.get("event_id"), "calendar_id": invitation.get("calendar_id"), "calendar_uid": invitation.get("uid"), "calendar_external_state": invitation.get("external_state"), "calendar_reply_ingress": invitation.get("reply_ingress"), }) if include_diagnostics: row.update( { "review_key": public_decision.get("review_key"), "review_message_sha256": public_decision.get( "message_sha256" ), "review_issue_fingerprint": public_decision.get( "issue_fingerprint" ), } ) return row def _job_evidence_addresses(recipients: dict[str, Any]) -> dict[str, str]: return {key: _address_summary(recipients.get(key)) for key in ("from", "to", "cc", "bcc", "reply_to")} def _job_postbox_target_summary(job: CampaignJob) -> str: targets = getattr(job, "resolved_postbox_targets", None) or [] return "; ".join( str(target.get("address") or target.get("name") or target.get("postbox_id") or "") for target in targets if isinstance(target, dict) ) def _job_print_and_route_evidence(job: CampaignJob) -> dict[str, Any]: output = ( job.resolved_print_output if isinstance(getattr(job, "resolved_print_output", None), dict) else {} ) artifact = output.get("artifact") if isinstance(output.get("artifact"), dict) else {} provenance = ( job.delivery_provenance if isinstance(getattr(job, "delivery_provenance", None), dict) else {} ) selected_route = ( provenance.get("selected_route") if isinstance(provenance.get("selected_route"), dict) else {} ) return { "distribution_list_id": provenance.get("list_id"), "distribution_list_revision": provenance.get("list_revision"), "distribution_snapshot_id": provenance.get("snapshot_id"), "distribution_expansion_hash": provenance.get("expansion_hash"), "distribution_recipient_key": provenance.get("recipient_key"), "selected_route_channel": selected_route.get("channel"), "selected_route_target_key": selected_route.get("target_key"), "route_reason": provenance.get("route_reason"), "print_render_id": output.get("render_id"), "print_template_id": output.get("template_id"), "print_template_revision_id": output.get("template_revision_id"), "print_template_hash": output.get("template_hash"), "print_input_hash": output.get("input_hash"), "print_output_sha256": output.get("output_sha256"), "print_artifact_kind": artifact.get("kind"), "print_artifact_file_id": artifact.get("file_asset_id"), "print_artifact_download_path": artifact.get("download_path"), "print_actor_account_id": output.get("actor_account_id"), } def _iso_timestamp(value: datetime | None) -> str | None: return value.isoformat() if value else None def _job_attempt_evidence( *, latest_smtp: SendAttempt | None, latest_imap: ImapAppendAttempt | None, ) -> dict[str, Any]: return { "latest_smtp_attempt_number": latest_smtp.attempt_number if latest_smtp else None, "latest_smtp_status": latest_smtp.status if latest_smtp else None, "latest_smtp_status_code": latest_smtp.smtp_status_code if latest_smtp else None, "latest_smtp_started_at": _iso_timestamp(latest_smtp.started_at) if latest_smtp else None, "latest_smtp_finished_at": _iso_timestamp(latest_smtp.finished_at) if latest_smtp else None, "latest_imap_attempt_number": latest_imap.attempt_number if latest_imap else None, "latest_imap_status": latest_imap.status if latest_imap else None, "latest_imap_folder": latest_imap.folder if latest_imap else None, "latest_imap_created_at": _iso_timestamp(latest_imap.created_at) if latest_imap else None, "latest_imap_updated_at": _iso_timestamp(latest_imap.updated_at) if latest_imap else None, } def generate_campaign_report( session: Session, *, tenant_id: str, campaign_id: str, version_id: str | None = None, include_jobs: bool = False, include_recent_failures: bool = False, include_diagnostics: bool = False, ) -> dict[str, Any]: """Generate a dashboard/report payload for one campaign. The shape is intentionally web-UI friendly: status counters for cards, issue/attachment summaries for review panels, and optional job rows for tables/export. """ campaign = _get_campaign(session, tenant_id=tenant_id, campaign_id=campaign_id) version = _selected_version(session, campaign, version_id) jobs = _report_jobs( session, tenant_id=tenant_id, campaign_id=campaign.id, version=version, ) aggregate = _JobReportAggregate() job_rows: list[dict[str, Any]] = [] included_jobs: list[CampaignJob] = [] review_decisions = _review_decisions_by_job(version) retry_max_attempts = _retry_max_attempts(version) iterator = jobs.yield_per(500) if hasattr(jobs, "yield_per") else jobs for job in iterator: aggregate.add( job, retry_max_attempts=retry_max_attempts, include_recent_failures=include_recent_failures, ) if include_jobs: if len(included_jobs) >= CAMPAIGN_JSON_JOB_LIMIT: raise CampaignReportError( "The recipient-level JSON report exceeds the safe row " f"limit of {CAMPAIGN_JSON_JOB_LIMIT}; use the CSV export " "or the paginated recipient table." ) included_jobs.append(job) report = _campaign_report_payload( session, campaign=campaign, version=version, aggregate=aggregate, tenant_id=tenant_id, include_recent_failures=include_recent_failures, include_diagnostics=include_diagnostics, review_decisions=review_decisions, ) if include_jobs: calendar_invitations = _calendar_invitations_for_jobs( session, included_jobs, ) job_rows.extend( _job_row( job, include_diagnostics=include_diagnostics, review_decision=review_decisions.get(job.id), calendar_invitation=calendar_invitations.get(job.id), ) for job in included_jobs ) report["jobs"] = job_rows return report def _report_jobs( session: Session, *, tenant_id: str, campaign_id: str, version: CampaignVersion | None, ) -> Any: jobs_query = session.query(CampaignJob).filter( CampaignJob.tenant_id == tenant_id, CampaignJob.campaign_id == campaign_id, ) if version: jobs_query = jobs_query.filter(CampaignJob.campaign_version_id == version.id) else: jobs_query = jobs_query.filter(False) return jobs_query.order_by(CampaignJob.entry_index.asc()) def _campaign_report_payload( session: Session, *, campaign: Campaign, version: CampaignVersion | None, aggregate: _JobReportAggregate, tenant_id: str, include_recent_failures: bool, include_diagnostics: bool, review_decisions: dict[str, dict[str, Any]] | None = None, ) -> dict[str, Any]: postbox_receipts = _campaign_postbox_receipt_summary( session, tenant_id=tenant_id, campaign_id=campaign.id, version=version, ) calendar_invitations = calendar_integration().summarize_invitations( session, tenant_id=tenant_id, source_resource_id=version.id if version else None, ) report = { "generated_at": _utcnow_iso(), "campaign": _campaign_report_campaign_payload(campaign), "current_version": _version_info(version, include_diagnostics=include_diagnostics), "selected_version_id": version.id if version else None, "cards": _campaign_report_cards_from_aggregate(version, aggregate), "status_counts": _campaign_report_status_counts_from_aggregate( version, aggregate, ), "issues": { "total": aggregate.issue_total, "by_severity": dict(aggregate.issue_severity), "by_code": dict(aggregate.issue_code), "by_behavior": dict(aggregate.issue_behavior), "persisted_campaign_issue_count": _persisted_campaign_issue_count( session, tenant_id=tenant_id, campaign_id=campaign.id, version=version, ), }, "attachments": { "total_attachment_configs": aggregate.attachment_total, "total_matched_files": aggregate.attachment_matches, "zip_enabled_configs": aggregate.attachment_zip_enabled, "missing_configs": aggregate.attachment_missing, "ambiguous_configs": aggregate.attachment_ambiguous, "by_status": dict(aggregate.attachment_status), "by_behavior": dict(aggregate.attachment_behavior), }, "review": _review_decision_summary(review_decisions or {}), "attempts": _campaign_report_attempt_counts( session, tenant_id=tenant_id, campaign_id=campaign.id, version=version, ), "postbox_receipts": postbox_receipts, "calendar_invitations": calendar_invitations, "message_actions": _campaign_message_action_summary( session, tenant_id=tenant_id, campaign_id=campaign.id, version=version, ), "delivery": _load_delivery_info( version, pending_job_count=aggregate.pending, include_diagnostics=include_diagnostics, ), "retention": _campaign_retention_projection( session, campaign=campaign, version=version, aggregate=aggregate, postbox_receipts=postbox_receipts, ), } if include_recent_failures: report["recent_failures"] = _recent_failures( aggregate.recent_failures, ) return report def _campaign_retention_projection( session: Session, *, campaign: Campaign, version: CampaignVersion | None, aggregate: _JobReportAggregate, postbox_receipts: dict[str, object], ) -> dict[str, Any]: service = capability(CAPABILITY_POLICY_PRIVACY_RETENTION) policy_status = "configured" policy_reason: str | None = None sources: list[dict[str, Any]] = [] if service is None: policy_status = "defaults" policy_reason = ( "The Policy module is not active. Platform defaults apply, but " "automated retention enforcement is unavailable." ) policy = PrivacyRetentionPolicyItem().model_dump(mode="json") sources = [{ "scope_type": "system", "path": "platform-defaults", "label": "Platform defaults", "applied_fields": list(policy), }] elif not isinstance(service, PrivacyRetentionService): policy_status = "unavailable" policy_reason = "The configured retention provider does not satisfy the platform contract." policy = PrivacyRetentionPolicyItem().model_dump(mode="json") else: try: policy = _model_payload( service.effective_privacy_policy( session, campaign_id=campaign.id, ) ) sources = _public_retention_sources( service.effective_privacy_policy_sources( session, campaign_id=campaign.id, ) ) except Exception: # pragma: no cover - report remains readable on provider failure. logger.exception("Effective Campaign retention policy could not be resolved") policy_status = "unavailable" policy_reason = "The effective retention policy could not be resolved." policy = PrivacyRetentionPolicyItem().model_dump(mode="json") evidence = _retention_evidence_state( version=version, aggregate=aggregate, postbox_receipts=postbox_receipts, ) minimized = [ name for name, item in evidence.items() if item.get("state") in {"redacted", "partially_redacted", "expired", "partially_expired"} ] retained = [ name for name, item in evidence.items() if item.get("state") in {"retained", "partially_expired"} ] impact_state = ( "policy_unavailable" if policy_status == "unavailable" else "partially_minimized" if minimized else "retained" ) return { "policy_status": policy_status, "policy_reason": policy_reason, "effective_policy": policy, "sources": sources, "evidence": evidence, "privacy_impact": { "state": impact_state, "summary": ( "Campaign reports expose delivery and recipient evidence only " "to authorized report readers. The effective retention policy " "controls when source JSON, generated message files, and stored " "report detail are removed or redacted; aggregate counters and " "audit references may remain." ), "retained_categories": retained, "minimized_categories": minimized, }, } def _model_payload(value: object) -> dict[str, Any]: if hasattr(value, "model_dump"): payload = value.model_dump(mode="json") return dict(payload) if isinstance(payload, dict) else {} return dict(value) if isinstance(value, dict) else {} def _public_retention_sources(value: object) -> list[dict[str, Any]]: if not isinstance(value, (list, tuple)): return [] sources: list[dict[str, Any]] = [] for item in value: if not isinstance(item, dict): continue sources.append({ "scope_type": item.get("scope_type"), "path": item.get("path"), "label": item.get("label"), "applied_fields": [ str(field) for field in item.get("applied_fields") or [] if field ], }) return sources def _retention_evidence_state( *, version: CampaignVersion | None, aggregate: _JobReportAggregate, postbox_receipts: dict[str, object], ) -> dict[str, dict[str, Any]]: if version is None: return { key: {"state": "not_applicable"} for key in ( "raw_campaign_json", "stored_report_detail", "generated_eml", "postbox_copies", ) } raw_marker = _retention_marker(version.raw_json, "raw_json_redacted") validation_marker = _retention_marker( version.validation_summary, "report_detail_redacted", ) build_marker = _retention_marker( version.build_summary, "report_detail_redacted", ) redacted_summaries = sum(bool(marker) for marker in (validation_marker, build_marker)) generated_total = aggregate.eml_retained + aggregate.eml_expired generated_state = ( "not_generated" if generated_total == 0 else "expired" if aggregate.eml_expired == generated_total else "partially_expired" if aggregate.eml_expired else "retained" ) expired_postbox = int(postbox_receipts.get("expired_message_count") or 0) withdrawn_postbox = int(postbox_receipts.get("withdrawn_message_count") or 0) readable_postbox = int(postbox_receipts.get("currently_readable_delivery_count") or 0) postbox_state = ( "not_applicable" if postbox_receipts.get("status") == "not_applicable" else "unavailable" if postbox_receipts.get("status") == "unavailable" else "partially_expired" if (expired_postbox or withdrawn_postbox) and readable_postbox else "expired" if expired_postbox or withdrawn_postbox else "retained" ) return { "raw_campaign_json": { "state": "redacted" if raw_marker else "retained", "redacted_at": raw_marker.get("redacted_at") if raw_marker else None, }, "stored_report_detail": { "state": ( "redacted" if redacted_summaries == 2 else "partially_redacted" if redacted_summaries else "retained" ), "redacted_summary_count": redacted_summaries, "summary_count": 2, "redacted_at": next( ( marker.get("redacted_at") for marker in (validation_marker, build_marker) if marker and marker.get("redacted_at") ), None, ), }, "generated_eml": { "state": generated_state, "retained_count": aggregate.eml_retained, "expired_count": aggregate.eml_expired, "not_generated_count": aggregate.eml_not_generated, }, "postbox_copies": { "state": postbox_state, "currently_readable_count": readable_postbox, "expired_count": expired_postbox, "withdrawn_count": withdrawn_postbox, }, } def _retention_marker(value: object, key: str) -> dict[str, Any] | None: if not isinstance(value, dict): return None marker = value.get("_retention") return dict(marker) if isinstance(marker, dict) and marker.get(key) else None def _campaign_postbox_receipt_summary( session: Session, *, tenant_id: str, campaign_id: str, version: CampaignVersion | None, ) -> dict[str, object]: if version is None: return {"status": "not_applicable", "delivery_count": 0} rows = ( session.query(PostboxDeliveryAttempt.provider_delivery_id) .join( CampaignJob, PostboxDeliveryAttempt.job_id == CampaignJob.id, ) .filter( CampaignJob.tenant_id == tenant_id, CampaignJob.campaign_id == campaign_id, CampaignJob.campaign_version_id == version.id, PostboxDeliveryAttempt.tenant_id == tenant_id, PostboxDeliveryAttempt.provider_delivery_id.is_not(None), ) .all() ) delivery_ids = tuple( dict.fromkeys( str(row[0]) for row in rows if row[0] ) ) if not delivery_ids: return {"status": "not_applicable", "delivery_count": 0} integration = postbox_integration() if not integration.receipt_evidence_available: return { "status": "unavailable", "delivery_count": len(delivery_ids), "missing_delivery_count": len(delivery_ids), } summaries = integration.delivery_receipt_summaries( session, tenant_id=tenant_id, delivery_ids=delivery_ids, ) values = tuple(summaries.values()) return { "status": "available", "delivery_count": len(delivery_ids), "reported_delivery_count": len(values), "missing_delivery_count": len(delivery_ids) - len(values), "currently_readable_delivery_count": sum( summary.currently_readable for summary in values ), "read_delivery_count": sum( summary.read_receipt_count > 0 for summary in values ), "acknowledged_delivery_count": sum( summary.acknowledged_receipt_count > 0 for summary in values ), "read_receipt_count": sum( summary.read_receipt_count for summary in values ), "acknowledged_receipt_count": sum( summary.acknowledged_receipt_count for summary in values ), "routed_message_count": sum( summary.routed_message_count for summary in values ), "withdrawn_message_count": sum( summary.withdrawn_message_count for summary in values ), "expired_message_count": sum( summary.expired_message_count for summary in values ), } def _campaign_report_campaign_payload(campaign: Campaign) -> dict[str, Any]: return { "id": campaign.id, "external_id": campaign.external_id, "name": campaign.name, "description": campaign.description, "status": campaign.status, "created_at": campaign.created_at.isoformat() if campaign.created_at else None, "updated_at": campaign.updated_at.isoformat() if campaign.updated_at else None, } def _campaign_report_attempt_counts( session: Session, *, tenant_id: str, campaign_id: str, version: CampaignVersion | None, ) -> dict[str, int]: def count(model: type[Any]) -> int: query = ( session.query(model) .join(CampaignJob, model.job_id == CampaignJob.id) .filter( CampaignJob.tenant_id == tenant_id, CampaignJob.campaign_id == campaign_id, ) ) if version is not None: query = query.filter( CampaignJob.campaign_version_id == version.id, ) else: query = query.filter(False) return int(query.count()) return { "send_attempts": count(SendAttempt), "imap_append_attempts": count(ImapAppendAttempt), "postbox_delivery_attempts": count(PostboxDeliveryAttempt), "message_actions": count(CampaignMessageAction), "message_action_attempts": int( session.query(CampaignMessageActionAttempt) .join( CampaignMessageAction, CampaignMessageAction.id == CampaignMessageActionAttempt.action_id, ) .filter( CampaignMessageAction.tenant_id == tenant_id, CampaignMessageAction.campaign_id == campaign_id, *( [CampaignMessageAction.campaign_version_id == version.id] if version is not None else [False] ), ) .count() ), } def _campaign_message_action_summary( session: Session, *, tenant_id: str, campaign_id: str, version: CampaignVersion | None, ) -> dict[str, Any]: base = [ CampaignMessageAction.tenant_id == tenant_id, CampaignMessageAction.campaign_id == campaign_id, ] if version is not None: base.append(CampaignMessageAction.campaign_version_id == version.id) else: base.append(False) by_kind = { str(kind): int(count) for kind, count in ( session.query( CampaignMessageAction.kind, func.count(CampaignMessageAction.id), ) .filter(*base) .group_by(CampaignMessageAction.kind) .all() ) } by_status = { str(action_status): int(count) for action_status, count in ( session.query( CampaignMessageAction.status, func.count(CampaignMessageAction.id), ) .filter(*base) .group_by(CampaignMessageAction.status) .all() ) } recent = ( session.query(CampaignMessageAction) .filter(*base) .order_by(CampaignMessageAction.created_at.desc()) .limit(100) .all() ) return { "total": sum(by_kind.values()), "by_kind": by_kind, "by_status": by_status, "recent": [ { "id": action.id, "job_id": action.job_id, "kind": action.kind, "status": action.status, "reason": action.reason, "message_sha256": action.message_sha256, "recipient_manifest_sha256": action.recipient_manifest_sha256, "recipient_count": action.recipient_count, "prior_send_status": action.prior_send_status, "final_send_status": action.final_send_status, "accepted_count": action.accepted_count, "refused_count": action.refused_count, "linked_send_attempt_id": action.linked_send_attempt_id, "actor_user_id": action.actor_user_id, "created_at": _iso_timestamp(action.created_at), "completed_at": _iso_timestamp(action.completed_at), } for action in recent ], } def _persisted_campaign_issue_count( session: Session, *, tenant_id: str, campaign_id: str, version: CampaignVersion | None, ) -> int: issue_query = session.query(CampaignIssue).filter( CampaignIssue.tenant_id == tenant_id, CampaignIssue.campaign_id == campaign_id, ) if version: issue_query = issue_query.filter(CampaignIssue.campaign_version_id == version.id) else: issue_query = issue_query.filter(False) return int(issue_query.count()) def _campaign_report_status_counts(version: CampaignVersion | None, jobs: list[CampaignJob]) -> dict[str, dict[str, int]]: aggregate = _aggregate_job_list(version, jobs) return _campaign_report_status_counts_from_aggregate(version, aggregate) def _campaign_report_status_counts_from_aggregate( version: CampaignVersion | None, aggregate: _JobReportAggregate, ) -> dict[str, dict[str, int]]: validation_counts = dict(aggregate.validation) inactive_entries = _inactive_entry_count(version) if inactive_entries: validation_counts["inactive"] = inactive_entries return { "build": dict(aggregate.build), "validation": validation_counts, "queue": dict(aggregate.queue), "send": dict(aggregate.send), "postbox": dict(aggregate.postbox), "print": dict(aggregate.print), "imap": dict(aggregate.imap), } def _campaign_report_cards_from_aggregate( version: CampaignVersion | None, aggregate: _JobReportAggregate, ) -> dict[str, object]: send_counts = aggregate.send imap_counts = aggregate.imap sent = ( send_counts.get("sent", 0) + send_counts.get("smtp_accepted", 0) + send_counts.get("postbox_accepted", 0) + send_counts.get("print_accepted", 0) + send_counts.get("delivered", 0) + send_counts.get("partially_accepted", 0) ) failed = ( send_counts.get("failed_temporary", 0) + send_counts.get("failed_permanent", 0) ) outcome_unknown = send_counts.get("outcome_unknown", 0) not_attempted = send_counts.get("not_queued", 0) cancelled = send_counts.get("cancelled", 0) inactive_entries = _inactive_entry_count(version) return { "jobs_total": aggregate.total, "inactive": inactive_entries, "queueable": aggregate.queueable, "queueable_unattempted": aggregate.queueable_unattempted, "retryable": aggregate.retryable, "cancellable": aggregate.cancellable, "needs_attention": aggregate.needs_attention, "sent": sent, "smtp_accepted": ( send_counts.get("sent", 0) + send_counts.get("smtp_accepted", 0) ), "postbox_accepted": send_counts.get("postbox_accepted", 0), "print_accepted": send_counts.get("print_accepted", 0), "delivered": send_counts.get("delivered", 0), "partially_accepted": send_counts.get("partially_accepted", 0), "failed": failed, "outcome_unknown": outcome_unknown, "not_attempted": not_attempted, "skipped": send_counts.get("skipped", 0), "queued_or_active": aggregate.pending, "cancelled": cancelled, "partially_completed": bool( send_counts.get("partially_accepted", 0) or sent and (failed or outcome_unknown or not_attempted or cancelled) ), "imap_appended": imap_counts.get("appended", 0), "imap_failed": imap_counts.get("failed", 0), "imap_skipped": imap_counts.get("skipped", 0), } def _campaign_report_cards(version: CampaignVersion | None, jobs: list[CampaignJob]) -> dict[str, object]: return _campaign_report_cards_from_aggregate( version, _aggregate_job_list(version, jobs), ) def _aggregate_job_list( version: CampaignVersion | None, jobs: list[CampaignJob], ) -> _JobReportAggregate: aggregate = _JobReportAggregate() retry_max_attempts = _retry_max_attempts(version) for job in jobs: aggregate.add( job, retry_max_attempts=retry_max_attempts, include_recent_failures=False, ) return aggregate def _retry_max_attempts(version: CampaignVersion | None) -> int | None: if version is None or not isinstance( getattr(version, "execution_snapshot", None), dict, ): return None try: return ExecutionSnapshot.model_validate(version.execution_snapshot).delivery.retry.max_attempts except Exception: return None def _inactive_entry_count(version: CampaignVersion | None) -> int: build_summary = version.build_summary if version and isinstance(version.build_summary, dict) else {} return int(build_summary.get("inactive_count") or build_summary.get("inactive_entries_count") or 0) def generate_jobs_csv( session: Session, *, tenant_id: str, campaign_id: str, version_id: str | None = None, include_diagnostics: bool = False, ) -> str: campaign = _get_campaign(session, tenant_id=tenant_id, campaign_id=campaign_id) version = _selected_version(session, campaign, version_id) jobs_query = session.query(CampaignJob).filter( CampaignJob.tenant_id == tenant_id, CampaignJob.campaign_id == campaign.id, ) if version: jobs_query = jobs_query.filter(CampaignJob.campaign_version_id == version.id) else: jobs_query = jobs_query.filter(False) jobs = ( jobs_query .order_by(CampaignJob.entry_index.asc()) .limit(CAMPAIGN_CSV_JOB_LIMIT + 1) .all() ) if len(jobs) > CAMPAIGN_CSV_JOB_LIMIT: raise CampaignReportError( "The campaign CSV export exceeds the safe row limit of " f"{CAMPAIGN_CSV_JOB_LIMIT}. Split the export by recipient filter." ) job_ids = [job.id for job in jobs] latest_smtp = _latest_attempts_for_jobs( session, SendAttempt, job_ids, ) latest_imap = _latest_attempts_for_jobs( session, ImapAppendAttempt, job_ids, ) latest_message_actions = _latest_message_actions_for_jobs( session, job_ids, ) review_decisions = _review_decisions_by_job(version) calendar_invitations = _calendar_invitations_for_jobs(session, jobs) rows = [ _job_evidence_row( job, latest_smtp=latest_smtp.get(job.id), latest_imap=latest_imap.get(job.id), latest_message_action=latest_message_actions.get(job.id), include_diagnostics=include_diagnostics, review_decision=review_decisions.get(job.id), calendar_invitation=calendar_invitations.get(job.id), ) for job in jobs ] fieldnames = [ "job_id", "campaign_id", "campaign_version_id", "entry_index", "entry_id", "recipient_email", "from", "to", "cc", "bcc", "reply_to", "subject", "message_id_header", "build_status", "validation_status", "queue_status", "send_status", "delivery_channel_policy", "postbox_status", "print_status", "imap_status", "attempt_count", "postbox_attempt_count", "print_attempt_count", "postbox_target_count", "postbox_targets", "distribution_list_id", "distribution_list_revision", "distribution_snapshot_id", "distribution_expansion_hash", "distribution_recipient_key", "selected_route_channel", "selected_route_target_key", "route_reason", "print_render_id", "print_template_id", "print_template_revision_id", "print_template_hash", "print_input_hash", "print_output_sha256", "print_artifact_kind", "print_artifact_file_id", "print_artifact_download_path", "print_actor_account_id", "queued_at", "outcome_unknown_at", "sent_at", "last_error", "eml_size_bytes", "eml_sha256", "issues_count", "attachment_config_count", "matched_file_count", "attachment_names", "latest_smtp_attempt_number", "latest_smtp_status", "latest_smtp_status_code", "latest_smtp_started_at", "latest_smtp_finished_at", "latest_imap_attempt_number", "latest_imap_status", "latest_imap_folder", "latest_imap_created_at", "latest_imap_updated_at", "latest_message_action_kind", "latest_message_action_status", "latest_message_action_id", "latest_message_action_at", "review_decision", "review_reason", "review_actor_user_id", "review_decided_at", "review_issue_codes", "calendar_invitation_state", "calendar_rsvp_status", "calendar_event_id", "calendar_id", "calendar_uid", "calendar_external_state", "calendar_reply_ingress", ] if include_diagnostics: sent_at_index = fieldnames.index("outcome_unknown_at") fieldnames[sent_at_index:sent_at_index] = ["claimed_at", "smtp_started_at"] fieldnames.extend( [ "review_key", "review_message_sha256", "review_issue_fingerprint", ] ) buffer = io.StringIO() writer = csv.DictWriter(buffer, fieldnames=fieldnames) writer.writeheader() writer.writerows(rows) return buffer.getvalue() def _latest_attempts_for_jobs( session: Session, model: type[Any], job_ids: list[str], *, batch_size: int = 500, ) -> dict[str, Any]: latest: dict[str, Any] = {} for offset in range(0, len(job_ids), batch_size): batch = job_ids[offset : offset + batch_size] attempts = ( session.query(model) .filter(model.job_id.in_(batch)) .order_by(model.job_id.asc(), model.attempt_number.asc()) .yield_per(batch_size) ) for attempt in attempts: current = latest.get(attempt.job_id) if ( current is None or (attempt.attempt_number or 0) >= (current.attempt_number or 0) ): latest[attempt.job_id] = attempt return latest def _latest_message_actions_for_jobs( session: Session, job_ids: list[str], *, batch_size: int = 500, ) -> dict[str, CampaignMessageAction]: latest: dict[str, CampaignMessageAction] = {} for offset in range(0, len(job_ids), batch_size): batch = job_ids[offset : offset + batch_size] actions = ( session.query(CampaignMessageAction) .filter(CampaignMessageAction.job_id.in_(batch)) .order_by( CampaignMessageAction.job_id.asc(), CampaignMessageAction.created_at.asc(), ) .yield_per(batch_size) ) for action in actions: latest[action.job_id] = action return latest