"""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, SendAttempt, 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", "body_pdf", "body_image", "body_word"}, "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|body_pdf|body_image|body_word") 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"]: errors.append("template_sequence_mode must be repeat|stop|random") if data.phone_selection not in VALID_ENUMS["phone_selection"]: errors.append("phone_selection must be round_robin|random|fixed") if data.recipient_list_id: rl = await db.get(RecipientList, data.recipient_list_id) if not rl or rl.user_id != user.id: errors.append("Recipient list not found") else: rc = (await db.execute(select(func.count()).select_from(Recipient) .where(Recipient.list_id == rl.id, Recipient.is_valid == True))).scalar() or 0 # noqa: E712 if rc == 0: errors.append("Recipient list is empty") if rc > cfg.MAX_RECIPIENTS_PER_CAMPAIGN: errors.append(f"Recipient list has {rc} entries; max {cfg.MAX_RECIPIENTS_PER_CAMPAIGN}") for sid in data.subject_set_ids or []: s = await db.get(SubjectLine, sid) if not s or s.user_id != user.id: errors.append(f"Subject line {sid} not found") if not (data.subject_set_ids or []): errors.append("Select at least one subject line") for tid in data.template_ids or []: t = await db.get(EmailTemplate, tid) if not t or t.user_id != user.id: errors.append(f"Template {tid} not found") if not (data.template_ids or []): errors.append("Select at least one HTML template") if not data.account_ids: errors.append("Select at least one Gmail account") elif len(data.account_ids) > cfg.MAX_GMAIL_ACCOUNTS_PER_CAMPAIGN: errors.append(f"Max {cfg.MAX_GMAIL_ACCOUNTS_PER_CAMPAIGN} accounts per campaign") else: for aid in data.account_ids: acc = await db.get(GmailAccount, aid) if not acc or acc.user_id != user.id: errors.append(f"Gmail account {aid} not found") elif acc.status != "authorized" or not acc.is_active: errors.append(f"Account {acc.email} is not authorized/active") if data.attach_mode == "pdf": if not (data.attach_pdf or data.attach_pdf_template_id or data.attachment_id): errors.append("PDF attachment mode requires a template or attachment to convert") if data.attach_mode in ("file", "image") and not data.attachment_id: errors.append("File/image attachment mode requires selecting an attachment") if len(data.rotation_weights or {}) > 0 and data.rotation_strategy != "weighted": errors.append("rotation_weights only applies when rotation_strategy is weighted") # variable config sanity vc = data.variable_config or {} try: if "amount_min" in vc and "amount_max" in vc and float(vc["amount_min"]) > float(vc["amount_max"]): errors.append("amount_min must be <= amount_max") if "shipping_min" in vc and "shipping_max" in vc and float(vc["shipping_min"]) > float(vc["shipping_max"]): errors.append("shipping_min must be <= shipping_max") except (TypeError, ValueError): errors.append("amount/shipping ranges must be numeric") return errors @router.get("/list") async def list_campaigns(status: str = Query(""), search: str = Query(""), page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)): q = select(Campaign).where(Campaign.user_id == user.id) if status: q = q.where(Campaign.status == status) if search: q = q.where(Campaign.name.ilike(f"%{search}%")) total = (await db.execute(select(func.count()).select_from(q.subquery()))).scalar() or 0 rows = (await db.execute(q.order_by(Campaign.id.desc()) .offset((page - 1) * page_size).limit(page_size))).scalars().all() return {"campaigns": [_ser(c) for c in rows], "total": total, "page": page, "page_size": page_size} @router.post("/create") async def create_campaign(payload: CampaignIn, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), _: None = Depends(require_csrf)): """Create a new campaign (wizard step 12 / Save). Validates all references.""" errors = await _validate_refs(db, user, payload) if errors: raise HTTPException(status_code=400, detail={"message": "Validation failed", "errors": errors}) c = Campaign(user_id=user.id, **payload.model_dump()) db.add(c) await db.commit() await log_audit(db, user.id, "campaign.create", "campaign", c.id, {"name": c.name}) return _ser(c) @router.get("/{cid}") async def get_campaign(cid: int, user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): return _ser(await _get_own(db, user, cid)) @router.post("/{cid}/patch") async def patch_campaign(cid: int, payload: CampaignIn, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), _: None = Depends(require_csrf)): """Full update (wizard edit). Only allowed while draft/paused/queued.""" c = await _get_own(db, user, cid) if c.status not in ("draft", "paused", "queued", "failed"): raise HTTPException(status_code=409, detail=f"Cannot edit a {c.status} campaign") errors = await _validate_refs(db, user, payload) if errors: raise HTTPException(status_code=400, detail={"message": "Validation failed", "errors": errors}) data = payload.model_dump() for k, v in data.items(): setattr(c, k, v) await db.commit() await log_audit(db, user.id, "campaign.patch", "campaign", c.id, {"fields": list(data.keys())}) return _ser(c) @router.post("/{cid}/delete") async def delete_campaign(cid: int, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), _: None = Depends(require_csrf)): c = await _get_own(db, user, cid) if c.status == "running": raise HTTPException(status_code=409, detail="Pause or cancel the campaign first") await db.delete(c) await db.commit() await log_audit(db, user.id, "campaign.delete", "campaign", cid, {"name": c.name}) return {"deleted": cid} @router.post("/clear-all") async def clear_all_campaigns(db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), _: None = Depends(require_csrf)): """Delete ALL campaigns (not running ones). Requires explicit UI confirmation.""" rows = (await db.execute(select(Campaign).where(Campaign.user_id == user.id))).scalars().all() deleted, skipped = 0, 0 for c in rows: if c.status == "running": skipped += 1 continue await db.delete(c) deleted += 1 await db.commit() await log_audit(db, user.id, "campaign.clear_all", "campaign", None, {"deleted": deleted, "skipped": skipped}) return {"deleted": deleted, "skipped": skipped} # --------------------------------------------------------------------------- # Preview: materialize the campaign plan and render REAL messages (no sending) # --------------------------------------------------------------------------- @router.post("/{cid}/preview") async def preview_campaign(cid: int, sample_size: int = Query(0), db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), _: None = Depends(require_csrf)): """Preview: generate the campaign plan for a sample of recipients and render the actual messages (subject + HTML + substituted variables). NOTHING is sent.""" c = await _get_own(db, user, cid) if c.status not in ("draft", "paused", "queued", "failed"): raise HTTPException(status_code=409, detail="Cannot preview a running campaign") size = sample_size or c.preview_sample_size or 5 size = max(1, min(size, 20)) if not c.recipient_list_id: raise HTTPException(status_code=400, detail="Campaign has no recipient list") rl = await db.get(RecipientList, c.recipient_list_id) if not rl or rl.user_id != user.id: raise HTTPException(status_code=400, detail="Recipient list not found") recipient_rows = (await db.execute( select(Recipient).where(Recipient.list_id == rl.id, Recipient.is_valid == True) # noqa: E712 .order_by(Recipient.id).limit(size))).scalars().all() if not recipient_rows: raise HTTPException(status_code=400, detail="Recipient list has no valid recipients") # materialize plan rows WITHOUT persisting CampaignRecipient entities result = await materialize_campaign(db, c, recipient_rows, preview=True, preview_size=size) plan = result.get("plan", []) # render actual message content per sample engine = VariableEngine(c.variable_config or {}) out = [] for item in plan: rec = next((r for r in recipient_rows if r.id == item["recipient_id"]), None) if not rec: continue tpl = await db.get(EmailTemplate, item["template_id"]) if item.get("template_id") else None subj = await db.get(SubjectLine, item["subject_id"]) if item.get("subject_id") else None g = GeneratedValues() g.invoice_number = item.get("invoice_number", "") g.amount = item.get("amount") g.shipping = item.get("shipping") g.invoice_date = item.get("invoice_date") g.due_date = item.get("due_date") phone = item.get("phone", "") unsub = f"{cfg.UNSUBSCRIBE_BASE or cfg.BASE_URL}/unsubscribe/{rec.unsub_token}" if c.unsubscribe_enabled else "" extra = {"unsubscribe_link": unsub, "company_phone": phone} body_html = "" subject = "" warn_list = list(item.get("warnings") or []) if isinstance(item.get("warnings"), list) else ([item["warnings"]] if item.get("warnings") else []) if tpl: res = engine.substitute(tpl.body_html, rec, g, custom_fields=rec.custom_fields or {}, extra=extra) body_html = res.text subject = subj.subject if subj else (tpl.subject_hint or tpl.name) for u in res.unknown: warn_list.append(f"Unknown variable in body: {u}") for m_ in res.missing: warn_list.append(f"Missing data for: {m_}") if subj: res_s = engine.substitute(subj.subject, rec, g, custom_fields=rec.custom_fields or {}, extra=extra) subject = res_s.text for u in res_s.unknown: warn_list.append(f"Unknown variable in subject: {u}") for m_ in res_s.missing: warn_list.append(f"Missing data in subject: {m_}") out.append({ "recipient": {"id": rec.id, "email": rec.email, "first_name": rec.first_name, "last_name": rec.last_name}, "from_account": item.get("account", ""), "subject": subject, "body_html": sanitize_preview(body_html), "raw_variables": {"invoice_number": g.invoice_number, "amount": g.amount, "shipping": g.shipping, "phone": phone}, "warnings": warn_list, }) return {"campaign_id": c.id, "sample_size": len(out), "messages": out, "plan_summary": result.get("summary", {})} # --------------------------------------------------------------------------- # Control: start / pause / resume / cancel # --------------------------------------------------------------------------- @router.post("/{cid}/start") async def start(cid: int, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), _: None = Depends(require_csrf)): c = await _get_own(db, user, cid) if c.status == "draft": # materialize campaign_recipients on first start if not c.recipient_list_id: raise HTTPException(status_code=400, detail="Campaign has no recipient list") rl = await db.get(RecipientList, c.recipient_list_id) if not rl: raise HTTPException(status_code=400, detail="Recipient list not found") recipient_rows = (await db.execute( select(Recipient).where(Recipient.list_id == c.recipient_list_id, Recipient.is_valid == True) # noqa: E712 .order_by(Recipient.id))).scalars().all() if not recipient_rows: raise HTTPException(status_code=400, detail="No valid recipients in list") if len(recipient_rows) > cfg.MAX_RECIPIENTS_PER_CAMPAIGN: raise HTTPException(status_code=400, detail=f"Too many recipients ({len(recipient_rows)}); max {cfg.MAX_RECIPIENTS_PER_CAMPAIGN}") result = await materialize_campaign(db, c, recipient_rows) c.total_recipients = result.get("total", 0) await db.commit() res = await start_campaign(db, c) if not res.get("ok"): raise HTTPException(status_code=409, detail=res.get("reason", "Cannot start")) await log_audit(db, user.id, "campaign.start", "campaign", c.id, {"queued": res.get("queued")}) return {"status": "running", "queued": res.get("queued")} @router.post("/{cid}/pause") async def pause(cid: int, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), _: None = Depends(require_csrf)): c = await _get_own(db, user, cid) if c.status != "running": raise HTTPException(status_code=409, detail=f"Cannot pause a {c.status} campaign") res = await pause_campaign(db, c) await log_audit(db, user.id, "campaign.pause", "campaign", c.id, {}) return {"status": "paused"} @router.post("/{cid}/resume") async def resume(cid: int, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), _: None = Depends(require_csrf)): c = await _get_own(db, user, cid) if c.status != "paused": raise HTTPException(status_code=409, detail=f"Cannot resume a {c.status} campaign") res = await resume_campaign(db, c) if not res.get("ok"): raise HTTPException(status_code=409, detail=res.get("reason", "Cannot resume")) await log_audit(db, user.id, "campaign.resume", "campaign", c.id, {}) return {"status": "running", "queued": res.get("queued")} @router.post("/{cid}/cancel") async def cancel(cid: int, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user), _: None = Depends(require_csrf)): c = await _get_own(db, user, cid) if c.status not in ("running", "paused", "queued", "draft"): raise HTTPException(status_code=409, detail=f"Cannot cancel a {c.status} campaign") res = await cancel_campaign(db, c) await log_audit(db, user.id, "campaign.cancel", "campaign", c.id, {}) return {"status": "cancelled"} # --------------------------------------------------------------------------- # Monitor: progress, SSE stream, recipient table # --------------------------------------------------------------------------- async def _progress_payload(db: AsyncSession, c: Campaign) -> dict: """Live counters for the monitor: campaign fields + per-status breakdown.""" breakdown = (await db.execute( select(CampaignRecipient.status, func.count()) .where(CampaignRecipient.campaign_id == c.id) .group_by(CampaignRecipient.status))).all() counts = {s: int(n) for s, n in breakdown} jobs_agg = (await db.execute( select(func.count()).select_from(SendJob).where(SendJob.campaign_id == c.id) )).scalar() or 0 done = counts.get("sent", 0) + counts.get("perm_failed", 0) + counts.get("suppressed", 0) + counts.get("skipped", 0) total = c.total_recipients or 0 return { "campaign_id": c.id, "status": c.status, "total": total, "sent": counts.get("sent", 0), "temp_failed": counts.get("temp_failed", 0), "perm_failed": counts.get("perm_failed", 0), "suppressed": counts.get("suppressed", 0), "skipped": counts.get("skipped", 0), "pending": counts.get("pending", 0), "queued": counts.get("queued", 0), "processing": counts.get("processing", 0), "jobs_total": int(jobs_agg), "completed": done >= total and total > 0 and c.status in ("completed",), "progress_pct": round(100.0 * done / total, 1) if total else 0.0, "started_at": c.started_at.isoformat() if c.started_at else None, "finished_at": c.finished_at.isoformat() if c.finished_at else None, } @router.get("/{cid}/progress") async def get_progress(cid: int, user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): c = await _get_own(db, user, cid) return await _progress_payload(db, c) @router.get("/{cid}/stream") async def campaign_stream(cid: int, request: Request, user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): """SSE live progress stream. Manual implementation (no sse-starlette): a Redis pub/sub subscription on campaign:{id}:progress plus periodic DB polling fallback. Emits event: progress with counters; event: done when campaign finishes.""" c = await _get_own(db, user, cid) import redis.asyncio as aioredis from app.db.session import AsyncSessionLocal async def event_gen(): r = aioredis.Redis.from_url(cfg.REDIS_URL, decode_responses=True) pubsub = r.pubsub() channel = f"campaign:{cid}:progress" await pubsub.subscribe(channel) heartbeat = 2.0 poll_interval = 5.0 last_payload = None last_poll = 0.0 import time try: yield "event: connected\ndata: {\"ok\": true}\n\n" while True: if await request.is_disconnected(): break # check for pub/sub tick (non-blocking) msg = await pubsub.get_message(ignore_subscribe_messages=True, timeout=0.1) now = time.monotonic() payload = None if msg is not None or now - last_poll >= poll_interval: # refresh counters from DB with a fresh session async with AsyncSessionLocal() as s: camp = await s.get(Campaign, cid) if camp: payload = await _progress_payload(s, camp) last_poll = now if payload is None and msg is None and now - last_poll < poll_interval: await asyncio.sleep(heartbeat) yield ": keepalive\n\n" continue if payload: data = json.dumps(payload) if data != last_payload or msg is not None: yield f"event: progress\ndata: {data}\n\n" last_payload = data if payload.get("status") in ("completed", "cancelled", "failed"): yield f"event: done\ndata: {json.dumps(payload)}\n\n" break await asyncio.sleep(0.2) finally: try: await pubsub.unsubscribe(channel) await pubsub.close() await r.aclose() except Exception: pass return StreamingResponse(event_gen(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no", "Connection": "keep-alive"}) async def _monitor_maps(db: AsyncSession, cid: int, rows): """Resolve display names for a set of campaign recipients. Returns (acc_map, subj_map, tpl_map, rendered_map) where rendered_map holds the *actually rendered* subject from the recipient's latest send attempt (send_attempts.subject) — preferred over the raw rotation template so the monitor/export show exactly what was sent. """ acc_map, subj_map, tpl_map, rendered_map = {}, {}, {}, {} for cr in rows: if cr.assigned_account_id and cr.assigned_account_id not in acc_map: a = await db.get(GmailAccount, cr.assigned_account_id) acc_map[cr.assigned_account_id] = a.email if a else "?" if cr.generated_subject_id and cr.generated_subject_id not in subj_map: s = await db.get(SubjectLine, cr.generated_subject_id) subj_map[cr.generated_subject_id] = (s.subject[:80] if s else "?") if cr.generated_template_id and cr.generated_template_id not in tpl_map: t = await db.get(EmailTemplate, cr.generated_template_id) tpl_map[cr.generated_template_id] = (t.name if t else "?") if rows: emails = [cr.email for cr in rows] att_rows = (await db.execute( select(SendAttempt.recipient_email, SendAttempt.subject) .where(SendAttempt.campaign_id == cid, SendAttempt.recipient_email.in_(emails)) .order_by(SendAttempt.id))).all() for em, subj in att_rows: if subj: rendered_map[em] = subj # ordered by id → last wins return acc_map, subj_map, tpl_map, rendered_map @router.get("/{cid}/recipients") async def campaign_recipients(cid: int, status: str = Query(""), search: str = Query(""), page: int = Query(1, ge=1), page_size: int = Query(50, ge=1, le=200), db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)): """Monitor table: per-recipient status, assigned account, subject/template, errors.""" c = await _get_own(db, user, cid) q = select(CampaignRecipient).where(CampaignRecipient.campaign_id == cid) if status: q = q.where(CampaignRecipient.status == status) if search: q = q.where(CampaignRecipient.email.ilike(f"%{search}%")) total = (await db.execute(select(func.count()).select_from(q.subquery()))).scalar() or 0 rows = (await db.execute(q.order_by(CampaignRecipient.seq) .offset((page - 1) * page_size).limit(page_size))).scalars().all() acc_map, subj_map, tpl_map, rendered_map = await _monitor_maps(db, cid, rows) out = [] for cr in rows: out.append({ "id": cr.id, "seq": cr.seq, "email": cr.email, "status": cr.status, "account": acc_map.get(cr.assigned_account_id, ""), # Prefer the actually-rendered subject from the latest send attempt; # fall back to the raw rotation template until an attempt exists. "subject": rendered_map.get(cr.email) or subj_map.get(cr.generated_subject_id, ""), "template": tpl_map.get(cr.generated_template_id, ""), "invoice_number": cr.generated_invoice_number, "amount": cr.generated_amount, "shipping": cr.generated_shipping, "attempts": cr.attempt_count, "message_id": cr.msg_id, "thread_id": cr.thread_id, "sent_at": cr.sent_at.isoformat() if cr.sent_at else None, "last_error": (cr.last_error or "")[:200], }) return {"recipients": out, "total": total, "page": page, "page_size": page_size} @router.get("/{cid}/export") async def export_campaign(cid: int, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)): """CSV export of campaign results (one row per campaign recipient).""" c = await _get_own(db, user, cid) rows = (await db.execute( select(CampaignRecipient).where(CampaignRecipient.campaign_id == cid) .order_by(CampaignRecipient.seq))).scalars().all() fname = re.sub(r"[^A-Za-z0-9_-]+", "_", c.name)[:60] or f"campaign_{cid}" acc_map, subj_map, tpl_map, rendered_map = await _monitor_maps(db, cid, rows) def generate(): buf = io.StringIO() w = csv.writer(buf) w.writerow(["seq", "email", "status", "account", "subject", "template", "invoice_number", "amount", "shipping", "attempts", "message_id", "thread_id", "sent_at", "last_error"]) yield buf.getvalue() buf.seek(0); buf.truncate(0) for cr in rows: w.writerow([cr.seq, cr.email, cr.status, acc_map.get(cr.assigned_account_id, ""), rendered_map.get(cr.email) or subj_map.get(cr.generated_subject_id, ""), tpl_map.get(cr.generated_template_id, ""), cr.generated_invoice_number, cr.generated_amount or "", cr.generated_shipping or "", cr.attempt_count, cr.msg_id, cr.thread_id, cr.sent_at.isoformat() if cr.sent_at else "", (cr.last_error or "")[:300]]) yield buf.getvalue() buf.seek(0); buf.truncate(0) return StreamingResponse(generate(), media_type="text/csv", headers={"Content-Disposition": f'attachment; filename="{fname}.csv"'}) # --------------------------------------------------------------------------- # Wizard options: what lists/subjects/templates/accounts/phones are selectable # --------------------------------------------------------------------------- @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), _: None = Depends(require_csrf)): """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") if step == 4: ids = data.get("template_ids") or [] if not ids: errors.append("Select at least one template") mode = data.get("template_sequence_mode") if mode not in VALID_ENUMS["template_sequence_mode"]: errors.append("Invalid template sequence mode") if step == 5: mode = data.get("attach_mode", "none") if mode not in VALID_ENUMS["attach_mode"]: errors.append("Invalid attachment mode") elif mode in ("pdf", "file", "image") and not ( data.get("attach_pdf_template_id") or data.get("attachment_id") ): # body_pdf / body_image / body_word need no template or stored # attachment — they convert the rendered email body itself. errors.append("Select a template/attachment for PDF or file mode") if step == 6: vc = data.get("variable_config") or {} try: if "amount_min" in vc and "amount_max" in vc: if float(vc["amount_min"]) > float(vc["amount_max"]): errors.append("Amount min must be <= max") if "shipping_min" in vc and "shipping_max" in vc: if float(vc["shipping_min"]) > float(vc["shipping_max"]): errors.append("Shipping min must be <= max") except (TypeError, ValueError): errors.append("Ranges must be numeric") if step == 7: if not data.get("account_ids"): errors.append("Select at least one Gmail account") elif len(data["account_ids"]) > cfg.MAX_GMAIL_ACCOUNTS_PER_CAMPAIGN: errors.append(f"Max {cfg.MAX_GMAIL_ACCOUNTS_PER_CAMPAIGN} accounts") if step == 8: strategy = data.get("rotation_strategy") if strategy not in VALID_ENUMS["rotation_strategy"]: errors.append("Invalid rotation strategy") if strategy == "weighted": weights = data.get("rotation_weights") or {} if not weights: errors.append("Weighted rotation requires weights per account") if step == 9: try: mpa = int(data.get("max_per_account", 200)) if not (1 <= mpa <= 2000): errors.append("max_per_account must be 1-2000") except (TypeError, ValueError): errors.append("max_per_account must be an integer") try: si = float(data.get("send_interval_seconds", 1.0)) if not (0.5 <= si <= 3600): errors.append("send_interval_seconds must be 0.5-3600") except (TypeError, ValueError): errors.append("send_interval_seconds must be numeric") if step == 10: mode = data.get("phone_selection") if mode not in VALID_ENUMS["phone_selection"]: errors.append("Invalid phone selection mode") if step == 11: if not data.get("unsubscribe_enabled", True): errors.append("Unsubscribe handling must stay enabled (compliance requirement)") return {"step": step, "valid": not errors, "errors": errors}