python_pipeline.py

"""Run a bounded JSONL query from Python; offline unless --live is explicit."""

import argparse
import json
from pathlib import Path
import subprocess


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--live", action="store_true", help="send synthetic records to the configured API")
    parser.add_argument("--cli", help="installed intentgrep executable; defaults to this package's built CLI")
    args = parser.parse_args()

    package_root = Path(__file__).resolve().parent.parent
    executable = [args.cli] if args.cli else ["node", str(package_root / "dist" / "cli.js")]
    # Real applications must select records for their authenticated caller first.
    records = [
        {"id": "failed", "text": "The deployment exited with code 1 before changing production."},
        {"id": "succeeded", "text": "The deployment finished successfully."},
    ]
    command = executable + [
        "--json" if args.live else "--plan",
        "--task", "filter",
        "--profile", "generic",
        "--input", "jsonl",
        "--max-requests", "8",
        "--max-input-bytes", "80000",
        "--timeout", "10000",
        "--retries", "1",
        "The record explicitly describes a failed deployment",
    ]
    payload = "".join(json.dumps(record, ensure_ascii=False) + "\n" for record in records)
    # API credentials are inherited from the environment, never command arguments.
    process = subprocess.run(command, input=payload, text=True, capture_output=True, timeout=150)
    if process.returncode not in (0, 1):
        raise SystemExit(process.stderr.strip() or f"intentgrep failed with exit code {process.returncode}")
    # A live no-match result (exit 1) is still valid JSON, not an API failure.
    result = json.loads(process.stdout)
    print(json.dumps({"mode": "live" if args.live else "offline-plan", "output": result}, indent=2))


if __name__ == "__main__":
    main()