Add presentation bundle (reports, curves, run log, code) for the Colab notebook
439e7b1 verified 15 days ago
57.4 kB
#!/usr/bin/env python
"""Template-generated synthetic supplement for phishing fine-tuning (TRAINING DATA ONLY).
What it is
  A small (<= config data.synthetic.max_rows = 2,000), label-balanced set of short emails in the
  register a small-business IT consultancy's clients actually see. Positive rows (label 1) imitate the
  lure categories those clients get targeted with; negative rows (label 0) are ordinary business mail
  written in the same tone so the model cannot shortcut on register alone. Every row is fabricated
  from schematic templates plus seeded random slot filling -- no network, no LLM calls, no real
  messages. Organisations and people are placeholders ("Northwind Dental", a fixed fictional name
  list); every domain sits under the reserved .example / .invalid TLDs, IPs use the RFC 5737
  documentation ranges and phone numbers use 555-01xx. Rows carry source="synthetic", doc_id
  "synthetic:NNNNNN", cluster_id -1, and the parquet schema carries
  {"generator": "make_synthetic.py", "note": "template-generated training supplement"}. The file is
  never a deliverable and must never be presented as real mail.
How it is used and reported
  * finetune.py --synthetic on concatenates data/processed/synthetic.parquet onto finetune_train ONLY.
    It never enters finetune_val (model selection / early stopping stay on real data), the TREC-06
    holdout, the tokenizer corpus or MLM pretraining.
  * Every headline metric is reported twice -- one run with and one without the supplement, identical
    dials otherwise -- and reports/experiments.jsonl records dials.synthetic so the comparison is
    auditable. The supplement is ~1.3% of the 157k training rows; its job is to nudge the model toward
    the business-email-compromise / Microsoft 365 lure register that the 2005-2008 spam corpora lack,
    not to move aggregate numbers. Any change on the holdout is reported as-is, including none.
Shortcut control
  Templated data invites the classifier to learn the template rather than the phenomenon. The
  generator therefore
  (a) draws greetings, sign-offs, signatures, disclaimers, HTML wrappers, subject prefixes, casing,
      TLDs, URL schemes and the per-row length target from pools shared by both labels;
  (b) puts a link in BOTH classes at comparable rates: every negative category carries a {url} slot
      pointing at a plausible internal / vendor portal whose host prefixes (login., secure., docs.,
      payments., *.sharepoint.*) and paths (/login, /auth/, /docs/, /inv/, ?id=) overlap the lookalike
      vocabulary used by the lures, so neither "contains a URL", "<a href" nor the URL sub-tokens
      (https, http, login, secure, doc, id, ...) separate the labels;
  (c) writes benign sentences on label 0 that reuse lure vocabulary (password, verify, urgent, overdue,
      DocuSign, OneDrive, wire, gift cards, expires, suspended, immediately, authenticator, approve)
      and mundane sentences on label 1 that reuse notice vocabulary (planned maintenance, posting for
      awareness, we're, just), so the words themselves are not label-predictive;
  (d) measures the result over the WHOLE vocabulary: `build` logs per-token label purity (document
      frequency over lower-cased word tokens) for the 30 most frequent tokens, and flags EVERY token
      with support >= 50 that is > 95% one class. The flagged list is written to
      <stem>_report.json ("token_purity_flagged") and logged, so a residual shortcut is auditable
      rather than hidden by a top-N cut-off. Exact duplicates are removed on common.normalize_text,
      so the written file's duplicate rate is 0.
Three-way labels (label3 / label_set)
  The classifier now outputs Legitimate / Phishing / Spam. Every row carries, beside the binary
  `label`, a `label3` int64 (0 legitimate, 1 phishing; never 2 spam or -1 unwanted here) and a
  `label_set` string ("legit" / "phish"). The mapping is by category, i.e. by the "<label>:<category>"
  keys of the report: every "1:*" category -> label3 1 / "phish", every "0:*" category -> label3 0 /
  "legit". This is correct by construction, not by assumption: all eight positive templates
  (m365_signin, mfa_prompt, helpdesk_quota, fake_fileshare, invoice_lure, bec_payment, bec_giftcard,
  vendor_bankchange) are credential-harvesting, account-takeover or business-email-compromise lures --
  targeted deception aimed at a credential, a payment or a gift-card code -- and none is bulk
  unsolicited advertising, which is what "spam" means in the label contract. The synthetic set therefore
  contributes NO spam rows; spam supervision comes from the real SpamAssassin / Enron-Spam folders.
  PHISH_CATEGORIES pins that list: a new positive template outside it makes `build` fail rather than be
  silently filed as phishing. Adding the two columns does not touch the RNG stream (they are derived
  after generation), so with the same seed every existing column -- doc_id, text, subject, body,
  label, category -- is byte-identical to the binary-era file and earlier experiments stay reproducible.
Length
  Each row draws a target length from a right-skewed triangular distribution over 50-200 words
  (mode at 50) and pads toward it from the shared filler pool, so the supplement spans a band of
  lengths instead of a 50-70 word spike. Bodies are hard-capped at 400 words.
Subcommands
  build [--n N] [--out PATH] [--seed S]   write the parquet + <stem>_report.json beside it
  show  [--k 6] [--path PATH]             print k sample rows per class
"""

from __future__ import annotations

import argparse
import logging
import random
import re
import sys
from collections import Counter
from dataclasses import dataclass, field
from datetime import date, timedelta
from pathlib import Path
from typing import Any

import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq

sys.path.insert(0, str(Path(__file__).parent))
import common  # noqa: E402

LOG = logging.getLogger("make_synthetic")

COLUMNS = ["doc_id", "text", "subject", "body", "label", "source", "cluster_id", "category", "label3", "label_set"]

# Three-way label contract (shared with data_prep / finetune / evaluate): 0 legitimate, 1 phishing, 2 spam.
# The synthetic generator only emits 0 and 1 -- see "Three-way labels" in the module docstring.
LABEL3_NAMES = {0: "Legitimate", 1: "Phishing", 2: "Spam"}
LABEL_SET_OF = {0: "legit", 1: "phish", 2: "spam"}
# Every positive template is a credential / account-takeover / BEC lure, hence phishing, not spam.
PHISH_CATEGORIES = frozenset({"m365_signin", "mfa_prompt", "helpdesk_quota", "fake_fileshare", "invoice_lure",
                              "bec_payment", "bec_giftcard", "vendor_bankchange"})
PARQUET_META = {"generator": "make_synthetic.py", "note": "template-generated training supplement"}
MIN_WORDS, TARGET_MAX_WORDS, MAX_WORDS = 50, 200, 400
PURITY_SUPPORT, PURITY_MAX, PURITY_TOP_N = 50, 0.95, 30
TOKEN_RE = re.compile(r"[a-z][a-z0-9']+")

