fr.company_phone = "555-0100" fr.customer_email = str(payload.to_email) fr.custom_fields = {} engine = VariableEngine({}) g = GeneratedValues() if payload.use_random_variables: g.invoice_number = engine.generate_invoice_number(set()) g.amount = engine.generate_amount() g.shipping = engine.generate_shipping() g.invoice_date = "2025-01-15" g.due_date = "2025-01-30" unsub_url = f"{cfg.BASE_URL}/unsubscribe/test-token" res_html = engine.substitute(body_html or "", fr, g, custom_fields={}, extra={"unsubscribe_link": unsub_url, "company_phone": fr.company_phone}).text res_text = engine.substitute(body_text or "", fr, g, custom_fields={}, extra={"unsubscribe_link": unsub_url, "company_phone": fr.company_phone}).text subject = engine.substitute(payload.subject, fr, g).text attachments = [] if payload.include_pdf and body_html: pdf = await asyncio.get_event_loop().run_in_executor( None, message_builder.html_to_pdf, res_html) if pdf: attachments.append({"filename": "Test_Preview.pdf", "content": pdf, "mime_type": "application/pdf"}) mime = message_builder.build_mime_message( from_email=account.email, to_email=str(payload.to_email), subject=subject, html_body=res_html, text_body=res_text, attachments=attachments, unsubscribe_url=unsub_url, ) # REAL send in a worker thread with a sync DB session (token refresh writes) def _real_send(): with SyncSession() as s: acc = s.get(GmailAccount, account.id) token = s.query(OAuthToken).filter_by(account_id=account.id).first() if not token: return {"ok": False, "error": "No OAuth token stored for this account", "error_class": "auth"} client = build_gmail_client_from_account(acc, token) if client.method == "smtp": return client.send_mime_smtp(mime) return client.send_mime_api(mime) try: result = await asyncio.get_event_loop().run_in_executor(None, _real_send) except Exception as e: await log_audit(db, user.id, "test_send.error", "gmail_account", account.id, {"error": str(e)[:200]}) return {"ok": False, "stage": "temp_failed", "detail": str(e)[:400], "error_class": "unknown", "elapsed_ms": int((time.monotonic() - t0) * 1000)} elapsed = int((time.monotonic() - t0) * 1000) if result.get("ok"): await log_audit(db, user.id, "test_send.success", "gmail_account", account.id, {"to": str(payload.to_email), "message_id": result.get("message_id")}) return {"ok": True, "stage": "accepted", "detail": "Gmail accepted the message", "message_id": result.get("message_id"), "thread_id": result.get("thread_id"), "method": client_method(account), "provider_response": _clip(str(result.get("raw_response", ""))), "elapsed_ms": elapsed} # provider refused — surface the REAL error, never fake success err = str(result.get("error", "") or result.get("detail", "")) await log_audit(db, user.id, "test_send.failed", "gmail_account", account.id, {"to": str(payload.to_email), "error": err[:200]}) return {"ok": False, "stage": "perm_failed", "detail": err[:400], "error_class": result.get("error_class", "unknown"), "method": client_method(account), "provider_response": _clip(str(result.get("raw_response", ""))), "elapsed_ms": elapsed} def _clip(s: str) -> str: return s[:300] def client_method(account: GmailAccount) -> str: return account.sending_method or "api" @router.get("/accounts") async def test_accounts(db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)): """Authorized accounts available for test sends.""" rows = (await db.execute( select(GmailAccount).where(GmailAccount.user_id == user.id, GmailAccount.status == "authorized", GmailAccount.is_active == True))).scalars().all() # noqa: E712 return {"accounts": [{"id": a.id, "email": a.email, "sending_method": a.sending_method} for a in rows]} @router.get("/templates") async def test_templates(db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)): rows = (await db.execute( select(EmailTemplate).where(EmailTemplate.user_id == user.id))).scalars().all() return {"templates": [{"id": t.id, "name": t.name} for t in rows]} =====CAMP_LIST===== @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") =====STREAM===== @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"}) @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() async def names(): acc_map, subj_map, tpl_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 "?") return acc_map, subj_map, tpl_map acc_map, subj_map, tpl_map = await names() 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, ""), "subject": 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)