Blablablab/audio-classification

1"""2Trace Converter CLI3 4Command-line interface for converting agent traces from various formats5to Potato's canonical JSONL format.6 7Usage:8    python -m potato.trace_converter --input traces.json --input-format react --output data.jsonl9    python -m potato.trace_converter --input traces.json --auto-detect --output data.jsonl10    python -m potato.trace_converter --list-formats11"""12 13import argparse14import json15import logging16import sys17from pathlib import Path18 19from .registry import converter_registry20 21logger = logging.getLogger(__name__)22 23 24def parse_args(args=None):25    parser = argparse.ArgumentParser(26        prog="potato-trace-convert",27        description="Convert agent traces from various formats to Potato's canonical JSONL format."28    )29 30    parser.add_argument(31        "--input", "-i",32        help="Input file path (JSON, JSONL, or Parquet)"33    )34    parser.add_argument(35        "--input-format", "-f",36        help="Input format name (e.g., react, langchain, langfuse, atif, webarena, openai, anthropic, swebench, otel, multi_agent, mcp)"37    )38    parser.add_argument(39        "--output", "-o",40        help="Output file path (JSONL). Defaults to stdout."41    )42    parser.add_argument(43        "--auto-detect",44        action="store_true",45        help="Auto-detect the input format"46    )47    parser.add_argument(48        "--list-formats",49        action="store_true",50        help="List all supported formats and exit"51    )52    parser.add_argument(53        "--pretty",54        action="store_true",55        help="Pretty-print JSON output (one object per line, indented)"56    )57    parser.add_argument(58        "--verbose", "-v",59        action="store_true",60        help="Enable verbose logging"61    )62 63    return parser.parse_args(args)64 65 66def load_input(file_path: str):67    """Load input data from JSON, JSONL, or Parquet file."""68    path = Path(file_path)69    if not path.exists():70        raise FileNotFoundError(f"Input file not found: {file_path}")71 72    # Handle Parquet files73    if path.suffix.lower() == ".parquet":74        import pyarrow.parquet as pq75        table = pq.read_table(str(path))76        return table.to_pandas().to_dict("records")77 78    content = path.read_text(encoding="utf-8").strip()79 80    # Try parsing as JSON first81    try:82        return json.loads(content)83    except json.JSONDecodeError:84        pass85 86    # Try parsing as JSONL (one JSON object per line)87    records = []88    for line_num, line in enumerate(content.splitlines(), 1):89        line = line.strip()90        if not line:91            continue92        try:93            records.append(json.loads(line))94        except json.JSONDecodeError as e:95            raise ValueError(f"Invalid JSON on line {line_num}: {e}")96    return records97 98 99def main(args=None):100    parsed = parse_args(args)101 102    if parsed.verbose:103        logging.basicConfig(level=logging.DEBUG)104    else:105        logging.basicConfig(level=logging.WARNING)106 107    # List formats108    if parsed.list_formats:109        print("Supported trace formats:")110        print()111        for info in converter_registry.list_converters():112            print(f"  {info['format_name']:15s} {info['description']}")113            if info.get('file_extensions'):114                print(f"  {'':15s} Extensions: {', '.join(info['file_extensions'])}")115            print()116        return 0117 118    # Validate arguments119    if not parsed.input:120        print("Error: --input is required (or use --list-formats)", file=sys.stderr)121        return 1122 123    # Load input124    try:125        data = load_input(parsed.input)126    except (FileNotFoundError, ValueError) as e:127        print(f"Error loading input: {e}", file=sys.stderr)128        return 1129 130    # Determine format131    format_name = parsed.input_format132    if not format_name:133        if parsed.auto_detect:134            format_name = converter_registry.detect_format(data)135            if not format_name:136                print("Error: Could not auto-detect input format. "137                      "Please specify with --input-format.", file=sys.stderr)138                return 1139            print(f"Auto-detected format: {format_name}", file=sys.stderr)140        else:141            print("Error: --input-format or --auto-detect is required", file=sys.stderr)142            return 1143 144    # Convert145    try:146        traces = converter_registry.convert(format_name, data)147    except ValueError as e:148        print(f"Error: {e}", file=sys.stderr)149        return 1150    except Exception as e:151        print(f"Conversion error: {e}", file=sys.stderr)152        return 1153 154    # Output155    output_lines = []156    for trace in traces:157        trace_dict = trace.to_dict()158        if parsed.pretty:159            output_lines.append(json.dumps(trace_dict, ensure_ascii=False, indent=2))160        else:161            output_lines.append(json.dumps(trace_dict, ensure_ascii=False))162 163    output_text = "\n".join(output_lines) + "\n"164 165    if parsed.output:166        Path(parsed.output).parent.mkdir(parents=True, exist_ok=True)167        Path(parsed.output).write_text(output_text, encoding="utf-8")168        print(f"Converted {len(traces)} traces to {parsed.output}", file=sys.stderr)169    else:170        sys.stdout.write(output_text)171 172    return 0173 174 175if __name__ == "__main__":176    sys.exit(main())177