# ----------------------------------------------------------------------------- placeholder world
ORGS = [
    "Northwind Dental", "Cedar Ridge Plumbing", "Harborview CPA", "Maple Street Veterinary",
    "Bluewater Landscaping", "Summit Peak Physical Therapy", "Oak & Ivy Law Group", "Riverbend Bakery",
    "Pinecrest Family Chiropractic", "Lakeshore Realty Partners", "Ironwood Machine Works",
    "Willow Creek Pediatrics", "Copper Kettle Catering", "Foxglove Interior Design",
]
VENDORS = [
    "Tri-County Office Supply", "Copperleaf Dental Lab", "Evergreen Payroll Services", "Keystone Janitorial",
    "Silverline Medical Supply", "Redwood Print & Sign", "Granite Coast Insurance", "Brookside Uniform Co.",
    "Clearpath Telecom", "Hollow Oak Accounting",
]
CLIENTS = [
    "Juniper Hills HOA", "Starling Logistics", "Meadowbrook Assisted Living", "Bramble & Finch Architects",
    "Sandstone Auto Repair", "Quarry Lane Brewing",
]
MSPS = ["Brightline IT Partners", "Signal Harbor Technology", "Northgate Managed IT"]
FIRST = [
    "Avery", "Jordan", "Morgan", "Casey", "Riley", "Taylor", "Jamie", "Quinn", "Reese", "Dana", "Elliot",
    "Harper", "Rowan", "Sasha", "Devon", "Marisol", "Tobias", "Priya", "Lena", "Malik", "Ingrid", "Theo",
    "Noor", "Callum", "Bea", "Rafael", "Wren", "Odette", "Kofi", "Imani",
]
LAST = [
    "Calloway", "Brightwater", "Fenwick", "Ashdown", "Merriweather", "Okafor", "Lindqvist", "Talbot",
    "Pemberton", "Quill", "Hartsell", "Dunmore", "Ravensworth", "Kilbride", "Marchetti", "Holloway", "Nakamura",
]
OWNER_TITLES = ["Owner", "Practice Owner", "Managing Partner", "President", "CEO", "Founder", "Principal"]
STAFF_TITLES = [
    "Office Manager", "Practice Administrator", "Bookkeeper", "Operations Lead", "Front Desk Coordinator",
    "Accounts Payable", "Project Coordinator", "Senior Associate",
]
PROJECTS = [
    "website refresh", "server migration", "new patient portal", "office network upgrade", "kitchen remodel bid",
    "Q4 tax planning engagement", "warehouse Wi-Fi survey", "phone system cutover", "brand refresh",
    "file share cleanup",
]
FILES = [
    "Q3_Payroll_Summary.xlsx", "Signed_Agreement_2026.pdf", "Remittance_Advice.pdf", "Employee_Handbook_v4.docx",
    "Statement_of_Work.pdf", "Bonus_Schedule.xlsx", "Vendor_W9.pdf", "Project_Timeline.xlsx",
    "Insurance_Renewal.pdf", "Board_Minutes.docx", "Quote_Revised.pdf", "Purchase_Order.pdf",
]
LOCATIONS = ["Frankfurt, Germany", "Toronto, Canada", "Sao Paulo, Brazil", "Singapore", "an unrecognized location",
             "Warsaw, Poland", "Dallas, TX", "Amsterdam, Netherlands"]
DEVICES = ["iPhone", "Windows laptop", "Android phone", "MacBook", "front desk PC", "iPad"]
TLDS = ["example", "example", "invalid"]
SCHEMES = ["https://", "https://", "http://"]
LOOKALIKES = [
    "micr0soft-365-login", "login-microsoftonline-secure", "office365-mailbox-center", "entra-id-auth",
    "sharepoint-docs-access", "onedrive-filetransfer", "docusign-envelope-review", "m365-quota-admin",
    "secure-invoice-portal", "outlook-web-sync", "doc-share-cloud", "payments-remit-portal",
]
# Legitimate hosts reuse the lookalikes' vocabulary (login, secure, docs, payments, sharepoint, portal)
# under the org's / vendor's / MSP's own placeholder domain, the way real tenants do.
LEGIT_HOST_PREFIXES = ["portal", "intranet", "billing", "status", "support", "login", "secure", "docs",
                       "payments", "mail", "tickets", "hr", "calendar"]
LEGIT_PATHS = ["/billing", "/hr/benefits", "/tickets/", "/calendar", "/projects/", "/docs/", "/", "/status",
               "/login", "/auth/", "/inv/", "/s/", "/review?id=", "/sites/", "/personal/", "/tickets?id=",
               "/invoices/", "/quote/", "/m/", "/doc/"]
PHISH_PATHS = ["/auth/", "/s/", "/review?id=", "/login/", "/doc/", "/session/", "/m/", "/inv/", "/docs/",
               "/billing", "/sites/", "/tickets?id="]
GENERIC_ANCHORS = ["click here", "here", "Open in browser", "View details", "Open"]


def slug(name: str) -> str:
    return re.sub(r"-+", "-", re.sub(r"[^a-z0-9]+", "-", name.lower().replace("&", "and"))).strip("-")


def money(rng: random.Random, lo: float, hi: float) -> str:
    return f"${rng.uniform(lo, hi):,.2f}"


def fmt_date(d: date, rng: random.Random) -> str:
    style = rng.randrange(4)
    if style == 0:
        return d.strftime("%A, %B ") + str(d.day)
    if style == 1:
        return d.strftime("%m/%d/%Y")
    if style == 2:
        return d.strftime("%b ") + str(d.day)
    return d.strftime("%B ") + str(d.day) + d.strftime(", %Y")


def _token(rng: random.Random) -> str:
    return "".join(rng.choice("abcdefghjkmnpqrstuvwxyz23456789") for _ in range(rng.randrange(6, 14)))


