diff --git a/etc/scripts/dataset_pipeline/add_ml_phrases.py b/etc/scripts/dataset_pipeline/add_ml_phrases.py new file mode 100644 index 0000000000..955e56a832 --- /dev/null +++ b/etc/scripts/dataset_pipeline/add_ml_phrases.py @@ -0,0 +1,400 @@ +# Run the trained phrase tagger over license rules and mark its predictions. +from dataclasses import dataclass +import json +import os +import sys +import unicodedata +from numbers import Integral +from pathlib import Path + +import click + +# Avoid importing TensorFlow through Transformers. +os.environ.setdefault("USE_TF", "0") + +sys.path.insert(0, str(Path(__file__).parent)) + +from licensedcode.models import rules_data_dir +from licensedcode.required_phrases import add_required_phrase_to_rule +from licensedcode.required_phrases import find_phrase_spans_in_text +from licensedcode.required_phrases import get_base_rules_by_expression +from licensedcode.required_phrases import RequiredPhraseRuleCandidate +from licensedcode.tokenize import get_existing_required_phrase_spans +from licensedcode.tokenize import required_phrase_splitter + +from train_model import extract_spans +from train_model import first_subword_positions +from train_model import ID2LABEL +from train_model import load_final_model + + +MIN_TOKENS = 2 +MIN_SINGLE_TOKEN_LEN = 5 +MAX_RULE_TEXT = 4000 + + +@dataclass(frozen=True) +class PhrasePrediction: + """One predicted required phrase.""" + + text: str + start_word: int + end_word: int + confidence: float + + +@dataclass(frozen=True) +class PredictionResult: + """Predictions and tokenization details for one rule.""" + + words: tuple[str, ...] + phrases: tuple[PhrasePrediction, ...] + truncated: bool + + +def load_model(model, hf_token=None): + """Load and validate a local or Hugging Face Final_Model.""" + model_dir = Path(model) + if not model_dir.is_dir(): + from huggingface_hub import snapshot_download + + model_dir = Path(snapshot_download(repo_id=model, token=hf_token)) + + tagger, tokenizer = load_final_model(model_dir, offline=True) + config = json.loads((model_dir / "train_config.json").read_text(encoding="utf-8")) + return tagger, tokenizer, config["max_length"] + + +def words_from_text(text): + """Return words tokenized as they are in the training dataset.""" + text = text.replace("\r\n", "\n").replace("\r", "\n") + return required_phrase_splitter(unicodedata.normalize("NFKC", text)) + + +def is_updatable(rule): + """Return True if a rule can receive predicted required phrases.""" + if rule.is_from_license: + return False + if len(rule.text) > MAX_RULE_TEXT: + return False + if not rule.is_approx_matchable: + return False + if rule.skip_for_required_phrase_generation: + return False + return not get_existing_required_phrase_spans(rule.text) + + +def select_rules(license_expression=None): + """Return eligible rules grouped by license expression.""" + try: + rules_by_expression = get_base_rules_by_expression(license_expression) + except KeyError: + raise click.ClickException( + f"No rules for license expression: {license_expression}" + ) from None + + selected = {} + for expression, rules in rules_by_expression.items(): + updatable = [rule for rule in rules if is_updatable(rule)] + if updatable: + selected[expression] = updatable + return selected + + +def _word_counts(word_ids): + counts = {} + for word_id in word_ids: + if word_id is not None: + counts[word_id] = counts.get(word_id, 0) + 1 + return counts + + +def encode_words(tokenizer, words, max_length): + """Encode the longest complete-word prefix and report truncation.""" + call = dict(is_split_into_words=True, add_special_tokens=True) + full = tokenizer(words, truncation=False, **call) + encoding = tokenizer(words, truncation=True, max_length=max_length, **call) + + full_counts = _word_counts(full.word_ids()) + retained_counts = _word_counts(encoding.word_ids()) + covered_words = max(retained_counts, default=-1) + 1 + complete_words = covered_words + + if covered_words and retained_counts[covered_words - 1] != full_counts[covered_words - 1]: + complete_words -= 1 + encoding = tokenizer(words[:complete_words], truncation=False, **call) + + if not complete_words: + raise ValueError("Tokenizer retained no complete words") + if len(encoding["input_ids"]) > max_length: + raise ValueError("Complete-word encoding exceeds the model maximum length") + + return encoding, complete_words < len(words) + + +def span_confidence(crf, word_emissions, tags, mask, free, span): + """Return the CRF probability mass agreeing with one decoded span.""" + start, end = span + pinned = word_emissions.clone() + floor = float(word_emissions.min()) - 10000.0 + + for position in range(start, end + 1): + label = int(tags[0, position]) + keep = float(pinned[0, position, label]) + pinned[0, position] = floor + pinned[0, position, label] = keep + + constrained = crf(pinned, tags, mask=mask, reduction="none") + confidence = float((free - constrained).detach().exp()) + return min(max(confidence, 0.0), 1.0) + + +def predict_rule(tagger, tokenizer, max_length, text): + """Return phrase predictions for rule text without changing a rule.""" + import torch + + words = words_from_text(text) + if not words: + return PredictionResult(words=(), phrases=(), truncated=False) + + encoding, truncated = encode_words(tokenizer, words, max_length) + word_ids = encoding.word_ids() + positions = first_subword_positions(word_ids) + device = next(tagger.parameters()).device + input_ids = torch.tensor([encoding["input_ids"]], dtype=torch.long, device=device) + attention_mask = torch.tensor( + [encoding["attention_mask"]], + dtype=torch.long, + device=device, + ) + + with torch.inference_mode(): + emissions = tagger.emissions(input_ids, attention_mask) + word_emissions = emissions[:, positions].float() + mask = torch.ones( + word_emissions.shape[:2], + dtype=torch.bool, + device=word_emissions.device, + ) + decoded = tagger.crf.decode(word_emissions, mask=mask)[0] + tags = torch.tensor([decoded], device=word_emissions.device) + free = tagger.crf(word_emissions, tags, mask=mask, reduction="none") + labels = [ID2LABEL[int(label)] for label in decoded] + predictions = [ + PhrasePrediction( + text=" ".join(words[start : end + 1]), + start_word=start, + end_word=end, + confidence=span_confidence( + tagger.crf, + word_emissions, + tags, + mask, + free, + (start, end), + ), + ) + for start, end in extract_spans(labels) + ] + + predictions.sort(key=lambda prediction: (prediction.start_word, prediction.end_word)) + return PredictionResult( + words=tuple(words), + phrases=tuple(predictions), + truncated=truncated, + ) + + +def phrases_from_tags(tags, words): + """Return unique predicted phrase texts, longest first.""" + phrases = { + " ".join(words[start : end + 1]) + for start, end in extract_spans(tags) + } + return sorted(phrases, key=lambda phrase: (-len(phrase), phrase)) + + +def predict_phrases(tagger, tokenizer, max_length, words): + """Return predicted phrases and whether the rule was truncated.""" + if not words: + return [], False + + import torch + + encoding, truncated = encode_words(tokenizer, words, max_length) + word_ids = encoding.word_ids() + input_ids = torch.tensor([encoding["input_ids"]], dtype=torch.long) + attention_mask = torch.tensor([encoding["attention_mask"]], dtype=torch.long) + + with torch.no_grad(): + predicted = tagger.predict_words(input_ids, attention_mask, word_ids) + + word_count = len(set(word_id for word_id in word_ids if word_id is not None)) + if len(predicted) != word_count: + raise ValueError("Model returned a different number of labels than encoded words") + + tags = [] + for label in predicted: + if isinstance(label, bool) or not isinstance(label, Integral) or label not in ID2LABEL: + raise ValueError(f"Model returned an invalid label ID: {label!r}") + tags.append(ID2LABEL[int(label)]) + + return phrases_from_tags(tags, words), truncated + + +def new_counts(): + return dict( + rules=0, + truncated=0, + rejected=0, + not_found=0, + injected=0, + skipped=0, + written=0, + ) + + +def inject(rule, phrases, counts, dry_run=False, verbose=False): + """Validate and add predicted phrases, writing the rule at most once.""" + candidates = [] + for phrase in phrases: + candidate = RequiredPhraseRuleCandidate.create(rule.license_expression, phrase) + if not candidate.is_good(rule, MIN_TOKENS, MIN_SINGLE_TOKEN_LEN): + counts["rejected"] += 1 + continue + if not find_phrase_spans_in_text(rule.text, phrase): + counts["not_found"] += 1 + continue + candidates.append(phrase) + + if not candidates: + return False + + original_text = rule.text + original_source = rule.source + source = f"{original_source} ml_model" if original_source else "ml_model" + + for phrase in candidates: + updated = add_required_phrase_to_rule( + rule=rule, + required_phrase=phrase, + source=source, + debug=verbose, + dry_run=True, + ) + if updated: + counts["injected"] += 1 + else: + counts["skipped"] += 1 + + if rule.text == original_text: + return False + if not dry_run: + rule.dump(rules_data_dir) + return True + + +def process_rules( + selected, + tagger, + tokenizer, + max_length, + dry_run=False, + limit=0, + verbose=False, +): + """Predict and mark phrases in selected rules and return run counts.""" + counts = new_counts() + total = sum(len(rules) for rules in selected.values()) + click.echo(f"Tagging {total} rules in {len(selected)} license expressions") + + for expression, rules in selected.items(): + if verbose: + click.echo(f"{expression}: {len(rules)} rules") + + for rule in rules: + if limit and counts["rules"] >= limit: + click.echo(f"Stopping at {limit} rules") + return counts + + counts["rules"] += 1 + words = words_from_text(rule.text) + phrases, truncated = predict_phrases(tagger, tokenizer, max_length, words) + if truncated: + counts["truncated"] += 1 + if not phrases: + continue + + if verbose: + click.echo(f" {rule.identifier}: {phrases}") + if inject(rule, phrases, counts, dry_run=dry_run, verbose=verbose): + counts["written"] += 1 + + return counts + + +@click.command() +@click.option( + "--model", + required=True, + help="Final model directory or Hugging Face repository.", +) +@click.option( + "--license-expression", + help="Only update rules for this license expression.", +) +@click.option( + "--dry-run", + is_flag=True, + help="Predict and validate phrases without saving rules.", +) +@click.option( + "--limit", + default=0, + type=click.IntRange(min=0), + help="Stop after this many rules; zero processes all rules.", +) +@click.option( + "-v", + "--verbose", + is_flag=True, + help="Print predictions for each rule.", +) +@click.help_option("-h", "--help") +def main(model, license_expression, dry_run, limit, verbose): + """Add model-predicted required phrases to license rules.""" + selected = select_rules(license_expression=license_expression) + if not selected: + click.echo("No eligible rules found") + return + + tagger, tokenizer, max_length = load_model( + model, + hf_token=os.environ.get("HF_TOKEN"), + ) + counts = process_rules( + selected=selected, + tagger=tagger, + tokenizer=tokenizer, + max_length=max_length, + dry_run=dry_run, + limit=limit, + verbose=verbose, + ) + + click.echo(f"\nrules processed : {counts['rules']}") + click.echo(f" truncated : {counts['truncated']}") + click.echo(f"phrases injected : {counts['injected']}") + click.echo(f" rejected : {counts['rejected']}") + click.echo(f" not found : {counts['not_found']}") + click.echo(f" nothing to add : {counts['skipped']}") + click.echo(f"rules written : {counts['written']}") + + if dry_run: + click.echo("Dry run: no rules were saved") + elif counts["written"]: + click.echo("Run scancode-reindex-licenses to use the new required phrases") + + +if __name__ == "__main__": + main() diff --git a/etc/scripts/dataset_pipeline/review_ml_phrases.py b/etc/scripts/dataset_pipeline/review_ml_phrases.py new file mode 100644 index 0000000000..3bc3a8d1bb --- /dev/null +++ b/etc/scripts/dataset_pipeline/review_ml_phrases.py @@ -0,0 +1,519 @@ +# Copyright (c) nexB Inc. and others. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Review model-predicted required phrases before changing ScanCode rules.""" + +import difflib +import hashlib +import json +import math +import os +from pathlib import Path +import sys +import tempfile + +import click + +sys.path.insert(0, str(Path(__file__).parent)) + +from licensedcode.models import Rule +from licensedcode.models import rules_data_dir +from licensedcode.required_phrases import add_required_phrase_to_rule +from licensedcode.required_phrases import find_phrase_spans_in_text +from licensedcode.required_phrases import RequiredPhraseRuleCandidate + +from add_ml_phrases import inject as add_predicted_phrases +from add_ml_phrases import load_model +from add_ml_phrases import MIN_SINGLE_TOKEN_LEN +from add_ml_phrases import MIN_TOKENS +from add_ml_phrases import new_counts +from add_ml_phrases import predict_rule +from add_ml_phrases import select_rules + + +PENDING = "pending" +APPROVED = "approved" +REJECTED = "rejected" +DECISIONS = {PENDING, APPROVED, REJECTED} + +RECORD_FIELDS = { + "identifier", + "license_expression", + "text_sha256", + "truncated", + "phrases", +} +PHRASE_FIELDS = { + "text", + "predicted_text", + "start_word", + "end_word", + "confidence", + "decision", +} + + +def text_sha256(text): + """Return a stable digest for rule text.""" + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +def phrase_sort_key(phrase): + """Return a stable display and injection order for a phrase entry.""" + return -len(phrase["text"]), phrase["text"], phrase["start_word"] + + +def prediction_record(rule, predictions, truncated): + """Return one review record with the strongest occurrence of each phrase.""" + predictions_by_text = {} + for prediction in predictions: + existing = predictions_by_text.get(prediction.text) + if existing is None or prediction.confidence > existing.confidence: + predictions_by_text[prediction.text] = prediction + + phrases = [ + { + "text": prediction.text, + "predicted_text": prediction.text, + "start_word": prediction.start_word, + "end_word": prediction.end_word, + "confidence": prediction.confidence, + "decision": PENDING, + } + for prediction in predictions_by_text.values() + ] + phrases.sort(key=phrase_sort_key) + return { + "identifier": rule.identifier, + "license_expression": rule.license_expression, + "text_sha256": text_sha256(rule.text), + "truncated": truncated, + "phrases": phrases, + } + + +def validate_phrase(phrase, path, line_number, phrase_number): + """Validate and return one phrase entry from a review file.""" + location = f"{path} line {line_number}, phrase {phrase_number}" + if type(phrase) is not dict: + raise click.ClickException(f"{location}: phrase must be an object") + if set(phrase) != PHRASE_FIELDS: + raise click.ClickException(f"{location}: phrase fields are invalid") + + for field in ("text", "predicted_text", "decision"): + if type(phrase[field]) is not str or not phrase[field]: + raise click.ClickException(f"{location}: {field} must be a non-empty string") + for field in ("start_word", "end_word"): + if isinstance(phrase[field], bool) or not isinstance(phrase[field], int): + raise click.ClickException(f"{location}: {field} must be an integer") + if phrase["start_word"] < 0 or phrase["end_word"] < phrase["start_word"]: + raise click.ClickException(f"{location}: word offsets are invalid") + + confidence = phrase["confidence"] + if isinstance(confidence, bool) or not isinstance(confidence, (int, float)): + raise click.ClickException(f"{location}: confidence must be a number") + if not math.isfinite(confidence) or not 0.0 <= confidence <= 1.0: + raise click.ClickException(f"{location}: confidence must be between 0 and 1") + if phrase["decision"] not in DECISIONS: + raise click.ClickException(f"{location}: decision is invalid") + return phrase + + +def validate_record(record, path, line_number): + """Validate and return one record from a review file.""" + location = f"{path} line {line_number}" + if type(record) is not dict: + raise click.ClickException(f"{location}: record must be an object") + if set(record) != RECORD_FIELDS: + raise click.ClickException(f"{location}: record fields are invalid") + + for field in ("identifier", "license_expression", "text_sha256"): + if type(record[field]) is not str or not record[field]: + raise click.ClickException(f"{location}: {field} must be a non-empty string") + if Path(record["identifier"]).name != record["identifier"]: + raise click.ClickException(f"{location}: identifier must be a rule filename") + digest = record["text_sha256"] + if len(digest) != 64 or any(character not in "0123456789abcdef" for character in digest): + raise click.ClickException(f"{location}: text_sha256 is invalid") + if type(record["truncated"]) is not bool: + raise click.ClickException(f"{location}: truncated must be a boolean") + if type(record["phrases"]) is not list or not record["phrases"]: + raise click.ClickException(f"{location}: phrases must be a non-empty list") + + seen = set() + for phrase_number, phrase in enumerate(record["phrases"], 1): + validate_phrase(phrase, path, line_number, phrase_number) + identity = (phrase["predicted_text"], phrase["start_word"], phrase["end_word"]) + if identity in seen: + raise click.ClickException(f"{location}: duplicate predicted phrase") + seen.add(identity) + return record + + +def read_review_file(path): + """Return all validated records from a JSONL review file.""" + records = [] + identifiers = set() + try: + lines = Path(path).open(encoding="utf-8") + except (OSError, UnicodeError) as error: + raise click.ClickException(f"Cannot read review file {path}: {error}") from error + + with lines: + for line_number, line in enumerate(lines, 1): + if not line.strip(): + continue + try: + record = json.loads(line) + except json.JSONDecodeError as error: + raise click.ClickException( + f"{path} line {line_number}: malformed JSON: {error.msg}" + ) from error + validate_record(record, path, line_number) + if record["identifier"] in identifiers: + raise click.ClickException( + f"{path} line {line_number}: duplicate rule identifier" + ) + identifiers.add(record["identifier"]) + records.append(record) + return records + + +def write_review_file(path, records): + """Atomically replace a review file with records in JSONL format.""" + path = Path(path) + path.parent.mkdir(parents=True, exist_ok=True) + descriptor, temporary_name = tempfile.mkstemp( + prefix=f".{path.name}.", + suffix=".tmp", + dir=path.parent, + ) + temporary = Path(temporary_name) + try: + with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as output: + for record in records: + output.write(json.dumps(record, ensure_ascii=False, allow_nan=False) + "\n") + output.flush() + os.fsync(output.fileno()) + os.replace(temporary, path) + finally: + if temporary.exists(): + temporary.unlink() + + +def load_current_rule(record): + """Return the unchanged rule named by record, or a stale reason.""" + rule_path = Path(rules_data_dir) / record["identifier"] + if not rule_path.is_file(): + return None, "rule file is missing" + rule = Rule.from_file(str(rule_path), is_builtin=True) + if rule.license_expression != record["license_expression"]: + return None, "license expression changed" + if text_sha256(rule.text) != record["text_sha256"]: + return None, "rule text changed" + return rule, None + + +def preview_injection(rule, phrase): + """Return the exact in-memory result of injecting phrase without saving.""" + original_text = rule.text + original_source = rule.source + changed = add_required_phrase_to_rule( + rule=rule, + required_phrase=phrase, + source="ml_model", + dry_run=True, + ) + preview = rule.text + rule.text = original_text + rule.source = original_source + return changed, preview + + +def render_diff(identifier, before, after): + """Print a unified diff for one proposed rule update.""" + lines = difflib.unified_diff( + before.splitlines(keepends=True), + after.splitlines(keepends=True), + fromfile=f"a/{identifier}", + tofile=f"b/{identifier}", + ) + colors = {"+": "green", "-": "red", "@": "cyan"} + for line in lines: + line = line.rstrip("\n") + click.echo(click.style(line, fg=colors.get(line[:1]))) + + +def is_candidate(rule, phrase): + """Return whether phrase passes ScanCode's candidate and location checks.""" + candidate = RequiredPhraseRuleCandidate.create(rule.license_expression, phrase) + return candidate.is_good(rule, MIN_TOKENS, MIN_SINGLE_TOKEN_LEN) and bool( + find_phrase_spans_in_text(rule.text, phrase) + ) + + +def edit_phrase(rule, phrase): + """Prompt for a valid replacement phrase, or return False to cancel.""" + click.echo("\nrule text") + click.echo(rule.text) + while True: + replacement = click.prompt( + "phrase, empty to cancel", + default="", + show_default=False, + ).strip() + if not replacement: + return False + if not is_candidate(rule, replacement): + click.echo("The phrase is not a valid candidate in this rule.") + continue + changed, preview = preview_injection(rule, replacement) + if not changed: + click.echo("The phrase cannot be added to this rule.") + continue + render_diff(rule.identifier, rule.text, preview) + phrase["text"] = replacement + phrase["decision"] = APPROVED + return True + + +def review_phrase(rule, phrase): + """Prompt for one phrase decision and return False when review should stop.""" + while True: + answer = click.prompt( + "[y] approve [n] reject [e] edit [q] quit", + default="", + show_default=False, + ).strip().lower() + if answer == "y": + phrase["decision"] = APPROVED + return True + if answer == "n": + phrase["decision"] = REJECTED + return True + if answer == "e": + if edit_phrase(rule, phrase): + return True + elif answer == "q": + return False + else: + click.echo("Enter y, n, e, or q.") + + +@click.group(name="review-model-required-phrases") +@click.help_option("-h", "--help") +def review_model_required_phrases(): + """Review model predictions before adding them to license rules.""" + + +def predict_records(selected, tagger, tokenizer, max_length, limit=0, verbose=False): + """Return review records and counts for selected rules.""" + records = [] + counts = {"rules": 0, "truncated": 0, "rejected": 0, "not_found": 0} + for rules in selected.values(): + for rule in rules: + if limit and counts["rules"] >= limit: + return records, counts + counts["rules"] += 1 + result = predict_rule(tagger, tokenizer, max_length, rule.text) + if result.truncated: + counts["truncated"] += 1 + + predictions = [] + for prediction in result.phrases: + candidate = RequiredPhraseRuleCandidate.create( + rule.license_expression, + prediction.text, + ) + if not candidate.is_good(rule, MIN_TOKENS, MIN_SINGLE_TOKEN_LEN): + counts["rejected"] += 1 + elif not find_phrase_spans_in_text(rule.text, prediction.text): + counts["not_found"] += 1 + else: + predictions.append(prediction) + if not predictions: + continue + + record = prediction_record(rule, predictions, result.truncated) + records.append(record) + if verbose: + click.echo(f"{rule.identifier}: {[phrase['text'] for phrase in record['phrases']]}") + return records, counts + + +@review_model_required_phrases.command() +@click.option("--model", required=True, help="Final model directory or Hugging Face repository.") +@click.option( + "--review-file", + required=True, + type=click.Path(dir_okay=False, path_type=Path), + help="New JSONL file for predictions.", +) +@click.option("--license-expression", help="Only predict for this license expression.") +@click.option( + "--limit", + default=0, + type=click.IntRange(min=0), + help="Stop after this many rules; zero processes all rules.", +) +@click.option("-v", "--verbose", is_flag=True, help="Print predictions for each rule.") +@click.help_option("-h", "--help") +def predict(model, review_file, license_expression, limit, verbose): + """Write validated model predictions for human review.""" + if review_file.exists(): + raise click.ClickException(f"Review file already exists: {review_file}") + + selected = select_rules(license_expression=license_expression) + if not selected: + click.echo("No eligible rules found") + return + tagger, tokenizer, max_length = load_model( + model, + hf_token=os.environ.get("HF_TOKEN"), + ) + + records, counts = predict_records( + selected=selected, + tagger=tagger, + tokenizer=tokenizer, + max_length=max_length, + limit=limit, + verbose=verbose, + ) + write_review_file(review_file, records) + filed = sum(len(record["phrases"]) for record in records) + click.echo(f"rules processed : {counts['rules']}") + click.echo(f" truncated : {counts['truncated']}") + click.echo(f"phrases filed : {filed}") + click.echo(f" rejected : {counts['rejected']}") + click.echo(f" not found : {counts['not_found']}") + click.echo(f"review file : {review_file}") + + +@review_model_required_phrases.command() +@click.option( + "--review-file", + required=True, + type=click.Path(exists=True, dir_okay=False, path_type=Path), + help="JSONL prediction file to review.", +) +@click.help_option("-h", "--help") +def review(review_file): + """Approve, reject, or edit every pending prediction.""" + records = read_review_file(review_file) + waiting = sum( + phrase["decision"] == PENDING + for record in records + for phrase in record["phrases"] + ) + if not waiting: + click.echo("Nothing left to review") + return + + click.echo(f"{waiting} phrases waiting") + for record in records: + pending = [phrase for phrase in record["phrases"] if phrase["decision"] == PENDING] + if not pending: + continue + rule, stale_reason = load_current_rule(record) + if stale_reason: + click.echo(f"{record['identifier']}: stale review record ({stale_reason})") + continue + + for phrase in pending: + if not is_candidate(rule, phrase["text"]): + phrase["decision"] = REJECTED + write_review_file(review_file, records) + click.echo(f"{record['identifier']}: phrase is no longer a valid candidate") + continue + changed, preview = preview_injection(rule, phrase["text"]) + if not changed: + phrase["decision"] = REJECTED + write_review_file(review_file, records) + click.echo(f"{record['identifier']}: phrase can no longer be added") + continue + + click.echo(f"\n{record['identifier']} {record['license_expression']}") + click.echo(f"phrase: {phrase['text']} confidence: {phrase['confidence']:.1%}") + render_diff(record["identifier"], rule.text, preview) + if not review_phrase(rule, phrase): + return + write_review_file(review_file, records) + + +def prepare_apply(records): + """Return validated rules and approved phrases before any mutation.""" + work = [] + for record in records: + phrases = { + phrase["text"] + for phrase in record["phrases"] + if phrase["decision"] == APPROVED + } + if not phrases: + continue + + rule, stale_reason = load_current_rule(record) + if stale_reason: + raise click.ClickException( + f"{record['identifier']}: stale review record ({stale_reason})" + ) + for phrase in phrases: + if not is_candidate(rule, phrase): + raise click.ClickException( + f"{record['identifier']}: approved phrase is no longer a valid candidate: " + f"{phrase!r}" + ) + changed, _preview = preview_injection(rule, phrase) + if not changed: + raise click.ClickException( + f"{record['identifier']}: approved phrase cannot be added: {phrase!r}" + ) + work.append((rule, sorted(phrases, key=lambda phrase: (-len(phrase), phrase)))) + return work + + +@review_model_required_phrases.command() +@click.option( + "--review-file", + required=True, + type=click.Path(exists=True, dir_okay=False, path_type=Path), + help="Reviewed JSONL prediction file to apply.", +) +@click.option("--dry-run", is_flag=True, help="Validate without saving rules.") +@click.option("-v", "--verbose", is_flag=True, help="Print phrases for each rule.") +@click.help_option("-h", "--help") +def apply(review_file, dry_run, verbose): + """Add approved phrases after validating the current rules.""" + records = read_review_file(review_file) + pending = sum( + phrase["decision"] == PENDING + for record in records + for phrase in record["phrases"] + ) + if pending: + raise click.ClickException(f"{pending} phrases still need review") + + work = prepare_apply(records) + counts = new_counts() + for rule, phrases in work: + counts["rules"] += 1 + if verbose: + click.echo(f"{rule.identifier}: {phrases}") + if add_predicted_phrases(rule, phrases, counts, dry_run=dry_run, verbose=verbose): + counts["written"] += 1 + + click.echo(f"rules processed : {counts['rules']}") + click.echo(f"phrases injected : {counts['injected']}") + click.echo(f" rejected : {counts['rejected']}") + click.echo(f" not found : {counts['not_found']}") + click.echo(f" nothing to add : {counts['skipped']}") + click.echo(f"rules written : {counts['written']}") + if dry_run: + click.echo("Dry run: no rules were saved") + elif counts["written"]: + click.echo("Run scancode-reindex-licenses to use the new required phrases") + + +if __name__ == "__main__": + review_model_required_phrases() diff --git a/etc/scripts/dataset_pipeline/test_add_ml_phrases.py b/etc/scripts/dataset_pipeline/test_add_ml_phrases.py new file mode 100644 index 0000000000..7a722ddc07 --- /dev/null +++ b/etc/scripts/dataset_pipeline/test_add_ml_phrases.py @@ -0,0 +1,386 @@ +import json +import sys +from pathlib import Path +from types import SimpleNamespace + +import click +import pytest +from click.testing import CliRunner + +sys.path.insert(0, str(Path(__file__).parent)) + +import add_ml_phrases +from add_ml_phrases import encode_words +from add_ml_phrases import inject +from add_ml_phrases import is_updatable +from add_ml_phrases import load_model +from add_ml_phrases import main +from add_ml_phrases import new_counts +from add_ml_phrases import phrases_from_tags +from add_ml_phrases import predict_phrases +from add_ml_phrases import predict_rule +from add_ml_phrases import process_rules +from add_ml_phrases import select_rules +from add_ml_phrases import words_from_text +from train_model import LABEL2ID + +from licensedcode.models import Rule + + +class FakeRule: + def __init__( + self, + text="some license text here", + is_from_license=False, + is_approx_matchable=True, + skip=False, + ): + self.text = text + self.is_from_license = is_from_license + self.is_approx_matchable = is_approx_matchable + self.skip_for_required_phrase_generation = skip + + +class FakeEncoding(dict): + def __init__(self, word_ids): + super().__init__( + input_ids=list(range(len(word_ids))), + attention_mask=[1] * len(word_ids), + ) + self._word_ids = word_ids + + def word_ids(self): + return list(self._word_ids) + + +class FakeTokenizer: + """Tokenize words with configurable subword counts.""" + + is_fast = True + + def __init__(self, subwords=None): + self.subwords = subwords or {} + + def __call__(self, words, truncation, max_length=None, **kwargs): + word_ids = [None] + for index, word in enumerate(words): + word_ids.extend([index] * self.subwords.get(word, 1)) + word_ids.append(None) + if truncation and len(word_ids) > max_length: + word_ids = word_ids[: max_length - 1] + [None] + return FakeEncoding(word_ids) + + +class StubTagger: + def __init__(self, labels): + self.labels = labels + self.calls = 0 + + def predict_words(self, input_ids, attention_mask, word_ids): + self.calls += 1 + count = len(set(word_id for word_id in word_ids if word_id is not None)) + labels = self.labels + [LABEL2ID["O"]] * count + return labels[:count] + + +class TestLoadModel: + def test_loads_a_valid_local_final_model(self, tmp_path, monkeypatch): + config = tmp_path / "train_config.json" + config.write_text(json.dumps({"max_length": 256}), encoding="utf-8") + tagger = object() + tokenizer = object() + calls = [] + + def load_final_model(model_dir, offline): + calls.append((model_dir, offline)) + return tagger, tokenizer + + monkeypatch.setattr(add_ml_phrases, "load_final_model", load_final_model) + assert load_model(tmp_path) == (tagger, tokenizer, 256) + assert calls == [(tmp_path, True)] + + def test_downloads_before_using_the_strict_loader(self, tmp_path, monkeypatch): + model_dir = tmp_path / "snapshot" + model_dir.mkdir() + (model_dir / "train_config.json").write_text( + json.dumps({"max_length": 512}), encoding="utf-8" + ) + downloads = [] + hub = SimpleNamespace( + snapshot_download=lambda repo_id, token: downloads.append((repo_id, token)) + or str(model_dir) + ) + monkeypatch.setitem(sys.modules, "huggingface_hub", hub) + monkeypatch.setattr( + add_ml_phrases, + "load_final_model", + lambda path, offline: (object(), object()), + ) + + load_model("owner/model", hf_token="secret") + assert downloads == [("owner/model", "secret")] + + +class TestWordsFromText: + def test_uses_dataset_tokenization(self): + assert words_from_text("Apache-2.0 License") == ["Apache", "2", "0", "License"] + + def test_normalizes_line_endings_and_unicode(self): + assert words_from_text("a\ufb01x\r\ntwo\rthree") == ["afix", "two", "three"] + + +class TestEncodeWords: + def test_keeps_all_complete_words(self): + encoding, truncated = encode_words(FakeTokenizer(), ["one", "two"], 10) + assert encoding.word_ids() == [None, 0, 1, None] + assert not truncated + + def test_removes_a_partially_truncated_word(self): + tokenizer = FakeTokenizer({"many": 3}) + encoding, truncated = encode_words(tokenizer, ["one", "many", "three"], 4) + assert encoding.word_ids() == [None, 0, None] + assert truncated + + def test_rejects_when_no_complete_word_fits(self): + with pytest.raises(ValueError, match="no complete words"): + encode_words(FakeTokenizer({"many": 4}), ["many"], 3) + + +class TestPhrasesFromTags: + def test_returns_longest_unique_phrases_first(self): + words = ["mit", "license", "mit", "other"] + tags = ["B-REQ", "E-REQ", "S-REQ", "O"] + assert phrases_from_tags(tags, words) == ["mit license", "mit"] + + def test_rejects_invalid_bioes(self): + with pytest.raises(ValueError, match="invalid BIOES"): + phrases_from_tags(["B-REQ", "O"], ["one", "two"]) + + +class TestPredictPhrases: + def test_predicts_from_complete_words(self): + labels = [LABEL2ID["B-REQ"], LABEL2ID["E-REQ"], LABEL2ID["O"]] + phrases, truncated = predict_phrases( + StubTagger(labels), FakeTokenizer(), 10, ["MIT", "License", "text"] + ) + assert phrases == ["MIT License"] + assert not truncated + + def test_keeps_a_valid_span_at_the_truncation_boundary(self): + labels = [LABEL2ID["O"], LABEL2ID["S-REQ"]] + phrases, truncated = predict_phrases( + StubTagger(labels), FakeTokenizer(), 4, ["one", "two", "three"] + ) + assert phrases == ["two"] + assert truncated + + @pytest.mark.parametrize("label", [99, 1.5, True]) + def test_rejects_an_invalid_label_id(self, label): + with pytest.raises(ValueError, match="invalid label ID"): + predict_phrases(StubTagger([label]), FakeTokenizer(), 10, ["one"]) + + def test_rejects_the_wrong_number_of_labels(self): + tagger = StubTagger([]) + tagger.predict_words = lambda *args: [] + with pytest.raises(ValueError, match="different number"): + predict_phrases(tagger, FakeTokenizer(), 10, ["one"]) + + def test_empty_text_does_not_call_the_model(self): + tagger = StubTagger([]) + assert predict_phrases(tagger, FakeTokenizer(), 10, []) == ([], False) + assert tagger.calls == 0 + + +class TestPredictRule: + def test_returns_scored_phrase_offsets(self): + torch = pytest.importorskip("torch") + from phrase_model import ConstrainedCRF + + class Tagger(torch.nn.Module): + def __init__(self): + super().__init__() + self.crf = ConstrainedCRF(5, batch_first=True) + with torch.no_grad(): + for parameter in self.crf.parameters(): + parameter.zero_() + + def emissions(self, input_ids, attention_mask): + emissions = torch.zeros((1, input_ids.shape[1], 5)) + emissions[0, 2, LABEL2ID["B-REQ"]] = 9.0 + emissions[0, 3, LABEL2ID["E-REQ"]] = 9.0 + return emissions + + result = predict_rule( + Tagger(), + FakeTokenizer(), + 20, + "under the MIT License terms", + ) + + assert [phrase.text for phrase in result.phrases] == ["the MIT"] + assert result.phrases[0].start_word == 1 + assert result.phrases[0].end_word == 2 + assert 0.0 <= result.phrases[0].confidence <= 1.0 + assert not result.truncated + + def test_empty_text_does_not_run_model(self): + assert predict_rule(object(), FakeTokenizer(), 20, "").phrases == () + + +class TestIsUpdatable: + @pytest.mark.parametrize( + "rule", + [ + FakeRule(is_from_license=True), + FakeRule(text="x" * 4001), + FakeRule(is_approx_matchable=False), + FakeRule(skip=True), + FakeRule(text="under the {{mit license}} terms"), + ], + ) + def test_excludes_ineligible_rules(self, rule): + assert not is_updatable(rule) + + def test_accepts_a_plain_rule(self): + assert is_updatable(FakeRule()) + + +class TestSelectRules: + def test_filters_and_groups_rules(self, monkeypatch): + monkeypatch.setattr( + add_ml_phrases, + "get_base_rules_by_expression", + lambda expression: { + "mit": [FakeRule(), FakeRule(is_from_license=True)], + "bsd-new": [FakeRule(skip=True)], + }, + ) + selected = select_rules() + assert list(selected) == ["mit"] + assert len(selected["mit"]) == 1 + + def test_forwards_the_expression_filter(self, monkeypatch): + expressions = [] + + def get_rules(expression): + expressions.append(expression) + return {expression: [FakeRule()]} + + monkeypatch.setattr(add_ml_phrases, "get_base_rules_by_expression", get_rules) + assert list(select_rules("mit")) == ["mit"] + assert expressions == ["mit"] + + def test_reports_an_unknown_expression(self, monkeypatch): + def get_rules(expression): + raise KeyError(expression) + + monkeypatch.setattr(add_ml_phrases, "get_base_rules_by_expression", get_rules) + with pytest.raises(click.ClickException, match="No rules"): + select_rules("unknown") + + +def make_rule(text, source=None): + rule = Rule( + license_expression="mit", + identifier="mit_test.RULE", + text=text, + is_license_reference=True, + relevance=100, + ) + rule.source = source + return rule + + +TEXT = "Permission is granted under the MIT License to do things with this" + + +class TestInject: + def test_marks_two_phrases_and_preserves_source(self): + rule = make_rule(TEXT, source="mit_1.RULE") + counts = new_counts() + assert inject(rule, ["MIT License", "do things"], counts, dry_run=True) + assert counts["injected"] == 2 + assert rule.text.count("{{") == rule.text.count("}}") == 2 + assert rule.source == "mit_1.RULE ml_model" + + def test_rejects_an_unsuitable_phrase(self): + rule = make_rule(TEXT) + counts = new_counts() + assert not inject(rule, ["is"], counts, dry_run=True) + assert counts["rejected"] == 1 + assert "{{" not in rule.text + + def test_counts_a_phrase_not_found_in_the_rule(self): + rule = make_rule(TEXT) + counts = new_counts() + assert not inject(rule, ["Apache License"], counts, dry_run=True) + assert counts["not_found"] == 1 + + def test_writes_a_rule_once(self, monkeypatch): + rule = make_rule(TEXT) + writes = [] + monkeypatch.setattr(Rule, "dump", lambda self, directory: writes.append(directory)) + assert inject(rule, ["MIT License", "do things"], new_counts()) + assert writes == [add_ml_phrases.rules_data_dir] + +class TestProcessRules: + def test_processes_selected_rules(self): + rule = make_rule(TEXT) + labels = [LABEL2ID["B-REQ"], LABEL2ID["I-REQ"], LABEL2ID["E-REQ"]] + counts = process_rules( + selected={"mit": [rule]}, + tagger=StubTagger(labels), + tokenizer=FakeTokenizer(), + max_length=50, + dry_run=True, + ) + assert counts["rules"] == 1 + assert counts["injected"] == 1 + assert counts["written"] == 1 + + def test_limit_stops_before_another_rule(self): + rules = [make_rule(TEXT) for _ in range(3)] + labels = [LABEL2ID["B-REQ"], LABEL2ID["I-REQ"], LABEL2ID["E-REQ"]] + counts = process_rules( + {"mit": rules}, StubTagger(labels), FakeTokenizer(), 50, dry_run=True, limit=2 + ) + assert counts["rules"] == 2 + + +class TestCommand: + def test_does_not_load_a_model_when_no_rules_are_eligible(self, monkeypatch): + monkeypatch.setattr(add_ml_phrases, "select_rules", lambda **kwargs: {}) + + def fail(*args, **kwargs): + raise AssertionError("model should not load") + + monkeypatch.setattr(add_ml_phrases, "load_model", fail) + result = CliRunner().invoke(main, ["--model", "unused"]) + assert result.exit_code == 0 + assert "No eligible rules found" in result.output + + def test_wires_selection_loading_and_processing(self, monkeypatch, tmp_path): + selected = {"mit": [object()]} + tagger = object() + tokenizer = object() + calls = [] + monkeypatch.setattr(add_ml_phrases, "select_rules", lambda **kwargs: selected) + monkeypatch.setattr( + add_ml_phrases, "load_model", lambda *args, **kwargs: (tagger, tokenizer, 256) + ) + + def process(**kwargs): + calls.append(kwargs) + return new_counts() + + monkeypatch.setattr(add_ml_phrases, "process_rules", process) + result = CliRunner().invoke( + main, + ["--model", str(tmp_path), "--license-expression", "mit", "--dry-run"], + ) + assert result.exit_code == 0 + assert calls[0]["selected"] is selected + assert calls[0]["tagger"] is tagger + assert calls[0]["tokenizer"] is tokenizer + assert calls[0]["max_length"] == 256 + assert calls[0]["dry_run"] is True diff --git a/etc/scripts/dataset_pipeline/test_review_ml_phrases.py b/etc/scripts/dataset_pipeline/test_review_ml_phrases.py new file mode 100644 index 0000000000..50904256d0 --- /dev/null +++ b/etc/scripts/dataset_pipeline/test_review_ml_phrases.py @@ -0,0 +1,435 @@ +# Copyright (c) nexB Inc. and others. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +import json + +import click +from click.testing import CliRunner +import pytest + +from licensedcode.models import Rule + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent)) + +import add_ml_phrases +from add_ml_phrases import PhrasePrediction +from add_ml_phrases import PredictionResult +import review_ml_phrases as review_module +from review_ml_phrases import apply +from review_ml_phrases import APPROVED +from review_ml_phrases import PENDING +from review_ml_phrases import predict_records +from review_ml_phrases import prediction_record +from review_ml_phrases import read_review_file +from review_ml_phrases import REJECTED +from review_ml_phrases import review_model_required_phrases +from review_ml_phrases import text_sha256 +from review_ml_phrases import validate_record +from review_ml_phrases import write_review_file + + +TEXT = "Permission is granted under the MIT License to do things with this software" + + +def make_rule(identifier="mit_test.RULE", text=TEXT, source=None): + return Rule( + identifier=identifier, + license_expression="mit", + text=text, + source=source, + is_license_reference=True, + relevance=100, + ) + + +def make_prediction(text="MIT License", start=5, end=6, confidence=0.9): + return PhrasePrediction( + text=text, + start_word=start, + end_word=end, + confidence=confidence, + ) + + +def make_record(rule=None, predictions=None): + rule = rule or make_rule() + predictions = predictions or [make_prediction()] + return prediction_record(rule, predictions, truncated=False) + + +def use_rules_directory(monkeypatch, tmp_path): + monkeypatch.setattr(review_module, "rules_data_dir", str(tmp_path)) + monkeypatch.setattr(add_ml_phrases, "rules_data_dir", str(tmp_path)) + + +def test_prediction_record_is_pending_and_sorted_longest_first(): + record = prediction_record( + make_rule(), + [make_prediction("MIT", 5, 5, 0.8), make_prediction("MIT License", 5, 6, 0.9)], + truncated=True, + ) + + assert [phrase["text"] for phrase in record["phrases"]] == ["MIT License", "MIT"] + assert all(phrase["decision"] == PENDING for phrase in record["phrases"]) + assert record["truncated"] is True + assert record["text_sha256"] == text_sha256(TEXT) + + +def test_prediction_record_keeps_the_strongest_repeated_phrase(): + record = prediction_record( + make_rule(text="MIT License and MIT License"), + [make_prediction("MIT License", 0, 1, 0.6), make_prediction("MIT License", 3, 4, 0.9)], + truncated=False, + ) + + assert len(record["phrases"]) == 1 + assert record["phrases"][0]["confidence"] == 0.9 + assert record["phrases"][0]["start_word"] == 3 + + +def test_review_file_round_trip_and_atomic_replacement(tmp_path): + path = tmp_path / "review.jsonl" + records = [make_record()] + + write_review_file(path, records) + first = path.read_bytes() + write_review_file(path, read_review_file(path)) + + assert path.read_bytes() == first + assert not list(tmp_path.glob("*.tmp")) + + +def test_failed_review_file_replacement_keeps_existing_file(tmp_path, monkeypatch): + path = tmp_path / "review.jsonl" + write_review_file(path, [make_record()]) + before = path.read_bytes() + monkeypatch.setattr(review_module.os, "replace", lambda *args: (_ for _ in ()).throw(OSError())) + + with pytest.raises(OSError): + write_review_file(path, [make_record()]) + + assert path.read_bytes() == before + assert not list(tmp_path.glob("*.tmp")) + + +@pytest.mark.parametrize( + "change,error", + [ + (lambda record: record.pop("truncated"), "record fields"), + (lambda record: record.update(identifier="../mit.RULE"), "rule filename"), + (lambda record: record.update(text_sha256="bad"), "text_sha256"), + (lambda record: record.update(truncated=1), "boolean"), + (lambda record: record.update(phrases=[]), "non-empty list"), + (lambda record: record["phrases"][0].update(confidence=float("nan")), "confidence"), + (lambda record: record["phrases"][0].update(decision="auto"), "decision"), + ], +) +def test_validate_record_rejects_invalid_data(change, error): + record = make_record() + change(record) + + with pytest.raises(click.ClickException, match=error): + validate_record(record, "review.jsonl", 1) + + +def test_read_review_file_rejects_malformed_json_and_duplicate_rules(tmp_path): + malformed = tmp_path / "malformed.jsonl" + malformed.write_text("{bad}\n", encoding="utf-8") + with pytest.raises(click.ClickException, match="line 1"): + read_review_file(malformed) + + duplicate = tmp_path / "duplicate.jsonl" + record = make_record() + duplicate.write_text(json.dumps(record) + "\n" + json.dumps(record) + "\n") + with pytest.raises(click.ClickException, match="duplicate rule"): + read_review_file(duplicate) + + +def test_load_current_rule_requires_same_expression_and_text(tmp_path, monkeypatch): + use_rules_directory(monkeypatch, tmp_path) + rule = make_rule() + rule.dump(str(tmp_path)) + record = make_record(rule) + + loaded, reason = review_module.load_current_rule(record) + assert loaded.identifier == rule.identifier + assert reason is None + + record["license_expression"] = "apache-2.0" + assert review_module.load_current_rule(record)[1] == "license expression changed" + record["license_expression"] = "mit" + record["text_sha256"] = "0" * 64 + assert review_module.load_current_rule(record)[1] == "rule text changed" + + +def test_preview_is_exact_and_does_not_mutate_rule(): + rule = make_rule(source="existing") + + changed, preview = review_module.preview_injection(rule, "MIT License") + + assert changed + assert "{{MIT License}}" in preview + assert rule.text == TEXT + assert rule.source == "existing" + + +class FakePredictor: + def predict(self, text): + return PredictionResult( + words=tuple(text.split()), + phrases=( + make_prediction(), + make_prediction("is", 1, 1, 0.99), + ), + truncated=True, + ) + + +def test_predict_records_validates_candidates_and_honors_limit(monkeypatch): + rules = [make_rule(identifier=f"mit_{index}.RULE") for index in range(3)] + + monkeypatch.setattr(review_module, "predict_rule", lambda *args: FakePredictor().predict(args[-1])) + + records, counts = predict_records( + selected={"mit": rules}, + tagger=object(), + tokenizer=object(), + max_length=512, + limit=2, + ) + + assert [record["identifier"] for record in records] == ["mit_0.RULE", "mit_1.RULE"] + assert counts == { + "rules": 2, + "truncated": 2, + "rejected": 2, + "not_found": 0, + } + assert all(len(record["phrases"]) == 1 for record in records) + + +def test_predict_command_writes_only_valid_candidates(tmp_path, monkeypatch): + review_file = tmp_path / "review.jsonl" + rule = make_rule() + monkeypatch.setattr(review_module, "select_rules", lambda **kwargs: {"mit": [rule]}) + monkeypatch.setattr( + review_module, + "load_model", + lambda *args, **kwargs: (object(), object(), 512), + ) + monkeypatch.setattr(review_module, "predict_rule", lambda *args: FakePredictor().predict(args[-1])) + + result = CliRunner().invoke( + review_model_required_phrases, + ["predict", "--model", "unused", "--review-file", str(review_file)], + ) + + assert result.exit_code == 0, result.output + records = read_review_file(review_file) + assert [phrase["text"] for phrase in records[0]["phrases"]] == ["MIT License"] + assert records[0]["phrases"][0]["decision"] == PENDING + assert "rejected : 1" in result.output + assert "truncated : 1" in result.output + + +def test_predict_checks_selection_before_loading_model(tmp_path, monkeypatch): + monkeypatch.setattr(review_module, "select_rules", lambda **kwargs: {}) + monkeypatch.setattr( + review_module, + "load_model", + lambda *args, **kwargs: pytest.fail("model must not load"), + ) + + result = CliRunner().invoke( + review_model_required_phrases, + ["predict", "--model", "unused", "--review-file", str(tmp_path / "review.jsonl")], + ) + + assert result.exit_code == 0 + assert "No eligible rules" in result.output + + +def test_predict_refuses_to_replace_an_existing_review_file(tmp_path, monkeypatch): + path = tmp_path / "review.jsonl" + path.write_text("keep\n", encoding="utf-8") + monkeypatch.setattr( + review_module, + "select_rules", + lambda **kwargs: pytest.fail("selection must not run"), + ) + + result = CliRunner().invoke( + review_model_required_phrases, + ["predict", "--model", "unused", "--review-file", str(path)], + ) + + assert result.exit_code != 0 + assert path.read_text() == "keep\n" + + +def prepare_review(tmp_path, monkeypatch, phrases=None): + use_rules_directory(monkeypatch, tmp_path) + rule = make_rule() + rule.dump(str(tmp_path)) + path = tmp_path / "review.jsonl" + record = make_record(rule, phrases) + write_review_file(path, [record]) + return path + + +def test_review_approves_and_rejects_predictions_resumably(tmp_path, monkeypatch): + path = prepare_review( + tmp_path, + monkeypatch, + [make_prediction(), make_prediction("do things", 8, 9, 0.7)], + ) + + first = CliRunner().invoke( + review_model_required_phrases, + ["review", "--review-file", str(path)], + input="y\nq\n", + ) + assert first.exit_code == 0, first.output + assert [phrase["decision"] for phrase in read_review_file(path)[0]["phrases"]] == [ + APPROVED, + PENDING, + ] + + second = CliRunner().invoke( + review_model_required_phrases, + ["review", "--review-file", str(path)], + input="n\n", + ) + assert second.exit_code == 0, second.output + assert [phrase["decision"] for phrase in read_review_file(path)[0]["phrases"]] == [ + APPROVED, + REJECTED, + ] + + +def test_review_edits_a_phrase_and_preserves_prediction(tmp_path, monkeypatch): + path = prepare_review(tmp_path, monkeypatch) + + result = CliRunner().invoke( + review_model_required_phrases, + ["review", "--review-file", str(path)], + input="e\nMIT License to\n", + ) + + assert result.exit_code == 0, result.output + phrase = read_review_file(path)[0]["phrases"][0] + assert phrase["text"] == "MIT License to" + assert phrase["predicted_text"] == "MIT License" + assert phrase["decision"] == APPROVED + + +def test_review_leaves_a_stale_record_pending(tmp_path, monkeypatch): + path = prepare_review(tmp_path, monkeypatch) + rule_path = tmp_path / "mit_test.RULE" + rule_path.write_text(rule_path.read_text().replace("Permission", "Permission now")) + + result = CliRunner().invoke( + review_model_required_phrases, + ["review", "--review-file", str(path)], + ) + + assert result.exit_code == 0 + assert "rule text changed" in result.output + assert read_review_file(path)[0]["phrases"][0]["decision"] == PENDING + + +def test_apply_refuses_pending_predictions(tmp_path, monkeypatch): + path = prepare_review(tmp_path, monkeypatch) + + result = CliRunner().invoke(apply, ["--review-file", str(path)]) + + assert result.exit_code != 0 + assert "still need review" in result.output + assert "{{" not in Rule.from_file(str(tmp_path / "mit_test.RULE")).text + + +def test_apply_writes_only_approved_phrases_once(tmp_path, monkeypatch): + path = prepare_review( + tmp_path, + monkeypatch, + [make_prediction(), make_prediction("do things", 8, 9, 0.7)], + ) + records = read_review_file(path) + records[0]["phrases"][0]["decision"] = APPROVED + records[0]["phrases"][1]["decision"] = REJECTED + write_review_file(path, records) + + result = CliRunner().invoke(apply, ["--review-file", str(path)]) + + assert result.exit_code == 0, result.output + saved = Rule.from_file(str(tmp_path / "mit_test.RULE")) + assert "{{MIT License}}" in saved.text + assert "{{do things}}" not in saved.text + assert saved.source == "ml_model" + assert "rules written : 1" in result.output + + +def test_apply_dry_run_does_not_write(tmp_path, monkeypatch): + path = prepare_review(tmp_path, monkeypatch) + records = read_review_file(path) + records[0]["phrases"][0]["decision"] = APPROVED + write_review_file(path, records) + rule_path = tmp_path / "mit_test.RULE" + before = rule_path.read_bytes() + + result = CliRunner().invoke(apply, ["--review-file", str(path), "--dry-run"]) + + assert result.exit_code == 0, result.output + assert rule_path.read_bytes() == before + assert "Dry run" in result.output + + +def test_apply_refuses_a_stale_rule_without_mutation(tmp_path, monkeypatch): + path = prepare_review(tmp_path, monkeypatch) + records = read_review_file(path) + records[0]["phrases"][0]["decision"] = APPROVED + write_review_file(path, records) + rule_path = tmp_path / "mit_test.RULE" + rule_path.write_text(rule_path.read_text().replace("Permission", "Changed")) + before = rule_path.read_bytes() + + result = CliRunner().invoke(apply, ["--review-file", str(path), "--verbose"]) + + assert result.exit_code != 0 + assert rule_path.read_bytes() == before + assert "stale review record" in result.output + + +def test_apply_preflights_every_rule_before_writing(tmp_path, monkeypatch): + use_rules_directory(monkeypatch, tmp_path) + first = make_rule(identifier="mit_first.RULE") + second = make_rule(identifier="mit_second.RULE") + first.dump(str(tmp_path)) + second.dump(str(tmp_path)) + records = [make_record(first), make_record(second)] + for record in records: + record["phrases"][0]["decision"] = APPROVED + path = tmp_path / "review.jsonl" + write_review_file(path, records) + second_path = tmp_path / second.identifier + second_path.write_text(second_path.read_text().replace("Permission", "Changed")) + first_path = tmp_path / first.identifier + before = first_path.read_bytes() + + result = CliRunner().invoke(apply, ["--review-file", str(path)]) + + assert result.exit_code != 0 + assert first_path.read_bytes() == before + assert "stale review record" in result.output + + +def test_command_group_exposes_all_stages(): + result = CliRunner().invoke(review_model_required_phrases, ["--help"]) + + assert result.exit_code == 0 + assert "predict" in result.output + assert "review" in result.output + assert "apply" in result.output