"""Campaign engine: materialization, rotation, sequencing, assignment. The engine generates per-recipient values (invoice number, amount, shipping, subject, template, phone, sender account) with collision-checked uniqueness and full audit data. """ import random from datetime import date, datetime, timedelta, timezone from typing import List, Optional, Sequence from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.models import (Campaign, CampaignRecipient, EmailTemplate, GmailAccount, SubjectLine, CompanyPhoneNumber, Recipient, SuppressionEntry) from app.services.variable_engine import VariableEngine, GeneratedValues from app.core.config import settings def utcnow(): 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 RotationAssigner: """Assigns sender accounts per message: sequential or weighted, with limits & cooldown.""" def __init__(self, accounts: list, strategy: str, weights: dict = None, max_per_account: int = 200): self.accounts = [a for a in accounts if a.status == "authorized" and a.is_active] self.strategy = strategy if strategy in ("sequential", "weighted") else "sequential" self.weights = weights or {} self.max_per_account = max_per_account self.usage = {a.id: 0 for a in self.accounts} self.cooling = {a.id: False for a in self.accounts} self._seq_index = 0 self._pool = [] self._pool_pos = 0 def _available(self): avail = [] for a in self.accounts: if self.usage[a.id] >= self.max_per_account: continue if a.cooldown_until and a.cooldown_until > utcnow(): continue avail.append(a) return avail def next_account(self, account_id=None): """Return next account or None if all at limit/cooldown.""" if account_id is not None: for a in self.accounts: if a.id == account_id: self.usage[a.id] += 1 return a avail = self._available() if not avail: return None if self.strategy == "weighted": if not self._pool: pool = [] for a in avail: w = int(self.weights.get(str(a.id), 1)) pool.extend([a] * max(1, w)) random.shuffle(pool) self._pool = pool self._pool_pos = 0 while self._pool and self._pool_pos < len(self._pool): a = self._pool[self._pool_pos] self._pool_pos += 1 if self.usage[a.id] < self.max_per_account and not ( a.cooldown_until and a.cooldown_until > utcnow()): self.usage[a.id] += 1 return a self._pool = [] return self.next_account() # sequential for _ in range(len(self.accounts)): a = self.accounts[self._seq_index % len(self.accounts)] self._seq_index += 1 if self.usage[a.id] < self.max_per_account and not ( a.cooldown_until and a.cooldown_until > utcnow()): self.usage[a.id] += 1 return a return None def status(self): return {"accounts": [{"id": a.id, "email": a.email, "usage": self.usage[a.id], "limit": self.max_per_account} for a in self.accounts]} class SubjectSelector: def __init__(self, subjects: List[SubjectLine], strategy: str = "round_robin"): self.subjects = subjects self.strategy = strategy if strategy in ("round_robin", "random", "fixed_first") else "round_robin" self._idx = 0 def next_subject(self) -> Optional[SubjectLine]: if not self.subjects: return None if self.strategy == "random": return random.choice(self.subjects) if self.strategy == "fixed_first": return self.subjects[0] s = self.subjects[self._idx % len(self.subjects)] self._idx += 1 return s class TemplateSequencer: """Template 1 -> recipient 1, template 2 -> recipient 2 ... then repeat/stop/random.""" def __init__(self, templates: List[EmailTemplate], mode: str = "repeat"): self.templates = templates self.mode = mode if mode in ("repeat", "stop", "random") else "repeat" self._idx = 0 self._stopped = False def next_template(self) -> Optional[EmailTemplate]: if not self.templates: return None if self.mode == "random": return random.choice(self.templates) if self.mode == "stop": if self._idx >= len(self.templates): self._stopped = True return None t = self.templates[self._idx] self._idx += 1 return t t = self.templates[self._idx % len(self.templates)] self._idx += 1 return t @property def stopped(self): return self._stopped class PhoneSelector: def __init__(self, phones: List[CompanyPhoneNumber], mode: str = "round_robin"): self.phones = [p for p in phones if p.is_active] self.mode = mode if mode in ("round_robin", "random", "fixed") else "round_robin" self._idx = 0 def next_phone(self) -> str: if not self.phones: return "" if self.mode == "random": return random.choice(self.phones).phone if self.mode == "fixed": return self.phones[0].phone p = self.phones[self._idx % len(self.phones)] self._idx += 1 return p.phone async def get_setting(db: AsyncSession, key: str, default): from app.models import ApplicationSettings row = (await db.execute(select(ApplicationSettings).where(ApplicationSettings.key == key))).scalar_one_or_none() if row is not None: return row.value return default def get_setting_sync(key: str, default): """Sync-context settings reader (worker).""" from app.db.session import SyncSession from app.models import ApplicationSettings try: with SyncSession() as s: row = s.execute(select(ApplicationSettings).where( ApplicationSettings.key == key)).scalar_one_or_none() return row.value if row is not None else default except Exception: return default async def get_campaign_entities(db: AsyncSession, campaign: Campaign): """Load all entities a campaign references (subjects, templates, phones, accounts).""" subjects = [] if campaign.subject_set_ids: for sid in campaign.subject_set_ids: s = await db.get(SubjectLine, sid) if s: subjects.append(s) templates = [] if campaign.template_ids: for tid in campaign.template_ids: t = await db.get(EmailTemplate, tid) if t: templates.append(t) phones = (await db.execute( select(CompanyPhoneNumber).where(CompanyPhoneNumber.user_id == campaign.user_id, CompanyPhoneNumber.is_active == True))).scalars().all() # noqa: E712 accounts = [] for aid in campaign.account_ids or []: acc = await db.get(GmailAccount, aid) if acc and acc.is_active and acc.status == "authorized": accounts.append(acc) return subjects, templates, phones, accounts async def materialize_campaign(db: AsyncSession, campaign: Campaign, recipient_rows: Sequence[Recipient], preview: bool = False, preview_size: int = 0) -> dict: """Create CampaignRecipient rows with generated values & assignments. This is the core deterministic plan: which subject/template/phone/account each recipient gets, plus generated invoice numbers (unique per campaign), amounts, shipping. """ subjects, templates, phones, accounts = await get_campaign_entities(db, campaign) var_cfg = dict(campaign.variable_config or {}) engine = VariableEngine(var_cfg) subj_sel = SubjectSelector(subjects, campaign.subject_rotation) tpl_seq = TemplateSequencer(templates, campaign.template_sequence_mode) phone_sel = PhoneSelector(phones, campaign.phone_selection) rot = RotationAssigner(accounts, campaign.rotation_strategy, campaign.rotation_weights, campaign.max_per_account) used_invoice_numbers: set = set() rows: List[CampaignRecipient] = [] plan = [] seq = 0 for r in recipient_rows: seq += 1 cr = CampaignRecipient( campaign_id=campaign.id, recipient_id=r.id, email=r.email, seq=seq, ) # generated values g = GeneratedValues() if var_cfg.get("invoice_date"): g.invoice_date = var_cfg["invoice_date"] else: g.invoice_date = date.today().isoformat() if var_cfg.get("due_date"): g.due_date = var_cfg["due_date"] else: # Due date is relative to the EFFECTIVE invoice date (which may be a # custom one), not to today — otherwise a custom invoice_date can end # up with a due date BEFORE the invoice date. g.due_date = (_effective_invoice_date(g.invoice_date) + timedelta(days=int(var_cfg.get("due_days", 14)))).isoformat() inv_num = "" tpl = tpl_seq.next_template() if tpl is None and templates: cr.status = "skipped" cr.last_error = "Template sequence exhausted (mode=stop)" body_uses_invoice = any("%RandomInvoiceNumber%" in (t.body_html + (t.body_text or "")) for t in templates) or (tpl and "%RandomInvoiceNumber%" in tpl.body_html) if body_uses_invoice or var_cfg.get("always_generate_invoice", True): inv_num = engine.generate_invoice_number(used_invoice_numbers) used_invoice_numbers.add(inv_num) cr.generated_invoice_number = inv_num g.invoice_number = inv_num if var_cfg.get("amount_enabled", True) or body_uses_invoice: g.amount = engine.generate_amount() cr.generated_amount = g.amount if var_cfg.get("shipping_enabled", True) or "%RandomShippingCharge%" in (tpl.body_html if tpl else ""): g.shipping = engine.generate_shipping() cr.generated_shipping = g.shipping subj = subj_sel.next_subject() cr.generated_subject_id = subj.id if subj else None cr.generated_template_id = tpl.id if tpl else None phone = phone_sel.next_phone() cr.generated_phone = phone g.phone = phone account = rot.next_account() if account is None: cr.status = "skipped" cr.last_error = "No available sender account (limits or cooldown)" else: cr.assigned_account_id = account.id rows.append(cr) plan.append({"seq": seq, "email": r.email, "invoice": inv_num, "amount": g.amount, "subject": subj.subject if subj else "", "template": tpl.name if tpl else None, "phone": phone, "account": account.email if account else None, "account_id": account.id if account else None, "recipient_id": r.id, "subject_id": subj.id if subj else None, "template_id": tpl.id if tpl else None, "invoice_number": inv_num, "invoice_date": g.invoice_date, "due_date": g.due_date, "shipping": g.shipping, "warnings": cr.last_error if cr.status == "skipped" else "", "status": cr.status}) if preview and seq >= (preview_size or campaign.preview_sample_size or 5): break if not preview: db.add_all(rows) await db.flush() return {"rows": rows, "plan": plan, "rotation": rot.status(), "total": len(rows), "queued": sum(1 for r in rows if r.status == "pending"), "account_count": len(accounts), "subjects": len(subjects), "templates": len(templates), "phones": len(phones)}