def make_ctx(rng: random.Random) -> dict[str, str]:
    """One fictional scenario: who sends what to whom. Every value is a placeholder.
    Both URL flavours are built from the same scheme pool and the same path/prefix vocabulary so that
    URL presence and URL sub-tokens carry no label signal; only the host's relationship to the sender
    (own domain vs. lookalike) differs, which is the cue a classifier should actually learn.
    """
    org = rng.choice(ORGS)
    org_dom = f"{slug(org)}.{rng.choice(TLDS)}"
    vendor = rng.choice(VENDORS)
    vendor_dom = f"{slug(vendor)}.{rng.choice(TLDS)}"
    msp = rng.choice(MSPS)
    msp_dom = f"{slug(msp)}.{rng.choice(TLDS)}"
    names = rng.sample(FIRST, 6)
    lasts = rng.sample(LAST, 6)
    base = date(2026, 1, 5) + timedelta(days=rng.randrange(0, 360))
    d2 = base + timedelta(days=rng.randrange(1, 15))
    hour = rng.choice([8, 9, 10, 11, 13, 14, 15, 16, 21, 22])
    t = f"{hour % 12 or 12}:{rng.choice(['00', '15', '30', '45'])} {'AM' if hour < 12 else 'PM'}"
    if rng.random() < 0.25:
        phish_host = f"{slug(org)}-{rng.choice(['it-support', 'helpdesk', 'payroll', 'secure'])}.{rng.choice(TLDS)}"
    else:
        phish_host = f"{rng.choice(LOOKALIKES)}.{rng.choice(TLDS)}"
    phish_path = rng.choice(PHISH_PATHS)
    own_dom = rng.choice([org_dom, org_dom, vendor_dom, msp_dom])
    r = rng.random()
    if r < 0.15:
        legit_host = f"{slug(org).split('-')[0]}.sharepoint.{rng.choice(TLDS)}"
    elif r < 0.25:
        legit_host = f"{slug(org).split('-')[0]}-my.sharepoint.{rng.choice(TLDS)}"
    elif r < 0.35:
        legit_host = own_dom
    else:
        legit_host = f"{rng.choice(LEGIT_HOST_PREFIXES)}.{own_dom}"
    legit_path = rng.choice(LEGIT_PATHS)
    used = rng.randrange(46, 50) + rng.random()
    return {
        "org": org, "org_dom": org_dom, "org_short": org.split()[0],
        "first": names[0], "last": lasts[0], "email": f"{names[0].lower()}@{org_dom}",
        "ceo_first": names[1], "ceo_last": lasts[1], "ceo_title": rng.choice(OWNER_TITLES),
        "col_first": names[2], "col_last": lasts[2], "col_title": rng.choice(STAFF_TITLES),
        "hr_first": names[3], "hr_last": lasts[3],
        "vendor": vendor, "vendor_dom": vendor_dom, "ven_first": names[4], "ven_last": lasts[4],
        "msp": msp, "msp_dom": msp_dom, "tech_first": names[5], "tech_last": lasts[5],
        "client": rng.choice(CLIENTS), "project": rng.choice(PROJECTS), "file": rng.choice(FILES),
        "amount": money(rng, 380, 48_000), "amount2": money(rng, 90, 2_500),
        "inv": f"{rng.choice(['INV-', 'INV', '#', 'No. ', 'A'])}{rng.randrange(10_000, 99_999)}",
        "po": f"PO-{rng.randrange(1000, 9999)}", "acct4": f"{rng.randrange(0, 9999):04d}",
        "date": fmt_date(base, rng), "date2": fmt_date(d2, rng), "time": t,
        "hours": str(rng.choice([2, 4, 12, 24, 48])), "days": str(rng.choice([3, 5, 7, 10, 14, 30])),
        "ticket": f"#{rng.randrange(10_000, 99_999)}", "location": rng.choice(LOCATIONS),
        "device": rng.choice(DEVICES),
        "ip": f"{rng.choice(['203.0.113', '198.51.100', '192.0.2'])}.{rng.randrange(2, 254)}",
        "code": f"{rng.randrange(0, 999_999):06d}", "gb_used": f"{used:.1f}",
        "gb_total": str(rng.choice([50, 50, 100])), "pct": str(rng.choice([92, 95, 97, 98, 99])),
        "phone": f"(555) 01{rng.randrange(0, 100):02d}", "ext": str(rng.randrange(100, 400)),
        "gift_n": str(rng.choice([4, 5, 6, 8, 10])), "gift_v": rng.choice(["$100", "$200", "$250", "$500"]),
        "room": rng.choice(["the conference room", "Room B", "Teams", "Zoom", "the upstairs office"]),
        "pages": str(rng.randrange(2, 14)),
        "phish_url": f"{rng.choice(SCHEMES)}{phish_host}{phish_path}{_token(rng)}",
        "legit_url": f"{rng.choice(SCHEMES)}{legit_host}{legit_path}"
                     + (_token(rng) if legit_path.endswith(("/", "=")) else ""),
    }


# ----------------------------------------------------------------------------- shared pools (both labels)
GREETINGS = ["Hi {first},", "Hello {first},", "{first},", "Good morning {first},", "Dear {first},", "Hi there,",
             "Hello,", "Dear {first} {last},", "Hey {first} -", "Team,", "Hi all,", "Good afternoon,", ""]
SIGNOFFS = ["Thanks,", "Thank you,", "Best,", "Regards,", "Kind regards,", "Thanks!", "Cheers,",
            "Sent from my iPhone", "Best regards,", "Many thanks,", "Talk soon,", ""]
# Padding drawn by BOTH labels. Deliberately mixes mundane office lines with security-awareness lines
# that carry lure vocabulary (password, verify, wire, gift cards, urgent, expires, suspended, approve) in
# a benign frame, and notice vocabulary (planned maintenance, posting for awareness) in a neutral frame.
SHARED_FILLER = [
    "Also, a reminder that the office closes early on Friday for the staff lunch.",
    "Please let me know if you have any questions.",
    "If this has already been handled, please disregard this message.",
    "I am out part of tomorrow but will be checking email.",
    "Feel free to forward this to anyone on the team who needs it.",
    "We appreciate your business and your patience.",
    "Let me know if another time works better for you.",
    "Apologies for the short notice on this one.",
    "Reply to this email or call {phone} ext. {ext} if you need anything.",
    "Thanks again for getting back to me so quickly last week.",
    "As always, reach out with any concerns.",
    "Hope your week is going well.",
    "I will follow up again on {date2} if I do not hear back.",
    "Copying {col_first} so they are in the loop.",
    "Reminder: IT will never ask for your password, verification codes or gift card numbers by email.",
    "If anything in a message looks off, verify it by calling the sender at a number you already have.",
    "Our bank and remittance details only change in writing on letterhead, confirmed by phone.",
    "Wire and ACH requests over {amount2} need a second approver, however urgent they seem.",
    "Urgent requests still go through the normal ticket queue; nothing here needs you to reply immediately.",
    "Your login and authenticator settings are unchanged unless you hear from {tech_first} directly.",
    "Anything that expires or gets suspended is announced at least {days} days ahead.",
    "The shared OneDrive and DocuSign folders are the only places we keep signed documents.",
    "Invoices are paid on {days}-day terms; a genuine overdue notice always comes with a phone call.",
    "Accounts Payable approves payments on {date2}; please have receipts in by then.",
    "We're keeping the planned maintenance window short, so expect only a brief Teams interruption.",
    "Posting this for awareness; the full notes are on the intranet.",
    "Just a heads up that the parking lot is being resurfaced on {date2}.",
    "The new coffee machine arrives {date2}, so the break room will be closed that morning.",
    "Quarterly numbers go to {ceo_first} by {date2}; send me anything you want included.",
    "The {project} kickoff moved to {date2}; same room, same time.",
    "Please keep your mailbox under quota; archive anything older than a year.",
    "Held messages release automatically once the mail filter finishes its scan.",
    "{col_first} is covering the front desk today, so route walk-ins to them.",
    "Please confirm you received this so I can check it off the list.",
    "The team calendar and the status page are the quickest ways to track what's going on.",
    "The PTO form and the benefits enrollment form are both on the intranet under Human Resources.",
    "Here's a copy of the agenda and the recap from the last meeting for reference.",
    "Appointment and booking changes go through the front desk; reschedule there, not by email.",
    "The estimate and pricing history for the {project} are in the shared folder - it's the usual place.",
    "Our banking and remittance details have not changed; remit to the address on the invoice.",
    "A few items are still open from last month's advisory; the resolved ones are posted already.",
    "Within a few days you should get the final document and the instructions to continue.",
    "Session storage and incoming mail activity were normal; the service health page says resolved.",
    "The IT team will be in {room} on {date} if you'd prefer to do this in person.",
    "This step replaces the paper form previously handled by {hr_first} in HR.",
    "Thanks for keeping your details current; nothing about your identity or codes needs to change.",
    "If a link in any email looks off, don't click it - call the sender and avoid guessing.",
    "Keep it quiet until it's done; the client would rather get the news from us directly.",
    "You can get more information about the proposal from {col_first} when they are available.",
    "The admin center will show progress on the migration; dates may move a little along the way.",
    "Direct deposit dates are posted; the deposit for the {project} is covered in the quote.",
    "Coverage for the front desk starts at {time}; {col_first} confirms the swap via email.",
    "Mailboxes over {gb_total} GB bounce new mail, so please validate your archive settings and process old items.",
    "Credentials for the new system go out from {tech_first} on {date2}; nothing is affected before then.",
    "Elections for the benefits plan are processed by {hr_first}; move anything else to the HR queue.",
    "Deactivation of old accounts happens {days} days after someone leaves, never the same day.",
    "The nightly backup will start at {time}; expect a couple of calls about slow file access until it finishes.",
    "We do not change banks without a phone call first, so if you ever see a monthly statement from a new account, ask.",
]
DISCLAIMERS = [
    "CONFIDENTIALITY NOTICE: This message and any attachments are intended only for the named recipient.",
    "This email was sent to {email} because of your account with {org}.",
    "Please consider the environment before printing this email.",
    "This mailbox is not monitored; please do not reply directly to this message.",
    "If you received this message in error, please notify the sender and delete it.",
]
SUBJECT_PREFIX = ["", "", "", "", "RE: ", "Re: ", "FW: ", "Fwd: ", "[External] ", "ACTION: ", "Reminder: "]


