"""Send task: real Gmail API/SMTP send with retry, backoff, DLQ, pause/cancel, rate limits. Status model (no fake success): queued -> processing -> sent (Gmail accepted, message ID recorded) temporary errors -> retry with exponential backoff + jitter, dead-letter after max attempts permanent errors -> perm_failed (no retry) suppressed / skipped -> dropped pre-send """ import logging import random import re as _re import time from datetime import date, datetime, timedelta, timezone import googleapiclient.errors import redis as redis_lib from celery import Task from sqlalchemy import select from app.core.config import settings from app.db.session import SyncSession from app.models import (Attachment, Campaign, CampaignRecipient, EmailEvent, EmailTemplate, GmailAccount, OAuthToken, Recipient, SendAttempt, SendJob, SubjectLine, SuppressionEntry) from app.services import attachments_service, message_builder from app.services.gmail_service import build_gmail_client_from_account from app.services.variable_engine import GeneratedValues, VariableEngine from app.worker.celery_app import celery_app log = logging.getLogger("send") try: rdb = redis_lib.from_url(settings.REDIS_URL, socket_timeout=2) except Exception: rdb = None TEMPORARY_PATTERNS = ["rate limit", "rate-limit", "429", "quota", "backend error", "transient", "temporarily", "503", "502", "500", "timeout", "connection", "try again", "resourceexhausted", "tempfail", "unavailable", "internal error", "deadline"] PERMANENT_PATTERNS = ["invalid grant", "401", "403", "unauthorized", "forbidden", "invalidrecipient", "failedprecondition", "invalid argument", "malformed", "does not exist", "no such user", "550", "553", "554", "rejected", "policy violation", "disabled", "authentication"] def classify_error(msg: str) -> str: m = (msg or "").lower() for p in TEMPORARY_PATTERNS: if p in m: return "temporary" for p in PERMANENT_PATTERNS: if p in m: return "permanent" return "unknown" def now(): return datetime.now(timezone.utc) def _effective_invoice_date(invoice_date_str: str) -> date: """Parse the effective invoice date (ISO) safely; fall back to today.""" try: return date.fromisoformat(invoice_date_str) except (TypeError, ValueError): return date.today() class SendTask(Task): def on_failure(self, exc, task_id, args, kwargs, einfo): log.error("Task %s failed: %s", task_id, str(exc)[:200]) @celery_app.task(bind=True, base=SendTask, name="app.tasks.send_tasks.send_one", ignore_result=False) def send_one(self, campaign_id: int, campaign_recipient_id: int): """Process one recipient: build message, send through Gmail, record real outcome.""" with SyncSession() as db: campaign = db.get(Campaign, campaign_id) if campaign is None: return {"ok": False, "status": "cancelled", "reason": "campaign deleted"} if campaign.status in ("paused", "cancelled", "completed"): return {"ok": False, "status": "cancelled", "reason": f"campaign {campaign.status}"} cr = db.get(CampaignRecipient, campaign_recipient_id) if cr is None: return {"ok": False, "status": "cancelled", "reason": "campaign_recipient deleted"} if cr.status in ("sent", "suppressed", "perm_failed"): return {"ok": True, "status": cr.status, "reason": "already finalized"} job = (db.execute(select(SendJob) .where(SendJob.campaign_recipient_id == cr.id, SendJob.campaign_id == campaign.id))).scalar_one_or_none() # Re-check suppression at send time (recipient may have unsubscribed since enqueue) sup = (db.execute(select(SuppressionEntry) .where(SuppressionEntry.email == cr.email))).scalar_one_or_none() if sup: cr.status = "suppressed" campaign.suppressed_count += 1 cr.last_error = "Suppressed at send time" db.add(EmailEvent(campaign_id=campaign.id, campaign_recipient_id=cr.id, event_type="suppressed", detail="Suppressed at send time")) db.commit() _publish_progress(campaign.id) return {"ok": True, "status": "suppressed", "reason": "suppression hit"} account = db.get(GmailAccount, cr.assigned_account_id) if cr.assigned_account_id else None if account is None or not account.is_active or account.status != "authorized": cr.status = "skipped" campaign.skipped_count += 1 cr.last_error = "Assigned account unavailable/unauthorized" db.commit() _publish_progress(campaign.id) return {"ok": False, "status": "skipped", "reason": "account unavailable"} # Per-account daily cap (never exceed provider limits) today = now().date() if account.last_send_date and account.last_send_date.date() == today \ and account.sends_today >= account.daily_limit: campaign.status = "paused" db.commit() _publish_control(campaign.id, "pause", f"Account {account.email} reached daily limit ({account.daily_limit})") return {"ok": False, "status": "paused", "reason": "daily limit reached"} if account.cooldown_until and account.cooldown_until > now(): _reschedule(db, self, campaign.id, cr, job, account.cooldown_until - now(), "account cooling down") return {"ok": False, "status": "rescheduled", "reason": "cooldown"} # Per-account rate window (0 = disabled; `or` coercion would silently # turn an explicit 0 into 1.0) interval = (float(campaign.send_interval_seconds) if campaign.send_interval_seconds is not None else 1.0) ok, wait = _rate_window(f"acct:{account.id}", interval) if not ok: _reschedule(db, self, campaign.id, cr, job, timedelta(seconds=wait), "rate limit window") return {"ok": False, "status": "rescheduled", "reason": "rate window"} # ------------------------------------------------------------- build message recipient = db.get(Recipient, cr.recipient_id) subject_row = db.get(SubjectLine, cr.generated_subject_id) if cr.generated_subject_id else None template = db.get(EmailTemplate, cr.generated_template_id) if cr.generated_template_id else None subject = subject_row.subject if subject_row else campaign.name if template is None: cr.status = "perm_failed" cr.last_error = "No template assigned" campaign.failed_count += 1 db.commit() return {"ok": False, "status": "perm_failed", "reason": "no template"} var_cfg = dict(campaign.variable_config or {}) engine = VariableEngine(var_cfg) _today = date.today() g = GeneratedValues( invoice_number=cr.generated_invoice_number, amount=cr.generated_amount, shipping=cr.generated_shipping, random_name=engine.generate_random_name(), invoice_date=var_cfg.get("invoice_date") or _today.isoformat(), due_date=var_cfg.get("due_date") or (_effective_invoice_date(var_cfg.get("invoice_date") or _today.isoformat()) + timedelta(days=int(var_cfg.get("due_days", 14)))).isoformat(), phone=cr.generated_phone or var_cfg.get("company_phone", ""), ) unsub_url = _unsubscribe_url(recipient, campaign) rec = _recipient_dict(recipient) cf = (recipient.custom_fields if recipient else None) or {} # Render variables in the subject line too (%RandomTotal%, %FirstName%, # %InvoiceNumber% …) — same engine and context as the body, so the # subject a recipient sees matches the rendered email body. subject = engine.substitute(subject, rec, g, custom_fields=cf, extra={"unsubscribe_link": unsub_url}).text result_html = engine.substitute(template.body_html, rec, g, custom_fields=cf, extra={"unsubscribe_link": unsub_url}).text result_text = engine.substitute(template.body_text or _plain_fallback(result_html), rec, g, custom_fields=cf, extra={"unsubscribe_link": unsub_url}).text attachments, att_names = [], [] # ---- auto-conversion modes: convert the rendered email BODY itself ---- # body_pdf / body_image / body_word take the fully substituted HTML body # and attach it as a real PDF / PNG / Word document per campaign config. if campaign.attach_mode in ("body_pdf", "body_image", "body_word"): _conv = { "body_pdf": ("pdf", message_builder.html_to_pdf, "application/pdf"), "body_image": ("png", message_builder.html_to_image, "image/png"), "body_word": ("docx", message_builder.html_to_docx, "application/vnd.openxmlformats-officedocument" ".wordprocessingml.document"), }[campaign.attach_mode] _ext, _fn, _mime = _conv try: conv_bytes = _fn(result_html) except message_builder.HtmlValidationError as e: cr.status = "perm_failed" cr.last_error = f"Body conversion failed: {e}" campaign.failed_count += 1 db.add(EmailEvent(campaign_id=campaign.id, campaign_recipient_id=cr.id, event_type="perm_failed", detail=f"conversion failed: {e}")) db.commit() _publish_progress(campaign.id) return {"ok": False, "status": "perm_failed", "reason": "body conversion failed"} fname = message_builder.generate_attachment_filename( var_cfg.get("attachment_kind", "invoice"), cr.generated_invoice_number, ext=_ext) attachments.append({"filename": fname, "content": conv_bytes, "mime_type": _mime}) att_names.append(fname) if campaign.attach_mode == "pdf" and campaign.attach_pdf: att_html = "" if campaign.attach_pdf_template_id: att_tpl = db.get(EmailTemplate, campaign.attach_pdf_template_id) if att_tpl: att_html = engine.substitute(att_tpl.body_html, rec, g, custom_fields=cf, extra={"unsubscribe_link": unsub_url}).text elif campaign.attachment_id: att = db.get(Attachment, campaign.attachment_id) if att and att.kind == "html": att_html = engine.substitute(att.html_attachment_body or "", rec, g, custom_fields=cf, extra={"unsubscribe_link": unsub_url}).text if att_html: pdf_bytes = message_builder.html_to_pdf(att_html) fname = message_builder.generate_attachment_filename( var_cfg.get("attachment_kind", "invoice"), cr.generated_invoice_number) attachments.append({"filename": fname, "content": pdf_bytes, "mime_type": "application/pdf"}) att_names.append(fname) if campaign.attach_mode in ("file", "image") and campaign.attachment_id: att = db.get(Attachment, campaign.attachment_id) if att: data = attachments_service.read_attachment_bytes(att) if data is not None: attachments.append({"filename": att.filename, "content": data, "mime_type": att.mime_type}) att_names.append(att.filename) html_problems = message_builder.validate_html(result_html) if html_problems: log.warning("HTML validation warnings for %s: %s", cr.email, html_problems) mime = message_builder.build_mime_message( from_email=account.email, # real authenticated account; never spoofed to_email=cr.email, subject=subject, html_body=result_html if campaign.message_type != "plain" else "", text_body=result_text, attachments=attachments, unsubscribe_url=unsub_url if campaign.unsubscribe_enabled else "",