"""Template variable engine. Supported variables (case-sensitive, %Delimited%): Recipient: %FirstName% %LastName% %CompanyName% %CompanyPhone% %CustomerEmail% %Email% plus custom CSV columns (auto-detected) Generated: %RandomInvoiceNumber% %RandomTotal% %RandomShippingCharge% %RandomName% Configured: %InvoiceDate% %DueDate% %CompanyPhone% System: %UnsubscribeLink% Unknown variables produce validation warnings (surfaced before campaign start). """ import re import secrets import string from dataclasses import dataclass, field from datetime import date, datetime from typing import Optional VARIABLE_PATTERN = re.compile(r"%([A-Za-z0-9_]+)%") KNOWN_VARIABLES = { "FirstName", "LastName", "CompanyName", "CompanyPhone", "CustomerEmail", "Email", "RandomInvoiceNumber", "RandomTotal", "RandomShippingCharge", "RandomName", "InvoiceDate", "DueDate", "UnsubscribeLink", } @dataclass class GeneratedValues: invoice_number: str = "" amount: Optional[float] = None shipping: Optional[float] = None random_name: str = "" invoice_date: str = "" due_date: str = "" phone: str = "" @dataclass class SubstitutionResult: text: str unknown: list = field(default_factory=list) missing: list = field(default_factory=list) # known but no data available used: list = field(default_factory=list) class VariableEngine: def __init__(self, config: Optional[dict] = None, rng=None): self.config = config or {} self.rng = rng or secrets.SystemRandom() # ------------------------------------------------------------------ helpers def _cfg(self, key: str, default): return self.config.get(key, default) def generate_invoice_number(self, used: set = None) -> str: """Generate a unique invoice/reference number using configurable format tokens. Tokens: {prefix} {R#} uppercase {r#} lowercase {d#} digits {seq4} {date} {hash} Cryptographically strong randomness via SystemRandom. """ used = used if used is not None else set() fmt = self._cfg("invoice_format", "{prefix}{R4}-{R5}-{R6}-{seq4}") prefix = self._cfg("invoice_prefix", "INV-") for _ in range(50): val = self._render_format(fmt, prefix) if val not in used: return val raise ValueError("Could not generate unique invoice number after 50 attempts") def _render_format(self, fmt: str, prefix: str) -> str: from string import ascii_letters, digits upper = string.ascii_uppercase + string.digits lower = string.ascii_lowercase + string.digits digits_only = string.digits def rnd(chars, n): return "".join(self.rng.choice(chars) for _ in range(n)) def repl(match): token = match.group(1) if token == "prefix": return prefix if token == "date": return datetime.utcnow().strftime("%Y%m%d") if token.startswith("R"): # e.g. R4, R5 try: n = int(token[1:]) except ValueError: return token return rnd(upper, n) if token.startswith("r"): try: n = int(token[1:]) except ValueError: return token return rnd(lower, n) if token.startswith("d"): try: n = int(token[1:]) except ValueError: return token return rnd(digits_only, n) if token == "seq4": return rnd(digits_only, 4) if token.startswith("hash"): n = int(token[4:] or 8) return secrets.token_hex(n) return match.group(0) return re.sub(r"\{([A-Za-z0-9_]+)(\d*)\}", repl, fmt) def generate_amount(self) -> float: lo = float(self._cfg("amount_min", 300.0)) hi = float(self._cfg("amount_max", 1500.0)) if hi < lo: lo, hi = hi, lo return round(self.rng.uniform(lo, hi), 2) def generate_shipping(self) -> float: lo = float(self._cfg("shipping_min", 5.0)) hi = float(self._cfg("shipping_max", 50.0)) if hi < lo: lo, hi = hi, lo return round(self.rng.uniform(lo, hi), 2) def generate_random_name(self) -> str: first = ("James", "Mary", "Robert", "Patricia", "John", "Jennifer", "Michael", "Linda", "David", "Elizabeth", "William", "Barbara", "Richard", "Susan", "Joseph", "Jessica", "Thomas", "Sarah", "Charles", "Karen", "Ava", "Noah") last = ("Smith", "Johnson", "Williams", "Brown", "Jones", "Garcia", "Miller", "Davis", "Rodriguez", "Martinez", "Hernandez", "Lopez", "Gonzalez", "Wilson", "Anderson", "Thomas", "Taylor", "Moore", "Jackson", "Martin") return f"{self.rng.choice(first)} {self.rng.choice(last)}" # ------------------------------------------------------------------ substitution def substitute(self, text: str, recipient, generated: GeneratedValues, custom_fields: Optional[dict] = None, extra: Optional[dict] = None) -> SubstitutionResult: recipient = recipient or {} custom_fields = custom_fields or {} extra = extra or {} # Accept ORM Recipient objects as well as plain dicts if not isinstance(recipient, dict): recipient = {c.name: getattr(recipient, c.name, None) for c in recipient.__table__.columns} # Normalize: snake_case columns -> CamelCase variables (first_name -> FirstName) norm = {} for k, v in recipient.items(): norm[k] = v camel = "".join(p.capitalize() for p in str(k).split("_")) norm[camel] = v norm["Email"] = norm.get("Email", norm.get("email")) # custom CSV columns pass through with their raw names for k, v in list(custom_fields.items()): norm[k] = v camel = "".join(p.capitalize() for p in str(k).split("_")) norm[camel] = v recipient = norm unknown, missing, used = [], [], [] pool = dict(recipient) pool.update({k: v for k, v in extra.items() if k != "unsubscribe_link"}) def replace_fn(m): var = m.group(1) if var in ("RandomInvoiceNumber", "RandomTotal", "RandomShippingCharge", "RandomName", "InvoiceDate", "DueDate", "UnsubscribeLink", "CompanyPhone"): return m.group(0) # handled below by generated values pass if var in pool: val = pool.get(var) if val is None or str(val).strip() == "": if var in KNOWN_VARIABLES: missing.append(var) return f"[{var}]" used.append(var) return str(val) if var in KNOWN_VARIABLES: missing.append(var) return f"[{var}]" unknown.append(var) return f"[UNKNOWN:{var}]" text = re.sub(VARIABLE_PATTERN, replace_fn, text) # Generated / configured values second pass gen_map = { "RandomInvoiceNumber": generated.invoice_number, "RandomTotal": f"${generated.amount:,.2f}" if generated.amount is not None else None, "RandomShippingCharge": f"${generated.shipping:,.2f}" if generated.shipping is not None else None, "RandomName": generated.random_name, "InvoiceDate": generated.invoice_date, "DueDate": generated.due_date, "CompanyPhone": generated.phone, "UnsubscribeLink": extra.get("unsubscribe_link", ""), } for var, val in gen_map.items(): if val not in (None, ""): text = text.replace(f"%{var}%", str(val)) used.append(var) return SubstitutionResult(text=text, unknown=sorted(set(unknown)), missing=sorted(set(missing)), used=sorted(set(used))) # ------------------------------------------------------------------ analysis def analyze(self, text: str, custom_fields: Optional[list] = None) -> dict: """Report which variables a template uses and any unknown ones (pre-send validation).""" found = set(VARIABLE_PATTERN.findall(text or "")) custom_fields = set(custom_fields or []) unknown = sorted(found - KNOWN_VARIABLES - custom_fields) return {"variables": sorted(found), "unknown": unknown} def analyze_template_variables(text: str, custom_fields: Optional[list] = None) -> dict: return VariableEngine().analyze(text, custom_fields)