Add script to extract required phrases from rule files

etc/scripts/dataset_pipeline/build_dataset.py

@@ -0,0 +1,139 @@
+# extracts required phrases from .RULE files
+# outputs a JSONL dataset for NER model training
+import json
+import re
+import unicodedata
+from pathlib import Path
+import click
+
+from licensedcode.models import Rule
+from licensedcode.models import rules_data_dir as default_rules_data_dir
+
+MARKER_RE = re.compile(r'\{\{([^}]*)\}\}', re.DOTALL)
+
+
+def extract_markers(text):
+    """Pull out all {{ }} markers and compute char positions
+    relative to plain text (with markers removed)"""
+    markers = []
+    offset = 0
+    for m in MARKER_RE.finditer(text):
+        phrase = m.group(1)
+        start = m.start() - offset
+        end = start + len(phrase)
+        markers.append({'phrase': phrase, 'start': start, 'end': end})
+        offset += 4  # account for removed {{ and }}
+    return markers
+
+
+def strip_markers(text):
+    """Remove {{ }} but keep text inside"""
+    return MARKER_RE.sub(lambda m: m.group(1), text)
+
+
+def normalize_phrase(phrase):
+    """Clean raw marker phrase for training"""
+    result = phrase
+    # replace html entities
+    result = result.replace('"', '"').replace('&', '&')
+    result = result.replace('&lt;', '<').replace('&gt;', '>')
+    # strip xml tags like <name>,</license>
+    result = re.sub(r'<[^>]+>', '', result)
+    # remove markdown backticks
+    result = result.replace('`', '')
+    # collapse whitespace and trim
+    result = re.sub(r'\s+', ' ', result).strip()
+    # strip trailing/leading punct thats not meaningful
+    result = result.strip('.,;:>')
+    return result
+
+
+@click.command()
+@click.option('--rules-dir', type=click.Path(exists=True), default=None,
+              help='Path to rules directory (defaults to repo rules dir)')
+@click.option('--output', default='dataset-output/required_phrases.jsonl',
+              help='Output JSONL path')
+def main(rules_dir, output):
+    """Extract required phrases from rule files for NER training"""
+    if not rules_dir:
+        repo_rules = Path(__file__).resolve().parents[3] / 'src' / 'licensedcode' / 'data' / 'rules'
+        rules_dir = str(repo_rules) if repo_rules.is_dir() else default_rules_data_dir
+
+    rules_path = Path(rules_dir)
+    out_path = Path(output)
+    out_path.parent.mkdir(parents=True, exist_ok=True)
+
+    total_rules = 0
+    annotated = 0
+    total_phrases = 0
+    results = []
+
+    click.echo(f'scanning rules from: {rules_path}')
+    for rf in sorted(rules_path.glob('*.RULE')):
+        try:
+            rule = Rule.from_file(rule_file=str(rf))
+        except Exception:
+            continue
+        total_rules += 1
+
+        # skip is_required_phrase files,theyre just the phrase itself
+        if getattr(rule, 'is_required_phrase', False):
+            continue
+        text = rule.text or ''
+        if not MARKER_RE.search(text):
+            continue
+
+        # normalize line endings and unicode
+        text = text.replace('\r\n', '\n').replace('\r', '\n')
+        text = unicodedata.normalize('NFKC', text)
+
+        markers = extract_markers(text)
+        plain_text = strip_markers(text)
+        # validate positions and normalize each phrase
+        valid_markers = []
+        for m in markers:
+            if m['start'] < 0 or m['end'] > len(plain_text):
+                continue
+            actual = plain_text[m['start']:m['end']]
+            if actual != m['phrase']:
+                continue
+            normalized = normalize_phrase(m['phrase'])
+            if not normalized:
+                continue
+            valid_markers.append({
+                'phrase': m['phrase'],
+                'phrase_normalized': normalized,
+                'start': m['start'],
+                'end': m['end'],
+            })
+
+        if not valid_markers:
+            continue
+        annotated += 1
+        total_phrases += len(valid_markers)
+        results.append({
+            'identifier': rule.identifier,
+            'license_expression': rule.license_expression or '',
+            'required_phrases': valid_markers,
+        })
+
+    # write jsonl
+    with open(out_path, 'w', encoding='utf-8') as f:
+        for entry in results:
+            f.write(json.dumps(entry, ensure_ascii=False) + '\n')
+
+    click.echo('\ndone')
+    click.echo(f'  rules scanned: {total_rules}')
+    click.echo(f'  annotated: {annotated}')
+    click.echo(f'  phrases extracted: {total_phrases}')
+    click.echo(f'  output: {out_path}')
+
+
+# stuff to do(follow up commits):
+# - add plain text field to output (whole rule text)
+# - BIOES labels
+# - train/val/test split by license expression
+# - additional fields (rule_type,etc)
+
+if __name__ == '__main__':
+    main()