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).

Generates new utterances by wrapping each existing *infinitive-led*
utterance (a bare command clause, e.g. "faire fondre le beurre") in a small
set of natural modal frames ("il faut ...", "veillez à ...", "make sure to
...") — grammatically valid, genuinely varied surface forms that still carry
the technique's own distinguishing vocabulary, not generic boilerplate.
Declarative/result-state utterances ("le beurre doit être liquide") are
never wrapped this way (would be ungrammatical) — `is_fr_infinitive_led`/
`is_en_imperative_led` decide which existing utterances are safe sources.
Frames are lowercase/unpunctuated, matching this corpus' own style exactly
(see `FR_FRAMES`/`EN_FRAMES`'s own comment for why that's not just
cosmetic). 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`).

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.
"""

import ast
import sys

SRC_PATH = "intent_service/training_data.py"

# Lowercase, no trailing period — matches this corpus' existing style
# exactly (every hand-written utterance so far is lowercase/unpunctuated).
# Not just cosmetic: `spacy.TextCatBOW.v3` hashes on token form, and mixing
# "Il"/"il" as if they were different tokens would needlessly fragment the
# bag-of-words signal for what should read as the exact same sentence to the
# classifier.
FR_FRAMES = [
    "il faut {u}",
    "veillez à {u}",
    "pensez à {u}",
    "n'oubliez pas de {u}",
    "la recette demande de {u}",
    "cette étape consiste à {u}",
    "il est important de {u}",
    "assurez-vous de {u}",
    "prenez soin de {u}",
    "commencez par {u}",
    "on vous demande de {u}",
    "il convient de {u}",
]

EN_FRAMES = [
    "make sure to {u}",
    "remember to {u}",
    "be sure to {u}",
    "take care to {u}",
    "you'll need to {u}",
    "don't forget to {u}",
    "it's important to {u}",
    "go ahead and {u}",
    "now {u}",
    "the recipe calls for you to {u}",
]

# Bare English cooking verbs (imperative == infinitive minus "to") — an
# utterance whose first word (or, for an adverb-led opener, second word — see
# `EN_ADVERB_SKIP`) is one of these is safe to wrap in an EN_FRAMES modal
# template. Built from every distinct first word actually used in
# `training_data.py`'s own English utterances (see the corpus-wide frequency
# scan this script's history was built from) plus the handful of verbs only
# ever appearing after a skipped adverb.
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",
}

# Adverbs/modifiers that can open an otherwise-imperative English clause
# ("coarsely chop the tomatoes", "deep fry until golden") — checked one word
# further in when the first word matches one of these, rather than treated
# as declarative.
EN_ADVERB_SKIP = {
    "coarsely", "roughly", "finely", "quickly", "lightly", "briefly", "gently", "carefully",
    "gradually", "very", "thoroughly", "evenly", "generously", "slowly", "thinly", "deep", "blind",
    "dry",
}


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 generate(existing: list[str], frames: list[str], is_led) -> list[str]:
    """Returns up to `len(frames) * len(sources)` new, deduplicated
    utterances wrapping every eligible source utterance in every frame —
    caller trims to however many it actually needs."""
    sources = [u for u in existing if is_led(u)]
    if not sources:
        return []
    existing_set = set(existing)
    out: list[str] = []
    seen = set(existing_set)
    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], locale: str) -> list[str]:
    target = 20
    if len(existing) >= target:
        return []
    if locale == "fr":
        pool = generate(existing, FR_FRAMES, is_fr_infinitive_led)
    else:
        pool = generate(existing, EN_FRAMES, is_en_imperative_led)
    needed = target - len(existing)
    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)

    # Find the TECH_STEP_TRAINING_DATA = [ ... ] assignment's list of
    # TechStepTrainingEntry(...) calls.
    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)

    # Collect (insertion_line_0indexed, indent, new_lines_to_insert) for
    # every utterances=[...] list that needs topping up, across every entry
    # — applied bottom-to-top so earlier line numbers stay valid.
    insertions: list[tuple[int, str, list[str]]] = []
    total_added = 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)
            for inner_kw in locale_call.keywords:
                if inner_kw.arg != "utterances":
                    continue
                utterances_list_node = inner_kw.value
                assert isinstance(utterances_list_node, ast.List)
                existing = [
                    elt.value for elt in utterances_list_node.elts if isinstance(elt, ast.Constant)
                ]
                new_ones = top_up(existing, locale)
                if not new_ones:
                    continue
                # Insert right after the last element's line, before the
                # closing "]" — indentation matched to the last existing
                # element's own line.
                last_elt = utterances_list_node.elts[-1]
                insert_after_line = last_elt.end_lineno - 1  # 0-indexed
                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 __name__ == "__main__":
    main()