Tree-Sitter Example Parser

ts-parse-examples.py

import glob
import subprocess
import os
import shutil
import re
import sys
import argparse
from concurrent.futures import ProcessPoolExecutor, as_completed
from ruamel.yaml import YAML

def parse_worker(executable, file_path, is_exclusion_mode):
    """Worker: Returns (path, msg, was_success)"""
    result = subprocess.run([executable, "parse", file_path], capture_output=True, text=True)
    
    if result.returncode != 0 or "(ERROR" in result.stdout:
        match = re.search(r"\(ERROR\s+\[(\d+),\s+(\d+)\]", result.stdout)
        line_info = f":{int(match.group(1))+1}:{int(match.group(2))+1}" if match else ""
        
        try:
            with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
                content = f.readlines()
                line_idx = int(match.group(1)) if match else -1
                code = f"\n  >> {content[line_idx].strip()}" if 0 <= line_idx < len(content) else ""
        except: code = ""
            
        return file_path, f"FAILURE: {file_path}{line_info}{code}", False
    
    success_msg = f"SUCCESS: {file_path} (Now passes!)" if is_exclusion_mode else None
    return file_path, success_msg, True

def run_local_parse():
    parser = argparse.ArgumentParser()
    parser.add_argument('--only-excluded', action='store_true', help="Test ONLY the '!' files.")
    parser.add_argument('--append-failures-to-exclude', action='store_true', help="Add new failures to CI YAML.")
    args = parser.parse_args()

    # Configure YAML for "Round-Trip" preservation
    yaml_io = YAML()
    yaml_io.preserve_quotes = True
    yaml_io.width = 4096  # Prevent long lines from wrapping
    # Standard GitHub Actions Indentation (2 spaces for keys, 4 for list items)
    yaml_io.indent(mapping=2, sequence=4, offset=2) 
    
    workflow_path = ".github/workflows/ci.yml"
    if not os.path.exists(workflow_path):
        return print(f"File not found: {workflow_path}")

    with open(workflow_path, 'r') as f:
        config = yaml_io.load(f)

    # 1. Find the tree-sitter action block
    target_step = None
    for job in config.get('jobs', {}).values():
        for step in job.get('steps', []):
            if 'tree-sitter/parse-action' in str(step.get('uses', '')):
                target_step = step
                break

    if not target_step: return print("CI step not found.")

    files_string = target_step['with']['files']
    lines = [l.strip() for l in files_string.split('\n') if l.strip()]
    inclusions = [l for l in lines if not l.startswith('!')]
    exclusions = [os.path.normpath(l.lstrip('!')) for l in lines if l.startswith('!')]

    # 2. Select file set
    if args.only_excluded:
        final_files = [f for f in exclusions if os.path.isfile(f)]
        print(f"Testing {len(final_files)} PREVIOUSLY EXCLUDED files...\n")
    else:
        all_files = set()
        for pat in inclusions: 
            all_files.update(glob.glob(pat, recursive=True))
        final_files = [f for f in all_files if not any(exc in os.path.normpath(f) for exc in exclusions) and os.path.isfile(f)]
        print(f"Testing {len(final_files)} standard files...\n")

    executable = shutil.which("tree-sitter")
    if not executable: return print("tree-sitter executable not found.")

    failed_paths = []
    
    # 3. Parallel Execution
    cpu_count = os.cpu_count() or 4
    with ProcessPoolExecutor(max_workers=cpu_count) as executor:
        futures = {executor.submit(parse_worker, executable, f, args.only_excluded): f for f in final_files}
        for future in as_completed(futures) :
            path, msg, was_success = future.result()
            if msg: print(msg)
            if not was_success: failed_paths.append(path)

    # 4. Handle YAML Update
    if args.append_failures_to_exclude and failed_paths:
        # Convert to forward slashes for CI compatibility
        new_entries = [f"!{p.replace(os.sep, '/')}" for p in failed_paths]
        
        current_lines = files_string.strip().split('\n')
        header_inclusions = [l for l in current_lines if not l.startswith('!')]
        existing_exclusions = [l for l in current_lines if l.startswith('!')]
        
        # Merge, deduplicate, and sort
        updated_exclusions = sorted(list(set(existing_exclusions + new_entries)))
        
        # Update the YAML object directly
        # Joining with newlines and ensuring a trailing newline for the scalar block
        target_step['with']['files'] = '\n'.join(header_inclusions + updated_exclusions) + '\n'
        
        with open(workflow_path, 'w') as f:
            yaml_io.dump(config, f)
        print(f"\nUpdated {workflow_path} with {len(failed_paths)} new exclusions (Sorted alphabetically).")

if __name__ == "__main__":
    run_local_parse()