@dataclass
class Category:
    """A template family. `parts` is a list of slots; each slot lists interchangeable sentences (a slot
    may contain "" to be optional). `extras` are optional padding paragraphs; `signers` name signature
    styles. `url` selects which link pool (if any) is woven in; `anchors` is the category-appropriate
    link text used in HTML bodies (plain-text bodies carry the bare URL because the template sentence
    already supplies the prefix)."""

    label: int
    subjects: list[str]
    parts: list[list[str]]
    extras: list[str]
    signers: list[str]
    url: str = "phish"            # "phish" | "legit" | "none"
    html_prob: float = 0.25
    anchors: list[str] = field(default_factory=lambda: list(GENERIC_ANCHORS))


SIGNATURES = {
    "ceo": ["{ceo_first}", "{ceo_first} {ceo_last}\n{ceo_title}, {org}", "{ceo_first} {ceo_last}\n{ceo_title}",
            "{ceo_first}\nSent from my iPhone", "- {ceo_first}"],
    "colleague": ["{col_first}", "{col_first} {col_last}\n{col_title}\n{org}",
                  "{col_first} {col_last} | {col_title} | {phone}"],
    "hr": ["{hr_first} {hr_last}\nHuman Resources, {org}", "{hr_first}\nHR & Payroll", "{org} Human Resources"],
    "vendor": ["{ven_first} {ven_last}\nAccounts Receivable\n{vendor}\n{phone}", "{vendor} Billing Department",
               "{ven_first} {ven_last} | {vendor}", "{ven_first}\n{vendor}"],
    "msp": ["{tech_first} {tech_last}\nService Desk, {msp}\n{phone}", "{msp} Support Team",
            "{tech_first}\n{msp} | support@{msp_dom}", "The {msp} Service Desk"],
    "brand": ["Microsoft 365 Message center", "Microsoft account team", "The Microsoft 365 Team",
              "Microsoft Security", "IT Service Desk", "Microsoft Entra ID"],
    "itdesk": ["IT Help Desk\n{org}", "{org} IT Department", "Mail Administrator", "System Administrator",
               "IT Support Team | {phone}", "The {msp} Service Desk", "Helpdesk"],
    "docs": ["OneDrive", "Shared via OneDrive for Business", "DocuSign Electronic Signature Service",
             "SharePoint Online", "{col_first} {col_last} (via OneDrive)", "Document Services"],
    "client": ["{ven_first} {ven_last}\n{client}", "{ven_first}\nOperations, {client}", "{ven_first} | {client}"],
}

# Optional trailing slot for every positive category: mundane notice vocabulary (planned maintenance,
# posting for awareness, we're, just, by) so those words are not negative-only.
PHISH_MUNDANE = [
    "We're posting this for awareness ahead of the planned maintenance on {date2}.",
    "Just a reminder that this is part of the scheduled {org} security maintenance.",
    "Add the maintenance window to your calendar; the status page will show progress.",
    "Benefits and payroll questions still go to {hr_first} in Human Resources as usual.",
    "Check the status of this request at the help desk if you have any questions.",
    "The IT team will be in {room} on {date} if you'd prefer to do this in person.",
    "This step replaces the paper form previously handled by {hr_first} in HR.",
    "", "", "", "",
]

C: dict[str, Category] = {}

# ------------------------------------------------------------------ positives (label 1)
# Lure sentences are deliberately short and schematic; the realism lives in the varied placeholders,
# not in persuasive copy. Each positive category pairs with a negative category in the same register.
C["m365_signin"] = Category(1, [
    "Your password expires today", "Action required: Microsoft 365 password expiration",
    "Password expiry notice for {email}", "Your Microsoft 365 account will be suspended",
    "Unusual sign-in activity on your account", "Microsoft 365: verification needed",
    "Account review required within {hours} hours", "Sign-in blocked for {email}",
], [
    ["Your Microsoft 365 password for {email} expires today.",
     "Your Office 365 credentials have not been updated under the new {org} security policy.",
     "A sign-in from {location} (IP {ip}) was blocked on {date} at {time}.",
     "Microsoft Entra ID flagged your account for a mandatory review."],
    ["Sign in to keep your current password and avoid interruption: {url}",
     "Confirm your account at the secure page to keep mailbox access: {url}",
     "Re-authenticate now: {url}", "Review the activity and secure your account here: {url}"],
    ["This link expires in {hours} hours.", "Failure to verify will suspend access to Outlook and Teams.",
     "If you do not recognize this activity, act within {hours} hours.",
     "Your usual password will stop working after that; check your spam folder if the code did not arrive.", ""],
    PHISH_MUNDANE,
], [], ["brand", "itdesk"], anchors=["Sign in", "Keep my password", "Review activity", "Verify account"])

C["mfa_prompt"] = Category(1, [
    "Approve your sign-in request", "Your verification code is {code}", "Authenticator re-registration required",
    "New MFA device pending approval", "Confirm your identity to continue",
], [
    ["We sent an approval request to your {device}. Approve it to finish signing in.",
     "Your authenticator must be re-registered after the {org} security update.",
     "A verification code ({code}) was requested for your account from {location}.",
     "Multi-factor authentication for {email} needs to be re-enrolled."],
    ["Re-register your authenticator here: {url}", "Approve or review the request: {url}",
     "Complete MFA setup at: {url}", "Confirm your identity: {url}"],
    ["If you did not request this, approve within {hours} hours to keep access.", "Do not share your code with anyone.", ""],
    PHISH_MUNDANE,
], [], ["brand", "itdesk"], anchors=["Approve request", "Re-register authenticator", "Confirm identity", "Review sign-in"])

C["bec_payment"] = Category(1, [
    "Quick request", "Are you available?", "Urgent - need your help", "Payment to process today",
    "Confidential - wire request", "Can you handle this now?",
], [
    ["Are you at your desk? I need a task handled quickly and I'm heading into {room}.",
     "I need you to process a payment of {amount} to a vendor before end of day. Keep it between us for now.",
     "Can you action a wire transfer today? I'll send the banking details once you confirm you're free.",
     "I'm tied up in meetings - handle this discreetly and let me know when it's done."],
    ["Reply and I'll send the account details.", "Let me know once it's sent.",
     "The beneficiary details are attached; confirm the {amount} is cleared today.",
     "Send me confirmation of the transfer to {email} when complete."],
    ["Sent from my iPhone, apologies for brevity.", "This is time-sensitive.",
     "I'm relying on you to keep this quiet until it posts.", ""],
    ["We're just finishing the planned budget review, so I can't call right now.", "", ""],
], [], ["ceo"], url="none", html_prob=0.0)

