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

"""Maintainer script — equalizes every technique's `utterances` count
(per locale) to the corpus's own current maximum for that locale, never a
fixed number picked in the abstract. Preserves every existing utterance,
synonym, and comment verbatim; only ever *adds*, never rewrites or removes.

**Why "equalize to the current max", not "pad everyone to 20"** — this
script's own history: three earlier attempts forced every technique up to
a flat 20 `utterances`/locale (12-17 new ones per technique on average).
All three measurably *failed*
`test/recipe-matching/tech-step-eval.test.ts`'s F1 >= 0.8 regression gate
(0.7999 -> 0.791 -> 0.744, each attempt worse than the last), regardless of
whether the added content was mostly generic modal-frame padding ("il
faut ...") or mostly synonym substitution. The common factor across all
three wasn't *how* the filler was generated, it was *how much*: this
corpus's real per-technique max was only 7 (fr) / 5 (en) before any of
this — forcing every technique up to 20 meant most of them tripled or
quadrupled in size on synthetic content alone, which measurably hurt
inter-class separability more than it helped. Equalizing to the corpus's
*own* current max instead means at most a few new utterances per
technique (most need 1-4), which is a small enough addition to plausibly
preserve the F1 gate while still satisfying "same amount of signal per
class" (the actual goal — consistent detection quality across techniques,
not a specific round number).

**Generation strategy** — synonym substitution first (see
`_synonym_variants`): for every existing utterance whose leading phrase
exactly matches one of the technique's own `synonyms`, swap in every
*other* synonym from the same list (e.g. `melt`'s "faire fondre le
beurre" -> "liquéfier le beurre") — genuinely technique-distinguishing
vocabulary, not filler shared across every class. A technique whose
`synonyms` only ever appear *mid-sentence* (the "cut style" techniques —
`julienne`, `brunoise`, `mirepoix`, `paysanne`... — e.g. "couper les
carottes en julienne" doesn't *start* with any of `julienne`'s own
synonyms) has no leading-phrase match to substitute, so a small modal-frame
fallback (`_FR_FRAMES`/`_EN_FRAMES`, 2 per locale — much smaller than the
12/10 used in the failed 20-target attempts) closes the remainder. Safe at
this scale specifically *because* the gap being closed is small (equalizing
to the corpus's own current max, 1-4 utterances short per technique, not
13-17) — see this module's own doc comment above for why volume, not
generation method, was the real problem in every failed attempt.

Run from `services/tech-step-intent-service/` (this directory):
`./.venv/Scripts/python.exe augment_utterances.py`. Rewrites
`training_data.py` in place by textual splicing (AST only to *locate* each
`utterances=[...]` list's line range — never to regenerate the file). Safe
to re-run: a technique already at the current per-locale max is left
untouched, and the max itself is recomputed from the file's *current*
state each time (so re-running after a manual edit re-equalizes against
whatever the new max is, not a stale one).
"""

import ast
import sys

SRC_PATH = "intent_service/training_data.py"

# Minimal fallback pool — only ever used for the small remainder synonym
# substitution can't reach (see this module's own doc comment for why 2,
# not the 12/10 tried in earlier, failed attempts).
_FR_FRAMES = ["il faut {u}", "veillez à {u}"]
_EN_FRAMES = ["make sure to {u}", "remember to {u}"]


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


_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", "secure", "mark", "butter", "crush", "julienne", "reheat", "smother",
    "build", "scoop", "plunge", "increase", "pass", "collect", "have", "salt", "soak",
}
_EN_ADVERB_SKIP = {
    "coarsely", "roughly", "finely", "quickly", "lightly", "briefly", "gently", "carefully",
    "gradually", "very", "thoroughly", "evenly", "generously", "slowly", "thinly", "deep", "blind",
    "dry",
}


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 _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 _synonym_variants(existing: list[str], synonyms: list[str], locale: 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.

    Both the matched *and* the replacement synonym must independently pass
    `_is_fr_infinitive_led`/`_is_en_imperative_led` — a technique's
    `synonyms` list mixes genuine verb forms ("mijoter", "frémir") with
    noun/adjective phrases used the same way a keyword-matcher needs them
    but never as a sentence's own leading verb ("à petit feu", "gros
    bouillons", "huile de friture") — without this check, swapping the
    verb "frémir" for the noun phrase "à petit feu" inside "laisser
    frémir..." produces a syntactically broken sentence ("à petit feu
    ..."), not just a stylistically different one. Filtering the
    replacement pool to the same grammatical shape as the ones this
    function already accepts as *sources* keeps every substitution a
    like-for-like swap."""
    if len(synonyms) < 2:
        return []
    is_led = _is_fr_infinitive_led if locale == "fr" else _is_en_imperative_led
    seen = set(existing)
    sorted_synonyms = sorted({syn for syn in synonyms if is_led(syn)}, key=len, reverse=True)
    if len(sorted_synonyms) < 2:
        return []
    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 top_up(existing: list[str], synonyms: list[str], target: int, locale: str) -> list[str]:
    if len(existing) >= target:
        return []
    needed = target - len(existing)
    pool = _synonym_variants(existing, synonyms, locale)
    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
        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)

    # First pass: collect every entry's current per-locale utterance/synonym
    # lists and find each locale's own current max — the equalization
    # target, not a number picked separately from the corpus itself.
    parsed: list[tuple[str, str, ast.List, list[str], list[str]]] = []
    targets = {"fr": 0, "en": 0}
    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 []
            )
            targets[locale] = max(targets[locale], len(existing))
            parsed.append((uid, locale, utterances_list_node, existing, synonyms))

    print(f"Equalizing to the corpus's own current max — fr: {targets['fr']}, en: {targets['en']}")

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

    for uid, locale, utterances_list_node, existing, synonyms in parsed:
        target = targets[locale]
        new_ones = top_up(existing, synonyms, target, 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 their locale's target — not")
        print("enough synonym variety to reach full equalization:")
        for uid, locale, count in shortfalls:
            print(f"  {uid} ({locale}): {count}/{targets[locale]}")
    else:
        print("Every technique now has exactly the same utterance count as every other, per locale.")


if __name__ == "__main__":
    main()