1#!/usr/bin/env nix-shell
  2#! nix-shell -i python3 -p python3 nurl
  3"""
  4Generate grammar information for Helix editor by parsing languages.toml
  5and fetching source information using nurl in parallel.
  6"""
  7
  8import argparse
  9import asyncio
 10import json
 11import os
 12import sys
 13import tomllib
 14from dataclasses import dataclass
 15from pathlib import Path
 16from typing import Any
 17
 18
 19@dataclass
 20class Grammar:
 21    name: str
 22    git_url: str
 23    rev: str
 24    subpath: str | None = None
 25
 26
 27async def run_nurl(url: str, rev: str, semaphore: asyncio.Semaphore) -> dict[str, Any]:
 28    """Run nurl command for a single grammar and return parsed JSON output."""
 29    async with semaphore:
 30        proc = await asyncio.create_subprocess_exec(
 31            "nurl",
 32            url,
 33            rev,
 34            "--json",
 35            stdout=asyncio.subprocess.PIPE,
 36            stderr=asyncio.subprocess.PIPE,
 37        )
 38        stdout, stderr = await proc.communicate()
 39
 40        if proc.returncode != 0:
 41            raise RuntimeError(f"nurl failed for {url}@{rev}: {stderr.decode()}")
 42
 43        return json.loads(stdout.decode())
 44
 45
 46def parse_languages_toml(toml_path: Path) -> list[Grammar]:
 47    """Parse languages.toml and extract grammar information."""
 48    with open(toml_path, "rb") as f:
 49        config = tomllib.load(f)
 50
 51    grammars = []
 52    for grammar in config.get("grammar", []):
 53        if "source" not in grammar:
 54            continue
 55
 56        source = grammar["source"]
 57        if "git" not in source or "rev" not in source:
 58            continue
 59
 60        grammars.append(
 61            Grammar(
 62                name=grammar["name"].replace("_", "-"),
 63                git_url=source["git"],
 64                rev=source["rev"],
 65                subpath=source.get("subpath"),
 66            )
 67        )
 68
 69    return grammars
 70
 71
 72async def fetch_all_grammars(
 73    grammars: list[Grammar], max_parallel: int
 74) -> dict[str, Any]:
 75    """Fetch nurl information for all grammars in parallel."""
 76    semaphore = asyncio.Semaphore(max_parallel)
 77    results = {}
 78    total = len(grammars)
 79    completed = 0
 80
 81    tasks = []
 82    for grammar in grammars:
 83        task = run_nurl(grammar.git_url, grammar.rev, semaphore)
 84        tasks.append((grammar, task))
 85
 86    for grammar, task in tasks:
 87        try:
 88            result = await task
 89            results[grammar.name] = {
 90                "nurl": result,
 91                "subpath": grammar.subpath,
 92            }
 93            completed += 1
 94            print(f"[{completed}/{total}] ✓ {grammar.name}", file=sys.stderr)
 95        except Exception as e:
 96            completed += 1
 97            print(f"[{completed}/{total}] ✗ {grammar.name}: {e}", file=sys.stderr)
 98            results[grammar.name] = {"error": str(e)}
 99
100    return results
101
102
103async def main():
104    parser = argparse.ArgumentParser(
105        description="Generate grammar information for Helix editor"
106    )
107    parser.add_argument(
108        "languages_toml",
109        type=Path,
110        help="path to languages.toml from Helix repository",
111    )
112    parser.add_argument(
113        "-o",
114        "--output",
115        type=Path,
116        default=Path("grammars.json"),
117        help="output JSON file (default: grammars.json)",
118    )
119    parser.add_argument(
120        "-j",
121        "--jobs",
122        type=int,
123        default=os.cpu_count(),
124        help=f"number of parallel nurl instances (default: {os.cpu_count()})",
125    )
126
127    args = parser.parse_args()
128
129    if not args.languages_toml.exists():
130        print(f"Error: {args.languages_toml} not found", file=sys.stderr)
131        sys.exit(1)
132
133    print(f"Parsing {args.languages_toml}...", file=sys.stderr)
134    grammars = parse_languages_toml(args.languages_toml)
135    print(f"Found {len(grammars)} grammars", file=sys.stderr)
136
137    print(f"Fetching grammar information ({args.jobs} parallel jobs)...", file=sys.stderr)
138    results = await fetch_all_grammars(grammars, args.jobs)
139
140    errors = [name for name, data in results.items() if "error" in data]
141    if errors:
142        print(f"\nFailed grammars ({len(errors)}):", file=sys.stderr)
143        for name in errors:
144            print(f"  - {name}: {results[name]['error']}", file=sys.stderr)
145        sys.exit(1)
146
147    with open(args.output, "w") as f:
148        json.dump(results, f, indent=2)
149        f.write('\n')
150
151    print(f"\nResults written to {args.output}", file=sys.stderr)
152
153
154if __name__ == "__main__":
155    asyncio.run(main())