C["bec_giftcard"] = Category(1, [
    "Quick favor", "Need these today", "Are you around?", "Client appreciation gifts", "Small errand",
], [
    ["I need you to pick up {gift_n} {gift_v} gift cards for a client thank-you. Can you do it now?",
     "Please grab {gift_n} gift cards ({gift_v} each) for the team - I'll reimburse you.",
     "I'm in back-to-back calls; can you buy gift cards and send me the codes?"],
    ["Scratch off the back and email me photos of the codes.", "Text me once you have them.",
     "Send the card numbers to {email} when done."],
    ["Keep the receipt and I'll sort the expense.", "It's a surprise, so keep it quiet.", ""],
    ["Just posting this to you rather than the whole team.", "", ""],
], [], ["ceo"], url="none", html_prob=0.0)

C["fake_fileshare"] = Category(1, [
    "{col_first} shared \"{file}\" with you", "Document shared via OneDrive", "You have a new shared file",
    "Please review and sign: {file}", "DocuSign: {file} awaiting your signature", "A file was shared with {email}",
], [
    ["{col_first} {col_last} shared the document \"{file}\" with you.",
     "A file (\"{file}\") has been shared with you via OneDrive for Business.",
     "{ven_first} {ven_last} sent you \"{file}\" to review and sign.",
     "You have a DocuSign envelope waiting: \"{file}\"."],
    ["Open the document here: {url}", "Review and sign: {url}", "View \"{file}\": {url}", "Access the shared file: {url}",
     "Open the shared folder: {url}"],
    ["This link is available for {days} days.", "Sign in with your email to view.",
     "You may be asked to verify your account to open it.", ""],
    PHISH_MUNDANE,
], [], ["docs", "hr", "colleague"], anchors=["Open document", "View document", "Review and sign", "Open in OneDrive"])

C["invoice_lure"] = Category(1, [
    "Overdue invoice {inv}", "Final notice: payment required", "Invoice {inv} past due", "Outstanding balance {amount}",
    "Payment reminder - {inv}", "Your account is overdue",
], [
    ["Invoice {inv} for {amount} is now overdue. Please remit payment to avoid service interruption.",
     "Our records show an unpaid balance of {amount} on your account. See the attached invoice.",
     "This is a final notice for overdue invoice {inv}. Immediate payment is required."],
    ["View and pay the invoice here: {url}", "Open the invoice {inv}: {url}", "Settle the balance online: {url}",
     "Download the statement: {url}"],
    ["A late fee applies after {days} days.", "Reference: {inv}.", "Please confirm once paid.",
     "Your monthly statement will show the charge once it clears.", ""],
    PHISH_MUNDANE,
], [], ["vendor", "itdesk"], anchors=["View invoice", "Pay now", "Download statement", "View balance"])

C["helpdesk_quota"] = Category(1, [
    "Mailbox almost full - {pct}%", "Your email storage is full", "Action needed: mailbox quota exceeded",
    "Incoming messages are being held", "Re-validate your mailbox", "Email account deactivation notice",
], [
    ["Your mailbox is at {gb_used} GB of {gb_total} GB ({pct}%). New mail will bounce once it is full.",
     "Your email storage for {email} is full and incoming messages are being held.",
     "Your account is scheduled for deactivation unless it is re-validated under the {org} policy."],
    ["Increase your quota here: {url}", "Re-validate your mailbox: {url}", "Release held messages: {url}",
     "Keep your account active: {url}"],
    ["This request expires in {hours} hours.", "Sign in with your current credentials to continue.", ""],
    PHISH_MUNDANE,
], [], ["itdesk", "brand"], anchors=["Increase quota", "Re-validate mailbox", "Release messages", "Keep account active"])

C["vendor_bankchange"] = Category(1, [
    "Updated remittance details", "Change to our banking information", "New payment instructions - {vendor}",
    "Please update our bank details", "Important: ACH information change",
], [
    ["We have changed banks. Please update our remittance details before your next payment of {amount}.",
     "Effective immediately, {vendor} payments should go to our new account. Updated details attached.",
     "Our ACH information has changed for invoice {inv}; please use the new account ending {acct4}."],
    ["Confirm the updated details here: {url}", "Download the updated banking form: {url}",
     "Review the new remittance instructions: {url}"],
    ["Please confirm receipt so future payments are not delayed.", "Apply this to all open invoices.", ""],
    PHISH_MUNDANE,
], [], ["vendor"], anchors=["Confirm details", "Download form", "View instructions"])

# ------------------------------------------------------------------ negatives (label 0)
# Same businesses, same tone, and every category carries a link to a plausible internal / vendor portal
# ({url} with url="legit"). Benign sentences reuse the lure vocabulary (password, verify, approve,
# overdue, OneDrive, DocuSign, gift cards, wire, expires, suspended, urgent, authenticator, quota,
# held messages, envelope, Entra) so those tokens are not label-predictive.
C["internal_schedule"] = Category(0, [
    "Staff meeting {date}", "Schedule for next week", "Can we move our 1:1?", "Coverage for {date}",
    "Reminder: team lunch {date}", "Shift swap for {date}",
], [
    ["Can we meet in {room} on {date} at {time} to go over the week?",
     "Just confirming our 1:1 on {date} at {time} - let me know if you need to move it.",
     "I'm putting together coverage for {date}; can you take the morning front desk?",
     "Team lunch is on {date}. Please reply with your order by {date2}."],
    ["The agenda doc is on OneDrive: {url}", "Shared calendar, if you want to check conflicts: {url}",
     "I'll book {room}; the room schedule is here: {url}", "No agenda, just a quick sync.", ""],
    ["Also, could you approve my PTO request in the HR portal when you get a sec?",
     "If the booking system asks you to sign in, your usual password works - nothing has changed.",
     "Bring your notes from last week.", "Thanks for being flexible.", ""],
], [], ["colleague"], url="legit", html_prob=0.1, anchors=["Agenda", "Open doc", "Room calendar"])

C["genuine_invoice"] = Category(0, [
    "Invoice {inv} from {vendor}", "Your {vendor} statement", "Monthly invoice - {date}", "Receipt for {inv}",
    "{vendor}: invoice attached", "Statement for {org}",
], [
    ["Please find invoice {inv} for {amount2} attached, covering this month's supplies.",
     "Your statement from {vendor} is ready. The balance of {amount2} is due {date2}.",
     "Thanks for your order. Invoice {inv} is attached; no action needed if already paid.",
     "Here is your monthly invoice. You can pay by the usual bank transfer or check."],
    ["View or pay invoice {inv} in the portal: {url}", "A copy is also in your {vendor} portal: {url}",
     "Download the PDF from your account: {url}", "Payment history and open invoices: {url}"],
    ["This is not overdue - the usual {days}-day terms apply.",
     "Our banking details are unchanged; if you ever get a message saying otherwise, call {phone} to verify before paying.",
     "If you already sent payment, please disregard this.",
     "The portal login uses your existing password; there is nothing to re-verify.", ""],
], [], ["vendor"], url="legit", html_prob=0.25, anchors=["View invoice", "Pay online", "Open portal"])

