batchCooking/services/tech-step-intent-service/augment_utterances.py

"""Maintainer script — tops up every technique's `utterances` (both locales)
to a minimum of 20 each, preserving all existing utterances/synonyms/comments
verbatim. Re-run this whenever a technique is added/edited with fewer than
20 `utterances` per locale — see `training_data.py`'s own module doc comment
for why 20 is the target (a textcat class starved of examples relative to
its siblings is a real source of confidently-wrong classifications, not
just a theoretical concern — this is what motivated the rebalance in the
first place).

**Generation strategy, in priority order** — this matters, see the
regression this script's own history records:

1. **Synonym substitution** (`_synonym_variants`) — for every existing
   utterance whose leading phrase exactly matches one of the technique's
   own `synonyms` (e.g. `melt`'s "faire fondre le beurre" starts with the
   synonym "faire fondre"), swap in every *other* synonym from the same
   list ("liquéfier le beurre", "faire chauffer le beurre", ...). This is
   the primary source precisely because it injects genuinely
   technique-*distinguishing* vocabulary (the corpus's own hand-picked
   synonym list) rather than filler shared across every class.
2. **Modal-frame wrapping** (`_frame_variants`) — only used to fill
   whatever's still missing after (1) is exhausted, and deliberately kept
   to a *small* frame pool (3 per locale, not the dozen tried in an
   earlier attempt at this script). A first version of this script relied
   on frame-wrapping as the *primary* mechanism with 12/10 frames per
   locale: it reached 20 utterances everywhere, but measurably **hurt**
   `test/recipe-matching/tech-step-eval.test.ts`'s aggregate F1 (0.80 ->
   0.79, confirmed twice in CI, once even after doubling
   `_TRAINING_ITERATIONS`) — every one of the 74 classes ended up sharing
   the same handful of high-frequency connector words ("il", "faut",
   "de", "à", "veillez"...), which a bag-of-words classifier reads as
   *reduced* inter-class separability, not neutral padding. Frame-wrapping
   is grammatically safe but structurally low-value; kept only as a
   fallback for techniques whose synonym list is too short to reach 20 on
   its own (e.g. `julienne`, 4 synonyms).

Declarative/result-state utterances ("le beurre doit être liquide") are
never used as a source for either strategy (would be ungrammatical once
wrapped/substituted) — `is_fr_infinitive_led`/`is_en_imperative_led` decide
which existing utterances are safe sources for (2); (1) has its own,
stricter "starts with a known synonym" check that already excludes them.

Run from `services/tech-step-intent-service/` (this directory):
`./.venv/Scripts/python.exe augment_utterances.py` (Windows) or
`.venv/bin/python augment_utterances.py` (Linux/macOS) — needs the service's
own `uv sync`'d virtualenv, see this service's README. Rewrites
`training_data.py` in place by textual splicing (AST only to *locate* each
`utterances=[...]` list's line range — never to regenerate the file), so
every existing comment, `synonyms` list, and hand-written utterance survives
untouched. A technique already at/above 20 for a locale is left untouched —
re-running this script is always safe, never re-pads an already-balanced
entry (see `top_up`).
"""

import ast
import sys

SRC_PATH = "intent_service/training_data.py"

# Small fallback frame pool — see this module's own doc comment for why it's
# deliberately short (3 per locale, not a dozen) and only ever a fallback
# behind synonym substitution.
FR_FRAMES = [
    "il faut {u}",
    "veillez à {u}",
    "pensez à {u}",
    "n'oubliez pas de {u}",
    "assurez-vous de {u}",
]
EN_FRAMES = [
    "make sure to {u}",
    "remember to {u}",
    "be sure to {u}",
    "don't forget to {u}",
    "take care to {u}",
]

EN_VERB_WHITELIST = {
    "make", "add", "pour", "mix", "stir", "cut", "place", "cover", "remove", "heat", "let",
    "keep", "turn", "cook", "bake", "roast", "grill", "fry", "boil", "simmer", "whisk", "fold",
    "chop", "mince", "peel", "drain", "season", "rest", "plate", "coat", "melt", "sauté", "saute",
    "braise", "blanch", "marinate", "brown", "glaze", "thicken", "reduce", "dilute", "loosen",
    "moisten", "sift", "toast", "zest", "scald", "pod", "shell", "hollow", "shock", "emulsify",
    "decant", "dust", "sweat", "rub", "punch", "confit", "caramelize", "score", "line", "clarify",
    "stew", "dice", "fillet", "proof", "poach", "pasteurize", "sterilize", "can", "preserve",
    "tie", "truss", "baste", "spoon", "brush", "whip", "beat", "work", "sear", "flatten", "press",
    "knead", "run", "cool", "warm", "combine", "blend", "arrange", "present", "sprinkle", "strain",
    "separate", "bring", "grate", "continue", "deglaze", "scrape", "char", "break", "slice", "set",
    "adjust", "switch", "sterilize", "secure", "mark", "butter", "crush", "julienne", "reheat",
    "smother", "build", "scoop", "plunge", "increase", "pass", "collect", "have", "adjust",
    "dry-toast", "dry-roast", "heat-treat", "pre-bake", "salt", "soak",
}
EN_ADVERB_SKIP = {
    "coarsely", "roughly", "finely", "quickly", "lightly", "briefly", "gently", "carefully",
    "gradually", "very", "thoroughly", "evenly", "generously", "slowly", "thinly", "deep", "blind",
    "dry",
}

_TARGET = 20


def is_fr_infinitive_led(u: str) -> bool:
    first = u.split(" ", 1)[0].lower()
    return first.endswith(("er", "ir", "re")) and len(first) > 2


