"""Campaigns: 12-step wizard (draft CRUD), preview with real rendered messages, start/pause/resume/cancel, SSE live progress stream, per-recipient monitor, CSV export.""" import asyncio import csv import io import json import re from typing import Optional from fastapi import APIRouter, Depends, HTTPException, Query, Request from fastapi.responses import StreamingResponse from pydantic import BaseModel, Field from sqlalchemy import delete as sa_delete, func, select from sqlalchemy.ext.asyncio import AsyncSession from app.api.deps import get_current_user, get_db, require_csrf from app.core.config import settings as cfg from app.core.security import log_audit from app.models import (Attachment, Campaign, CampaignRecipient, CompanyPhoneNumber, EmailTemplate, GmailAccount, Recipient, RecipientList, SendJob, SubjectLine, SuppressionEntry, User) from app.services.campaign_engine import get_setting, materialize_campaign from app.services.campaign_service import (cancel_campaign, pause_campaign, resume_campaign, start_campaign) from app.services.message_builder import sanitize_preview from app.services.variable_engine import GeneratedValues, VariableEngine router = APIRouter(prefix="/campaigns", tags=["campaigns"]) MSG_TYPES = {"html", "plain", "both"} class CampaignIn(BaseModel): name: str = Field(min_length=1, max_length=250) recipient_list_id: Optional[int] = None subject_set_ids: list = [] template_ids: list = [] message_type: str = "html" # html|plain|both attach_mode: str = "none" # none|pdf|file|image attach_pdf: bool = False attach_pdf_template_id: Optional[int] = None attachment_id: Optional[int] = None variable_config: dict = {} account_ids: list = [] rotation_strategy: str = "sequential" rotation_weights: dict = {} subject_rotation: str = "round_robin" template_sequence_mode: str = "repeat" phone_selection: str = "round_robin" max_per_account: int = Field(default=200, ge=1, le=2000) send_interval_seconds: float = Field(default=1.0, ge=0.5, le=3600) rate_per_minute: int = Field(default=60, ge=1, le=600) max_retries: int = Field(default=3, ge=0, le=10) retry_delay_seconds: int = Field(default=60, ge=5, le=3600) unsubscribe_enabled: bool = True preview_sample_size: int = Field(default=5, ge=1, le=20) VALID_ENUMS = { "message_type": MSG_TYPES, "attach_mode": {"none", "pdf", "file", "image"}, "rotation_strategy": {"sequential", "weighted"}, "subject_rotation": {"round_robin", "random", "fixed_first"}, "template_sequence_mode": {"repeat", "stop", "random"}, "phone_selection": {"round_robin", "random", "fixed"}, } def _ser(c: Campaign) -> dict: return { "id": c.id, "name": c.name, "status": c.status, "recipient_list_id": c.recipient_list_id, "subject_set_ids": c.subject_set_ids or [], "template_ids": c.template_ids or [], "message_type": c.message_type, "attach_mode": c.attach_mode, "attach_pdf": bool(c.attach_pdf), "attach_pdf_template_id": c.attach_pdf_template_id, "attachment_id": c.attachment_id, "variable_config": c.variable_config or {}, "account_ids": c.account_ids or [], "rotation_strategy": c.rotation_strategy, "rotation_weights": c.rotation_weights or {}, "subject_rotation": c.subject_rotation, "template_sequence_mode": c.template_sequence_mode, "phone_selection": c.phone_selection, "max_per_account": c.max_per_account, "send_interval_seconds": c.send_interval_seconds, "rate_per_minute": c.rate_per_minute, "max_retries": c.max_retries, "retry_delay_seconds": c.retry_delay_seconds, "unsubscribe_enabled": bool(c.unsubscribe_enabled), "preview_sample_size": c.preview_sample_size, "total_recipients": c.total_recipients, "sent_count": c.sent_count, "failed_count": c.failed_count, "suppressed_count": c.suppressed_count, "skipped_count": c.skipped_count, "queued_count": c.queued_count, "started_at": c.started_at.isoformat() if c.started_at else None, "finished_at": c.finished_at.isoformat() if c.finished_at else None, "created_at": c.created_at.isoformat() if c.created_at else None, } async def _get_own(db: AsyncSession, user: User, cid: int) -> Campaign: c = await db.get(Campaign, cid) if not c or c.user_id != user.id: raise HTTPException(status_code=404, detail="Campaign not found") return c async def _validate_refs(db: AsyncSession, user: User, data: CampaignIn): """Wizard-level validation: all referenced entities must exist & belong to the user.""" errors = [] if data.message_type not in MSG_TYPES: errors.append("message_type must be html|plain|both") if data.attach_mode not in VALID_ENUMS["attach_mode"]: errors.append("attach_mode must be none|pdf|file|image") if data.rotation_strategy not in VALID_ENUMS["rotation_strategy"]: errors.append("rotation_strategy must be sequential|weighted") if data.subject_rotation not in VALID_ENUMS["subject_rotation"]: errors.append("subject_rotation must be round_robin|random|fixed_first") if data.template_sequence_mode not in VALID_ENUMS["template_sequence_mode"]: =====WIZOPTS===== @router.get("/wizard/options") async def wizard_options(db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)): """All selectable entities for the 12-step wizard in one call.""" lists = (await db.execute(select(RecipientList).where(RecipientList.user_id == user.id) .order_by(RecipientList.id.desc()))).scalars().all() subjects = (await db.execute(select(SubjectLine).where(SubjectLine.user_id == user.id) .order_by(SubjectLine.id.desc()))).scalars().all() templates = (await db.execute(select(EmailTemplate).where(EmailTemplate.user_id == user.id) .order_by(EmailTemplate.id.desc()))).scalars().all() accounts = (await db.execute(select(GmailAccount).where(GmailAccount.user_id == user.id, GmailAccount.status == "authorized", GmailAccount.is_active == True))).scalars().all() # noqa: E712 phones = (await db.execute(select(CompanyPhoneNumber).where(CompanyPhoneNumber.user_id == user.id, CompanyPhoneNumber.is_active == True))).scalars().all() # noqa: E712 attachments = (await db.execute(select(Attachment).where(Attachment.user_id == user.id) .order_by(Attachment.id.desc()))).scalars().all() async def valid_count(list_id: int) -> int: return (await db.execute(select(func.count()).select_from(Recipient) .where(Recipient.list_id == list_id, Recipient.is_valid == True))).scalar() or 0 # noqa: E712 list_out = [] for rl in lists: list_out.append({"id": rl.id, "name": rl.name, "row_count": rl.row_count, "valid_count": await valid_count(rl.id)}) return { "lists": list_out, "subjects": [{"id": s.id, "subject": s.subject[:160]} for s in subjects], "templates": [{"id": t.id, "name": t.name} for t in templates], "accounts": [{"id": a.id, "email": a.email, "sends_today": a.sends_today, "daily_limit": a.daily_limit, "sending_method": a.sending_method} for a in accounts], "phones": [{"id": p.id, "phone": p.phone, "label": p.label} for p in phones], "attachments": [{"id": a.id, "name": a.name, "kind": a.kind, "filename": a.filename} for a in attachments], "limits": {"max_recipients": cfg.MAX_RECIPIENTS_PER_CAMPAIGN, "max_accounts": cfg.MAX_GMAIL_ACCOUNTS_PER_CAMPAIGN, "max_subjects": cfg.MAX_SUBJECTS}, } @router.post("/validate-step") async def validate_step(payload: dict, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)): """Wizard step validation between steps (step 3, 6, 9, 11 checks).""" step = int(payload.get("step", 0)) data = payload.get("data", {}) errors = [] if step == 1: if not (data.get("name") or "").strip(): errors.append("Campaign name is required") if step == 2: if not data.get("recipient_list_id"): errors.append("Select a recipient list") if step == 3: ids = data.get("subject_set_ids") or [] if not ids: errors.append("Select at least one subject line") if len(ids) > cfg.MAX_SUBJECTS: errors.append(f"Max {cfg.MAX_SUBJECTS} subject variations") strategy = data.get("subject_rotation") if strategy not in VALID_ENUMS["subject_rotation"]: errors.append("Invalid subject rotation strategy")