C["service_health"] = Category(0, [
    "Microsoft 365 service health update", "Scheduled maintenance notice", "Outlook degradation - resolved",
    "Service advisory for {date}", "Planned Teams maintenance",
], [
    ["Microsoft 365 reported a service issue affecting Outlook on the web; it is now resolved. No action is needed.",
     "There is a planned maintenance window on {date} from {time}. Some Teams features may be briefly unavailable.",
     "This is an informational service-health advisory. You do not need to sign in or change anything.",
     "Entra ID sign-in was slow for about {hours} hours this morning; Microsoft has posted the root cause."],
    ["Full status: {url}", "Details are on the status page: {url}", "Message center post: {url}",
     "Track the advisory here: {url}"],
    ["No sign-in, password reset or authenticator re-registration is required.",
     "Mailbox quota and held messages were not affected.",
     "Nobody from IT will ever ask you to approve an MFA request you did not start.",
     "Posting for awareness only.", ""],
], [], ["itdesk", "msp"], url="legit", html_prob=0.25, anchors=["Full status", "Status page", "View advisory"])

C["hr_benefits"] = Category(0, [
    "Open enrollment reminder", "Updated employee handbook", "Payroll calendar for 2026", "Benefits Q&A on {date}",
    "New PTO policy", "Holiday schedule",
], [
    ["Open enrollment closes {date2}. Review your benefits elections and let HR know of any changes.",
     "The updated employee handbook (\"{file}\") is posted on the intranet for your reference.",
     "Here is the 2026 payroll calendar. Direct deposit dates are unchanged.",
     "We're holding a benefits Q&A on {date} at {time} in {room}."],
    ["Sign the handbook acknowledgement in the DocuSign envelope by {date2}: {url}",
     "The payroll calendar is in the shared OneDrive folder: {url}",
     "Enrollment portal (same login as payroll): {url}", "Handbook on the intranet: {url}",
     "Please confirm your elections on the enrollment form by {date2}."],
    ["Enrollment access expires {date2}; late elections are held until next year.",
     "Holiday gift cards are ordered centrally by Accounts Payable - nobody will ask you to buy them.",
     "No action is needed if your current elections are correct.", "Reach HR at {phone} ext. {ext}.",
     "It's for all staff, so please pass it along within the week.", ""],
], [], ["hr"], url="legit", html_prob=0.2, anchors=["Open handbook", "Sign in DocuSign", "Enrollment portal", "View calendar"])

C["project_update"] = Category(0, [
    "{project} - weekly update", "Status: {project}", "Next steps on the {project}", "{project} recap",
    "Quick update on the {project}",
], [
    ["Quick update on the {project}: we finished the first phase and start the next on {date2}.",
     "Here's where the {project} stands. We're on track for the {date2} milestone.",
     "Recap from today's call on the {project} - action items are below.",
     "The {project} is moving along; I'll need sign-off on the timeline by {date2}."],
    ["The updated plan (\"{file}\") is in the shared OneDrive folder: {url}", "Timeline and action items: {url}",
     "The project SharePoint site: {url}", "Ticket history for the {project}: {url}",
     "I'll share the planning document with the client once it's final."],
    ["Nothing urgent - this can wait until our call.",
     "Access to the shared folder was granted today; sign in with your work account.",
     "I'll need your approval on the revised {amount} budget line before we commit.",
     "Please confirm the dates work and I'll lock them in.", "No blockers on our end.", ""],
], [], ["colleague", "msp"], url="legit", html_prob=0.15, anchors=["Updated plan", "Open folder", "View timeline"])

C["vendor_quote"] = Category(0, [
    "Quote for {project}", "Revised estimate - {vendor}", "Pricing you requested", "Proposal for {org}",
    "Estimate {inv}",
], [
    ["Thanks for the opportunity. Our quote for the {project} comes to {amount}, valid for {days} days.",
     "Attached is the revised estimate (\"{file}\") for the {project}.",
     "Here's the pricing you asked about. Happy to adjust the scope if needed.",
     "Our proposal for {org} is attached; the {amount} total includes materials and labor."],
    ["View the quote online: {url}", "Approve or request changes here: {url}",
     "Full estimate (\"{file}\"): {url}", "Our project portfolio, if useful: {url}",
     "It's attached as a PDF; the signed order form can come back the same way."],
    ["The quote expires in {days} days; just let us know if you need longer.",
     "Payment terms are 50% deposit by bank transfer, balance on completion - no wire is needed until you approve.",
     "Please confirm within {days} days so we can hold the pricing.",
     "Call {phone} with any questions.", "No obligation, of course.", ""],
], [], ["vendor", "client"], url="legit", html_prob=0.2, anchors=["View quote", "Open estimate", "Approve quote"])

C["calendar_confirm"] = Category(0, [
    "Confirmed: {date} at {time}", "Appointment reminder", "Your booking with {org}", "See you {date}",
    "Reminder: {date} appointment",
], [
    ["This confirms your appointment with {org} on {date} at {time}.",
     "Reminder: you're booked for {date} at {time}. Reply to reschedule.",
     "Looking forward to seeing you on {date}. Please arrive a few minutes early."],
    ["Reschedule or cancel online: {url}", "Manage your booking: {url}", "Add it to your calendar: {url}",
     "Please confirm by replying YES, or call us to reschedule.", ""],
    ["Your confirmation code is {code} - keep it handy at check-in.",
     "Your {device} may ask you to approve the calendar invite.",
     "No action needed - just a reminder.", "We'll send a text the day before.", ""],
], [], ["colleague"], url="legit", html_prob=0.2, anchors=["Reschedule", "Manage booking", "Add to calendar"])

C["msp_maintenance"] = Category(0, [
    "Scheduled maintenance window {date}", "After-hours patching {date}", "Firewall update tonight",
    "Backup verification complete", "Planned reboot {date}",
], [
    ["{msp} will perform scheduled maintenance on {date} from {time}. Brief downtime is possible after hours.",
     "We're applying security patches to your server on {date}. No action is needed on your end.",
     "Tonight's firewall update may interrupt the VPN for a few minutes.",
     "This month's backup verification for {org} completed successfully; copy on file."],
    ["Ticket {ticket}: {url}", "Live status during the window: {url}",
     "Maintenance notes and affected systems: {url}", "Ticket {ticket} tracks this work.", ""],
    ["Please verify your laptop rebooted and that you can sign in as usual afterwards.",
     "If a login prompt appears after the reboot, use your normal password; we will never ask for it by email.",
     "Authenticator and MFA settings are unchanged.", "Reach the service desk at {phone}.", ""],
], [], ["msp", "itdesk"], url="legit", html_prob=0.2, anchors=["View ticket", "Status page", "Open ticket"])

POSITIVE = [k for k, v in C.items() if v.label == 1]
NEGATIVE = [k for k, v in C.items() if v.label == 0]


