"""Deterministic message template engine. This module is the whole of LeadPilot's "message generation". There is no model, no API key and no network call: a message is produced by template selection (rules) -> placeholder substitution -> conditional blocks -> rule-driven personalisation phrases -> whitespace cleanup Given the same lead, campaign and templates the output is byte-for-byte stable. Supported template syntax ------------------------- ``{firstName}`` merge field ``{firstName|there}`` merge field with a fallback when the value is empty ``{{#if industry contains "real estate"}}...{{else}}...{{/if}}`` conditional block using any rule-engine operator """ from __future__ import annotations import hashlib import re import shlex from dataclasses import dataclass, field as dataclass_field from typing import Any from app.models.enums import RuleMatchType from app.rules.fields import lead_to_context, normalize_field from app.rules.operators import RuleEvaluationError, apply_operator PLACEHOLDER_RE = re.compile(r"\{([a-zA-Z][a-zA-Z0-9_]*)(?:\|([^{}]*))?\}") # Non-greedy up to the closing "}}", so a lone "}" inside a condition value is fine. BLOCK_TOKEN_RE = re.compile(r"\{\{\s*(#if|else|/if)([\s\S]*?)\}\}") #: Merge fields offered in the UI's template editor. TEMPLATE_VARIABLES: list[dict[str, str]] = [ {"token": "{firstName}", "description": "Lead first name"}, {"token": "{lastName}", "description": "Lead last name"}, {"token": "{fullName}", "description": "Lead full name"}, {"token": "{company}", "description": "Company name"}, {"token": "{jobTitle}", "description": "Job title"}, {"token": "{industry}", "description": "Industry"}, {"token": "{location}", "description": "Location"}, {"token": "{headline}", "description": "LinkedIn headline"}, {"token": "{website}", "description": "Company website"}, {"token": "{companySize}", "description": "Company size (headcount)"}, {"token": "{companyDescriptor}", "description": "Derived: growing company / team / organisation"}, {"token": "{industryPhrase}", "description": "Derived: industry specific sentence"}, {"token": "{roleLabel}", "description": "Derived: CEO / founder / owner wording"}, {"token": "{senderName}", "description": "Your name (Settings)"}, {"token": "{senderCompany}", "description": "Your company (Settings)"}, {"token": "{senderTitle}", "description": "Your job title (Settings)"}, {"token": "{calendarLink}", "description": "Your booking link (Settings)"}, ] #: Deterministic industry sentences used by {industryPhrase}. INDUSTRY_PHRASES: dict[str, str] = { "real estate": "I've been working with businesses in the real estate space on tools that keep listings and client follow-ups moving.", "construction": "I've been helping construction teams replace spreadsheet workflows with something their crews actually use.", "healthcare": "I've been working with healthcare teams on scheduling and intake tooling.", "finance": "I've been building reporting and client-portal tooling for finance teams.", "insurance": "I've been working with insurance teams on quoting and policy workflows.", "software": "I've been helping software teams ship internal tooling faster.", "technology": "I've been helping technology teams ship internal tooling faster.", "marketing": "I've been building campaign reporting and CRM automation for marketing teams.", "retail": "I've been working with retail businesses on storefronts and inventory workflows.", "e-commerce": "I've been working with e-commerce teams on storefronts and order automation.", "education": "I've been working with education providers on enrolment and portal tooling.", "legal": "I've been working with legal teams on document and intake workflows.", "hospitality": "I've been working with hospitality businesses on booking and guest workflows.", "manufacturing": "I've been working with manufacturers on production tracking dashboards.", "logistics": "I've been working with logistics teams on tracking and dispatch tooling.", "consulting": "I've been helping consultancies productise their delivery with internal tools.", } #: Deterministic role wording used by {roleLabel}. ROLE_LABELS: list[tuple[str, str]] = [ ("chief executive", "as CEO"), ("ceo", "as CEO"), ("founder", "as a founder"), ("co-founder", "as a founder"), ("owner", "as the owner"), ("president", "as president"), ("managing director", "as managing director"), ("director", "on the leadership team"), ("head of", "leading the team"), ("manager", "in your team"), ("cto", "on the technical side"), ("vp", "on the leadership team"), ] #: Fallback used when a campaign has no templates at all. FALLBACK_TEMPLATE_BODY = ( "Hi {firstName|there}, I came across {company|your work} " '{{#if industry is_not_empty}}and noticed you\'re working in {industry}{{/if}}. ' "Would be great to connect." ) DEFAULT_FALLBACKS: dict[str, str] = { "firstName": "there", "company": "your company", "industry": "your industry", "location": "your area", "jobTitle": "your role", "companyDescriptor": "growing team", } class TemplateError(ValueError): """Raised when a template cannot be parsed.""" # --------------------------------------------------------------------------- # Conditional block parsing # --------------------------------------------------------------------------- @dataclass(slots=True) class TextNode: text: str @dataclass(slots=True) class IfNode: condition: str then_nodes: list[Any] = dataclass_field(default_factory=list) else_nodes: list[Any] = dataclass_field(default_factory=list) def parse_blocks(template: str) -> list[Any]: """Parse ``{{#if}}/{{else}}/{{/if}}`` blocks into a small node tree.""" root: list[Any] = [] stack: list[tuple[IfNode, str]] = [] cursor = 0 def current_list() -> list[Any]: if not stack: return root node, branch = stack[-1] return node.then_nodes if branch == "then" else node.else_nodes for match in BLOCK_TOKEN_RE.finditer(template): literal = template[cursor : match.start()] if literal: current_list().append(TextNode(literal)) cursor = match.end() keyword = match.group(1) argument = match.group(2).strip() if keyword == "#if": node = IfNode(condition=argument) current_list().append(node) stack.append((node, "then")) elif keyword == "else": if not stack: raise TemplateError("{{else}} outside of an {{#if}} block") node, _ = stack.pop() stack.append((node, "else")) else: # "/if" if not stack: raise TemplateError("{{/if}} without a matching {{#if}}") stack.pop() if stack: raise TemplateError("unclosed {{#if}} block") tail = template[cursor:] if tail: current_list().append(TextNode(tail)) return root def evaluate_condition(condition: str, context: dict[str, Any]) -> bool: """Evaluate ``field operator [value] [value2]`` against the context.""" condition = condition.strip() if not condition: return False try: parts = shlex.split(condition) except ValueError as exc: raise TemplateError(f"invalid condition '{condition}': {exc}") from exc if len(parts) < 2: raise TemplateError(f"invalid condition '{condition}'") field = normalize_field(parts[0]) operator = parts[1] value = parts[2] if len(parts) > 2 else None value2 = parts[3] if len(parts) > 3 else None try: return apply_operator(operator, context.get(field), value, value2) except RuleEvaluationError as exc: raise TemplateError(f"invalid condition '{condition}': {exc}") from exc def _render_nodes(nodes: list[Any], context: dict[str, Any]) -> str: out: list[str] = [] for node in nodes: if isinstance(node, TextNode): out.append(node.text) else: branch = node.then_nodes if evaluate_condition(node.condition, context) else node.else_nodes out.append(_render_nodes(branch, context)) return "".join(out) # --------------------------------------------------------------------------- # Placeholder substitution # --------------------------------------------------------------------------- def render_placeholders(text: str, context: dict[str, Any]) -> str: """Replace ``{field}`` and ``{field|fallback}`` tokens.""" def substitute(match: re.Match[str]) -> str: key = normalize_field(match.group(1)) fallback = match.group(2) value = context.get(key) if value is None or (isinstance(value, str) and not value.strip()): if fallback is not None: return fallback return DEFAULT_FALLBACKS.get(key, "") if isinstance(value, (list, tuple)): return ", ".join(str(item) for item in value) return str(value) return PLACEHOLDER_RE.sub(substitute, text) def clean_whitespace(text: str) -> str: """Tidy the artefacts left behind by empty merge fields and dropped blocks. Paragraph breaks survive: only runs of three or more newlines collapse. Only duplicated *separator* punctuation is squashed, so an author's "..." or "!!" is left intact while ", ," from an empty merge field is not. """ text = re.sub(r"[ \t]+", " ", text) text = re.sub(r" +([,.!?;:])", r"\1", text) text = re.sub(r"([,;:])(\s*\1)+", r"\1", text) text = re.sub(r"[ \t]+\n", "\n", text) text = re.sub(r"\n{3,}", "\n\n", text) text = re.sub(r"(^|\n)[ \t]+", r"\1", text) return text.strip() # --------------------------------------------------------------------------- # Personalisation context # --------------------------------------------------------------------------- def company_descriptor(size: int | None) -> str: if size is None: return "growing team" if size < 20: return "growing company" if size < 200: return "growing team" if size < 1000: return "organisation" return "large organisation" def industry_phrase(industry: str | None) -> str: if not industry: return "" needle = industry.strip().lower() for key, phrase in INDUSTRY_PHRASES.items(): if key in needle: return phrase return "" def role_label(job_title: str | None) -> str: if not job_title: return "" needle = job_title.strip().lower() for key, label in ROLE_LABELS: if key in needle: return label return "" def build_context( lead: Any, sender: Any | None = None, campaign: Any | None = None, extra: dict[str, Any] | None = None, ) -> dict[str, Any]: """Assemble the full merge context: lead fields + derived tokens + sender.""" context = lead_to_context(lead) first_name = context.get("firstName") if not first_name and context.get("fullName"): first_name = str(context["fullName"]).split()[0] context["firstName"] = first_name context["companyDescriptor"] = company_descriptor(context.get("companySize")) context["industryPhrase"] = industry_phrase(context.get("industry")) context["roleLabel"] = role_label(context.get("jobTitle")) # Sender fields default to empty rather than to a greeting word: "Best, there" # reads worse than "Best,". A template can still opt in with {senderName|there}. context["senderName"] = getattr(sender, "name", None) or "" context["senderCompany"] = getattr(sender, "company", None) or "" context["senderTitle"] = getattr(sender, "job_title", None) or "" context["calendarLink"] = getattr(sender, "calendar_link", None) or "" context["campaignName"] = getattr(campaign, "name", None) or "" context["outreachObjective"] = getattr(campaign, "outreach_objective", None) or "" if extra: context.update(extra) return context # --------------------------------------------------------------------------- # Rendering + template selection # --------------------------------------------------------------------------- def render_template(body: str, context: dict[str, Any]) -> str: """Render one template body against a merge context.""" rendered = _render_nodes(parse_blocks(body), context) rendered = render_placeholders(rendered, context) return clean_whitespace(rendered) def template_matches(template: Any, context: dict[str, Any]) -> bool: """Deterministic template selection: do the template's conditions hold?""" conditions = getattr(template, "conditions", None) or [] if not conditions: return True match_type = getattr(template, "match_type", RuleMatchType.ALL) outcomes: list[bool] = [] for condition in conditions: field = normalize_field(condition.get("field", "")) try: outcomes.append( apply_operator( condition.get("operator", "contains"), context.get(field), condition.get("value"), condition.get("value2"), bool(condition.get("caseSensitive", condition.get("case_sensitive", False))), ) ) except RuleEvaluationError: outcomes.append(False) if not outcomes: return True return all(outcomes) if match_type == RuleMatchType.ALL else any(outcomes) def select_template( templates: list[Any], context: dict[str, Any], category: str | None = None, preferred_keys: list[str] | None = None, ) -> Any | None: """Pick a template deterministically. Order of preference: 1. a template named by a ``select_template`` rule (``preferred_keys``) 2. the highest-priority template whose conditions match 3. the campaign default template 4. any active template in the category """ candidates = [ template for template in templates if getattr(template, "is_active", True) and (category is None or getattr(template, "category", None) == category) ] if not candidates: return None if preferred_keys: wanted = {key.strip().lower() for key in preferred_keys} for template in candidates: name = (getattr(template, "name", "") or "").strip().lower() if name in wanted: return template matching = [template for template in candidates if template_matches(template, context)] pool = matching or candidates pool = sorted( pool, key=lambda template: ( 0 if (getattr(template, "conditions", None) or []) else 1, # a campaign's own template outranks a shared one at equal priority 0 if getattr(template, "campaign_id", None) else 1, getattr(template, "priority", 100), getattr(template, "id", 0) or 0, ), ) defaults = [template for template in pool if getattr(template, "is_default", False)] if defaults and not matching: return defaults[0] return pool[0] def select_variant(template: Any, seed: str) -> Any | None: """Pick one of a template's variations deterministically from a stable seed. Uses a hash of the lead identity, so the same lead always gets the same variation while different leads spread evenly across the variations. """ variants = [v for v in (getattr(template, "variants", None) or []) if getattr(v, "is_active", True)] if not variants: return None weighted: list[Any] = [] for variant in variants: weighted.extend([variant] * max(1, int(getattr(variant, "weight", 1) or 1))) digest = hashlib.sha256((seed or "leadpilot").encode("utf-8")).hexdigest() index = int(digest[:8], 16) % len(weighted) return weighted[index] def apply_phrases(message: str, phrases: list[str]) -> str: """Append rule-driven personalisation sentences to a rendered message.""" if not phrases: return message body = message.rstrip() additions = [phrase.strip() for phrase in phrases if phrase and phrase.strip()] if not additions: return message return clean_whitespace(body + " " + " ".join(additions))