def is_en_imperative_led(u: str) -> bool:
    words = u.lower().replace(",", "").split()
    if not words:
        return False
    first = words[0]
    if first in EN_VERB_WHITELIST:
        return True
    if first in EN_ADVERB_SKIP and len(words) > 1:
        return words[1] in EN_VERB_WHITELIST
    return False


def _synonym_variants(existing: list[str], synonyms: list[str]) -> list[str]:
    """Substitutes every *other* synonym in place of whichever synonym an
    existing utterance's leading phrase exactly matches — see this module's
    own doc comment for why this is the primary generation strategy."""
    if len(synonyms) < 2:
        return []
    seen = set(existing)
    sorted_synonyms = sorted(set(synonyms), key=len, reverse=True)
    out: list[str] = []
    for u in existing:
        lower_u = u.lower()
        matched = next(
            (
                syn
                for syn in sorted_synonyms
                if lower_u == syn.lower() or lower_u.startswith(f"{syn.lower()} ")
            ),
            None,
        )
        if matched is None:
            continue
        rest = u[len(matched) :]
        for syn in sorted_synonyms:
            if syn == matched:
                continue
            candidate = f"{syn}{rest}"
            if candidate in seen:
                continue
            seen.add(candidate)
            out.append(candidate)
    return out


def _frame_variants(existing: list[str], frames: list[str], is_led) -> list[str]:
    sources = [u for u in existing if is_led(u)]
    if not sources:
        return []
    seen = set(existing)
    out: list[str] = []
    for frame in frames:
        for u in sources:
            candidate = frame.format(u=u)
            if candidate in seen:
                continue
            seen.add(candidate)
            out.append(candidate)
    return out


def top_up(existing: list[str], synonyms: list[str], locale: str) -> list[str]:
    if len(existing) >= _TARGET:
        return []
    needed = _TARGET - len(existing)
    pool = _synonym_variants(existing, synonyms)
    if len(pool) < needed:
        frames = FR_FRAMES if locale == "fr" else EN_FRAMES
        is_led = is_fr_infinitive_led if locale == "fr" else is_en_imperative_led
        # Frame variants must also dedupe against the synonym-substitution
        # pool already chosen, not just `existing` — otherwise the two
        # sources could independently produce the same string.
        already = set(existing) | set(pool)
        for candidate in _frame_variants(existing, frames, is_led):
            if candidate in already:
                continue
            pool.append(candidate)
            already.add(candidate)
    return pool[:needed]


def main() -> None:
    with open(SRC_PATH, encoding="utf-8") as f:
        source = f.read()
    tree = ast.parse(source)
    lines = source.splitlines(keepends=True)

    module_body = tree.body
    training_data_list = None
    for node in module_body:
        if isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name):
            if node.target.id == "TECH_STEP_TRAINING_DATA":
                training_data_list = node.value
                break
    if training_data_list is None or not isinstance(training_data_list, ast.List):
        print("Could not locate TECH_STEP_TRAINING_DATA list", file=sys.stderr)
        sys.exit(1)

    insertions: list[tuple[int, str, list[str]]] = []
    total_added = 0
    shortfalls: list[tuple[str, str, int]] = []

    for entry_call in training_data_list.elts:
        assert isinstance(entry_call, ast.Call)
        uid = None
        for kw in entry_call.keywords:
            if kw.arg == "uid":
                assert isinstance(kw.value, ast.Constant)
                uid = kw.value.value
        for kw in entry_call.keywords:
            if kw.arg not in ("fr", "en"):
                continue
            locale = kw.arg
            locale_call = kw.value
            assert isinstance(locale_call, ast.Call)
            utterances_list_node = None
            synonyms_list_node = None
            for inner_kw in locale_call.keywords:
                if inner_kw.arg == "utterances":
                    utterances_list_node = inner_kw.value
                elif inner_kw.arg == "synonyms":
                    synonyms_list_node = inner_kw.value
            if utterances_list_node is None:
                continue
            assert isinstance(utterances_list_node, ast.List)
            existing = [
                elt.value for elt in utterances_list_node.elts if isinstance(elt, ast.Constant)
            ]
            synonyms = (
                [elt.value for elt in synonyms_list_node.elts if isinstance(elt, ast.Constant)]
                if isinstance(synonyms_list_node, ast.List)
                else []
            )
            new_ones = top_up(existing, synonyms, locale)
            final_count = len(existing) + len(new_ones)
            if final_count < _TARGET:
                shortfalls.append((uid, locale, final_count))
            if not new_ones:
                continue
            last_elt = utterances_list_node.elts[-1]
            insert_after_line = last_elt.end_lineno - 1
            indent = lines[insert_after_line][
                : len(lines[insert_after_line]) - len(lines[insert_after_line].lstrip())
            ]
            new_lines = [f'{indent}"{s}",\n' for s in new_ones]
            insertions.append((insert_after_line, uid, new_lines))
            total_added += len(new_ones)

    insertions.sort(key=lambda t: t[0], reverse=True)
    for line_idx, uid, new_lines in insertions:
        lines[line_idx + 1 : line_idx + 1] = new_lines

    with open(SRC_PATH, "w", encoding="utf-8", newline="\n") as f:
        f.writelines(lines)

    print(f"Added {total_added} new utterances across {len(insertions)} (technique, locale) pairs.")
    if shortfalls:
        print(f"{len(shortfalls)} (uid, locale) pair(s) still below {_TARGET} — not enough synonym")
        print("variety to reach the target without falling back to more generic frames:")
        for uid, locale, count in shortfalls:
            print(f"  {uid} ({locale}): {count}")


if __name__ == "__main__":
    main()