from __future__ import annotations import csv import io import math from collections import Counter from datetime import datetime, timezone from typing import Any from sqlalchemy.orm import Session from govoplan_core.settings import settings as core_settings from govoplan_campaign.backend.db.models import ( Campaign, CampaignIssue, CampaignJob, CampaignVersion, ImapAppendAttempt, SendAttempt, ) from govoplan_campaign.backend.sending.execution import ExecutionSnapshot class CampaignReportError(RuntimeError): pass 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) -> 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": version.source_filename, "created_at": version.created_at.isoformat() if version.created_at else None, "validation_summary": version.validation_summary, "build_summary": version.build_summary, "execution_snapshot_hash": version.execution_snapshot_hash, "execution_snapshot_at": version.execution_snapshot_at.isoformat() if version.execution_snapshot_at else None, } def _load_delivery_info(version: CampaignVersion | None, jobs: list[CampaignJob]) -> 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, "build_token": None, "built_at": None, "job_manifest_sha256": None, "job_count": 0, "queueable_job_count": 0, "effective_policy_sha256": None, "smtp_config_fingerprint": None, "imap_config_fingerprint": None, "estimated_remaining_send_seconds": None, "estimated_remaining_send_human": 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 as exc: # pragma: no cover - reporting should remain available default["load_error"] = str(exc) return default messages_per_minute = snapshot.delivery.rate_limit.messages_per_minute pending = [job for job in jobs if job.send_status in {"queued", "claimed", "sending"}] estimated_seconds = None if messages_per_minute and pending: estimated_seconds = int(math.ceil((len(pending) / messages_per_minute) * 60)) return { "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, "build_token": snapshot.build_token, "built_at": snapshot.built_at, "job_manifest_sha256": snapshot.job_manifest_sha256, "job_count": snapshot.job_count, "queueable_job_count": snapshot.queueable_job_count, "effective_policy_sha256": snapshot.effective_policy_sha256, "smtp_config_fingerprint": snapshot.smtp_config_fingerprint, "imap_config_fingerprint": snapshot.imap_config_fingerprint, "estimated_remaining_send_seconds": estimated_seconds, "estimated_remaining_send_human": _human_duration(estimated_seconds), "background_workers_enabled": bool(core_settings.celery_enabled), } 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), } def _recent_failures(jobs: list[CampaignJob], *, limit: int = 20) -> list[dict[str, Any]]: failed = [job for job in jobs if job.last_error or str(job.send_status).startswith("failed") or job.imap_status == "failed"] 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, "imap_status": job.imap_status, "attempt_count": job.attempt_count, "last_error": job.last_error, "updated_at": job.updated_at.isoformat() if job.updated_at else None, } for job in failed[:limit] ] def _job_row(job: CampaignJob) -> dict[str, Any]: return { "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, "imap_status": job.imap_status, "attempt_count": job.attempt_count, "queued_at": job.queued_at.isoformat() if job.queued_at else None, "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, "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": job.last_error, "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)), } 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 = True, ) -> 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_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()) .all() ) job_ids = [job.id for job in jobs] send_attempts = session.query(SendAttempt).filter(SendAttempt.job_id.in_(job_ids)).count() if job_ids else 0 imap_attempts = session.query(ImapAppendAttempt).filter(ImapAppendAttempt.job_id.in_(job_ids)).count() if job_ids else 0 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) persisted_issues = issue_query.count() validation_counts = _counter([job.validation_status for job in jobs]) queue_counts = _counter([job.queue_status for job in jobs]) send_counts = _counter([job.send_status for job in jobs]) imap_counts = _counter([job.imap_status for job in jobs]) build_counts = _counter([job.build_status for job in jobs]) queueable = sum(1 for job in jobs if job.validation_status in {"ready", "warning"} and job.build_status == "built") needs_attention = sum( 1 for job in jobs if job.validation_status in {"needs_review", "blocked"} or job.send_status in {"failed_temporary", "failed_permanent", "outcome_unknown", "claimed", "sending"} or job.imap_status == "failed" ) sent = send_counts.get("sent", 0) + send_counts.get("smtp_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) queued = send_counts.get("queued", 0) + send_counts.get("claimed", 0) + send_counts.get("sending", 0) cancelled = send_counts.get("cancelled", 0) build_summary = version.build_summary if version and isinstance(version.build_summary, dict) else {} inactive_entries = int(build_summary.get("inactive_count") or build_summary.get("inactive_entries_count") or 0) if inactive_entries: validation_counts["inactive"] = inactive_entries report: dict[str, Any] = { "generated_at": _utcnow_iso(), "campaign": { "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, }, "current_version": _version_info(version), "selected_version_id": version.id if version else None, "cards": { "jobs_total": len(jobs), "inactive": inactive_entries, "queueable": queueable, "needs_attention": needs_attention, "sent": sent, "smtp_accepted": sent, "failed": failed, "outcome_unknown": outcome_unknown, "not_attempted": not_attempted, "queued_or_active": queued, "cancelled": cancelled, "partially_completed": bool(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), }, "status_counts": { "build": build_counts, "validation": validation_counts, "queue": queue_counts, "send": send_counts, "imap": imap_counts, }, "issues": { **_issue_summary_from_jobs(jobs), "persisted_campaign_issue_count": persisted_issues, }, "attachments": _attachment_summary(jobs), "attempts": { "send_attempts": int(send_attempts), "imap_append_attempts": int(imap_attempts), }, "delivery": _load_delivery_info(version, jobs), } if include_recent_failures: report["recent_failures"] = _recent_failures(jobs) if include_jobs: report["jobs"] = [_job_row(job) for job in jobs] return report def generate_jobs_csv( session: Session, *, tenant_id: str, campaign_id: str, version_id: str | None = None, ) -> 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()) .all() ) rows = [_job_row(job) for job in jobs] fieldnames = [ "job_id", "entry_index", "entry_id", "recipient_email", "subject", "build_status", "validation_status", "queue_status", "send_status", "imap_status", "attempt_count", "queued_at", "claimed_at", "smtp_started_at", "outcome_unknown_at", "sent_at", "last_error", "eml_size_bytes", "eml_sha256", "issues_count", "attachment_config_count", "matched_file_count", ] buffer = io.StringIO() writer = csv.DictWriter(buffer, fieldnames=fieldnames) writer.writeheader() writer.writerows(rows) return buffer.getvalue()