def label3_for_category(cat_name: str) -> int:
    """Three-way label of a template category: "0:*" -> 0 (legit), "1:*" -> 1 (phish).
    A positive category not listed in PHISH_CATEGORIES raises: a future spam-style template must be
    classified deliberately (label3 2) instead of inheriting "phishing" from the binary label.
    """
    binary = C[cat_name].label
    if binary == 0:
        return 0
    if cat_name not in PHISH_CATEGORIES:
        raise ValueError(f"positive category {cat_name!r} is not in PHISH_CATEGORIES; decide phishing vs spam")
    return 1


# ----------------------------------------------------------------------------- rendering
def _fill(s: str, ctx: dict[str, str]) -> str:
    return s.format(**ctx)


def _recase_subject(sub: str, rng: random.Random) -> str:
    r = rng.random()
    if r < 0.12:
        return sub.upper()
    if r < 0.24:
        return sub.lower()
    return sub


def _anchor(url: str, rng: random.Random, html: bool, anchors: list[str]) -> str:
    """HTML bodies get category-appropriate link text; plain-text bodies get the bare URL because the
    template sentence already carries the call-to-action prefix."""
    if html:
        return f'<a href="{url}">{rng.choice(anchors)}</a>'
    return url


def render(cat_name: str, ctx: dict[str, str], rng: random.Random) -> tuple[str, str]:
    cat = C[cat_name]
    html = rng.random() < cat.html_prob
    subject = _recase_subject(_fill(rng.choice(SUBJECT_PREFIX) + rng.choice(cat.subjects), ctx), rng)

    lines: list[str] = []
    greet = _fill(rng.choice(GREETINGS), ctx)
    if greet:
        lines.append(greet)

    url = ctx["phish_url"] if cat.url == "phish" else ctx["legit_url"]
    body_sents: list[str] = []
    for slot in cat.parts:
        choice = rng.choice(slot)
        if not choice:
            continue
        # weave the link in before slot filling so the anchor text (which may contain no braces)
        # survives str.format; {url} is the only marker not present in ctx.
        if "{url}" in choice:
            if cat.url == "none":
                continue
            choice = choice.replace("{url}", _anchor(url, rng, html, cat.anchors))
        body_sents.append(_fill(choice, ctx))
    lines.append(" ".join(body_sents))

    signoff = _fill(rng.choice(SIGNOFFS), ctx)
    signature = _fill(rng.choice(SIGNATURES[rng.choice(cat.signers)]), ctx)
    disclaimer = _fill(rng.choice(DISCLAIMERS), ctx) if rng.random() < 0.3 else ""
    tail = [p for p in (signoff, signature, disclaimer) if p]

    # Per-row length target: right-skewed triangular over [MIN_WORDS, TARGET_MAX_WORDS] with the mode at
    # MIN_WORDS, so most rows stay short like real mail but a tail reaches ~200 words. The draw is
    # label-independent, and the padding comes from pools shared by both labels, so length and filler
    # vocabulary carry no label signal. The whole prospective message is counted so every body clears
    # MIN_WORDS.
    target = int(rng.triangular(MIN_WORDS, TARGET_MAX_WORDS, MIN_WORDS))
    pad_pool = list(dict.fromkeys(cat.extras + SHARED_FILLER))
    rng.shuffle(pad_pool)
    i = 0
    tail_words = sum(len(p.split()) for p in tail)
    while len(" ".join(lines).split()) + tail_words < target and i < len(pad_pool):
        lines.append(_fill(pad_pool[i], ctx))
        i += 1
    lines.extend(tail)

    body = "\n\n".join(p for p in lines if p).strip()
    if html:
        body = "<html><body>" + body.replace("\n\n", " <br><br> ") + "</body></html>"
    # trim to the upper word bound without cutting a word
    words = body.split()
    if len(words) > MAX_WORDS:
        body = " ".join(words[:MAX_WORDS])
    return subject, body


# ----------------------------------------------------------------------------- build
def _token_purity(df: pd.DataFrame) -> dict[str, Any]:
    """Document-frequency label purity over the WHOLE vocabulary.
    For each lower-cased word token, count how many positive / negative documents contain it (presence,
    not multiplicity). Purity = max(pos, neg) / support. A high-support, high-purity token is a shortcut
    the classifier could exploit. The guard flags every token with support >= PURITY_SUPPORT and purity
    > PURITY_MAX regardless of rank; the top-N table is kept only for display, because a shortcut token
    confined to one category (support ~ n/16) never reaches the top of the frequency list.
    """
    pos_df: Counter[str] = Counter()
    neg_df: Counter[str] = Counter()
    for text, label in zip(df["text"], df["label"]):
        toks = set(TOKEN_RE.findall(text.lower()))
        (pos_df if label == 1 else neg_df).update(toks)
    support = pos_df + neg_df

    def row(tok: str) -> dict[str, Any]:
        p, n = pos_df[tok], neg_df[tok]
        return {"token": tok, "support": support[tok], "pos": p, "neg": n,
                "purity": round(max(p, n) / support[tok], 3), "majority": 1 if p >= n else 0}

    top = [row(t) for t, _ in support.most_common(PURITY_TOP_N)]
    eligible = [t for t, s in support.items() if s >= PURITY_SUPPORT]
    flagged = sorted((row(t) for t in eligible if max(pos_df[t], neg_df[t]) / support[t] > PURITY_MAX),
                     key=lambda r: (-r["support"], r["token"]))
    return {"top": top, "flagged": flagged, "n_tokens": len(support), "n_eligible": len(eligible)}


def build(n: int, seed: int) -> tuple[pd.DataFrame, dict[str, Any]]:
    rng = random.Random(seed)
    half = n // 2
    plan: list[str] = []
    for i in range(half):
        plan.append(POSITIVE[i % len(POSITIVE)])
    for i in range(n - half):
        plan.append(NEGATIVE[i % len(NEGATIVE)])
    rng.shuffle(plan)

    seen: set[str] = set()
    records: list[dict[str, Any]] = []
    attempts = 0
    idx = 0
    while idx < len(plan) and attempts < n * 60:
        attempts += 1
        cat_name = plan[idx]
        ctx = make_ctx(rng)
        subject, body = render(cat_name, ctx, rng)
        text = f"{subject}\n\n{body}"
        key = common.normalize_text(text)
        if not key or key in seen:
            continue
        seen.add(key)
        records.append({
            "doc_id": f"synthetic:{idx:06d}", "text": text, "subject": subject, "body": body,
            "label": C[cat_name].label, "source": "synthetic", "cluster_id": -1, "category": cat_name,
        })
        idx += 1
    if idx < len(plan):
        LOG.warning("only generated %d/%d unique rows before attempt cap", idx, len(plan))

    # label3 / label_set are derived after generation so the RNG stream (and therefore every existing
    # column) is identical to the binary-era build for the same seed.
    for rec in records:
        rec["label3"] = label3_for_category(rec["category"])
        rec["label_set"] = LABEL_SET_OF[rec["label3"]]
    df = pd.DataFrame.from_records(records, columns=COLUMNS)
    df["label3"] = df["label3"].astype("int64")
    assert ((df["label3"] != 0) == (df["label"] == 1)).all(), "label3 / binary label disagree"
    return df, _token_purity(df)


