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), @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)