def _length_stats(df: pd.DataFrame) -> dict[str, Any]:
    wl = df["body"].str.split().map(len)
    out: dict[str, Any] = {"min": int(wl.min()), "max": int(wl.max()), "mean": round(float(wl.mean()), 1),
                           "median": int(wl.median())}
    for label in (1, 0):
        w = wl[df["label"] == label]
        out[f"label_{label}"] = {"min": int(w.min()), "max": int(w.max()), "mean": round(float(w.mean()), 1),
                                 "p10": int(w.quantile(0.1)), "p90": int(w.quantile(0.9))}
    return out


def _url_stats(df: pd.DataFrame) -> dict[str, Any]:
    """Presence of a URL / HTML anchor / HTML wrapper per label; these must be close to balanced or the
    supplement teaches 'link => phishing'."""
    out: dict[str, Any] = {}
    for key, pat in (("url", "://"), ("href", "<a href"), ("html", "<html>")):
        hit = df["text"].str.contains(pat, regex=False)
        out[key] = {"pos": int(hit[df["label"] == 1].sum()), "neg": int(hit[df["label"] == 0].sum())}
    return out


def _write_parquet(df: pd.DataFrame, out: Path) -> None:
    table = pa.Table.from_pandas(df, preserve_index=False)
    md = {**(table.schema.metadata or {}), **{k.encode(): v.encode() for k, v in PARQUET_META.items()}}
    table = table.replace_schema_metadata(md)
    out.parent.mkdir(parents=True, exist_ok=True)
    pq.write_table(table, out)


def cmd_build(args: argparse.Namespace, cfg: dict) -> None:
    cap = int(cfg["data"]["synthetic"]["max_rows"])
    n = args.n if args.n is not None else cap
    if n > cap:
        LOG.warning("requested n=%d exceeds config max_rows=%d; clamping", n, cap)
        n = cap
    out = Path(args.out) if args.out else Path(cfg["paths"]["data_processed"]) / "synthetic.parquet"
    if not out.is_absolute():
        out = common.project_root() / out
    seed = args.seed if args.seed is not None else int(cfg["seed"])

    df, purity = build(n, seed)
    _write_parquet(df, out)

    dup_rate = 1.0 - df["text"].map(common.normalize_text).nunique() / len(df)
    by_cat = df.groupby(["label", "category"]).size().reset_index(name="n")
    counts = {
        "rows": len(df), "label_1": int((df["label"] == 1).sum()), "label_0": int((df["label"] == 0).sum()),
        "per_category": {f"{int(r.label)}:{r.category}": int(r.n) for r in by_cat.itertuples()},
        # Three-way counts (all three classes listed so a reader sees spam = 0 explicitly).
        "label3": {LABEL3_NAMES[k]: int((df["label3"] == k).sum()) for k in sorted(LABEL3_NAMES)},
        "label_set": {v: int((df["label_set"] == v).sum()) for v in ("legit", "phish", "spam", "unwanted")},
        "per_category_label3": {f"{int(r.label)}:{r.category}": label3_for_category(r.category)
                                for r in by_cat.itertuples()},
    }
    url_stats = _url_stats(df)
    report = {
        "generator": "make_synthetic.py", "note": PARQUET_META["note"], "seed": seed, "requested_n": n,
        "output": str(out), "duplicate_rate": round(float(dup_rate), 6), "counts": counts,
        "length_words": _length_stats(df), "url_presence": url_stats,
        "purity_support_threshold": PURITY_SUPPORT, "purity_warn_above": PURITY_MAX,
        "vocab_tokens": purity["n_tokens"], "tokens_with_support_ge_threshold": purity["n_eligible"],
        "token_purity_top30": purity["top"],
        "token_purity_flagged_count": len(purity["flagged"]), "token_purity_flagged": purity["flagged"],
    }
    report_path = out.with_name(out.stem + "_report.json")
    common.write_json(report, report_path)

    LOG.info("wrote %d rows (%d pos / %d neg) -> %s", counts["rows"], counts["label_1"], counts["label_0"], out)
    LOG.info("three-way: %s (label_set %s)", counts["label3"], counts["label_set"])
    LOG.info("duplicate rate (normalize_text): %.6f", dup_rate)
    LOG.info("length words: %s", report["length_words"])
    LOG.info("url presence (pos/neg): %s", url_stats)
    LOG.info("report -> %s", report_path)

    LOG.info("per-token label purity (top %d by support; support = #docs containing the token):", len(purity["top"]))
    LOG.info("%-16s%8s%7s%7s%8s", "token", "support", "pos", "neg", "purity")
    for r in purity["top"]:
        LOG.info("%-16s%8d%7d%7d%8.3f", r["token"], r["support"], r["pos"], r["neg"], r["purity"])
    flagged = purity["flagged"]
    LOG.info("purity scan over full vocabulary: %d tokens, %d with support>=%d",
             purity["n_tokens"], purity["n_eligible"], PURITY_SUPPORT)
    if flagged:
        LOG.warning("%d token(s) with support>=%d are >%.0f%% one class (possible shortcut):",
                    len(flagged), PURITY_SUPPORT, PURITY_MAX * 100)
        for r in flagged:
            LOG.warning("  %-16s support=%4d pos=%4d neg=%4d purity=%.3f majority=%d",
                        r["token"], r["support"], r["pos"], r["neg"], r["purity"], r["majority"])
    else:
        LOG.info("no token with support>=%d exceeds %.0f%% one class.", PURITY_SUPPORT, PURITY_MAX * 100)


def cmd_show(args: argparse.Namespace, cfg: dict) -> None:
    path = Path(args.path) if args.path else Path(cfg["paths"]["data_processed"]) / "synthetic.parquet"
    if not path.is_absolute():
        path = common.project_root() / path
    df = pd.read_parquet(path)
    for label in (1, 0):
        sub = df[df["label"] == label].head(args.k)
        print(f"\n{'=' * 78}\nlabel={label} ({'phishing/spam' if label else 'legitimate'}) - {len(sub)} samples")
        for row in sub.itertuples():
            l3 = getattr(row, "label3", None)
            three = f"  label3={l3} ({LABEL3_NAMES.get(l3, '?')}) label_set={getattr(row, 'label_set', '?')}" \
                if l3 is not None else "  label3=<missing: rebuild>"
            print(f"\n--- {row.doc_id}  [{row.category}]  source={row.source} cluster_id={row.cluster_id}{three}")
            print(f"SUBJECT: {row.subject}")
            print(row.body)


def main(argv: list[str] | None = None) -> None:
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--config", default="config.yaml")
    sub = ap.add_subparsers(dest="cmd", required=True)
    b = sub.add_parser("build", help="write the synthetic parquet + report json")
    b.add_argument("--n", type=int, default=None, help="rows to generate (default: config data.synthetic.max_rows)")
    b.add_argument("--out", default=None, help="output parquet path (default: data/processed/synthetic.parquet)")
    b.add_argument("--seed", type=int, default=None, help="override config seed")
    s = sub.add_parser("show", help="print k sample rows per class")
    s.add_argument("--k", type=int, default=6)
    s.add_argument("--path", default=None, help="parquet to read (default: data/processed/synthetic.parquet)")
    args = ap.parse_args(argv)

    cfg = common.load_config(args.config)
    global LOG
    LOG = common.setup_logging("make_synthetic", cfg["paths"]["logs"])
    {"build": cmd_build, "show": cmd_show}[args.cmd](args, cfg)


if __name__ == "__main__":
    main()