dataforge-cli/dataforge/commands.py

"""CLI commands for DataForge CLI."""

import glob
import os
from pathlib import Path
from typing import List, Optional

import click

from .parsers import (
    detect_format,
    dump_data,
    load_data,
    SUPPORTED_FORMATS,
)
from .validator import SchemaValidator
from .type_check import validate_types, infer_schema_from_data


def find_files(
    pattern: str,
    recursive: bool = False,
    directory: Optional[str] = None,
) -> List[str]:
    """Find files matching a pattern."""
    search_dir = directory or "."
    if recursive:
        pattern = os.path.join(search_dir, "**", pattern)
        files = glob.glob(pattern, recursive=True)
    else:
        pattern = os.path.join(search_dir, pattern)
        files = glob.glob(pattern)
    return sorted(files)


def resolve_format(fmt: Optional[str], file_path: str) -> str:
    """Resolve format from explicit option or file extension."""
    if fmt:
        if fmt not in SUPPORTED_FORMATS:
            raise click.BadParameter(
                f"Unsupported format: {fmt}. Supported: {', '.join(SUPPORTED_FORMATS)}",
                param_hint="--format",
            )
        return fmt
    return detect_format(file_path)


@click.command()
@click.argument("input_file", type=click.Path(exists=True))
@click.argument("output_file", type=click.Path())
@click.option(
    "--from", "-f", "from_format", help="Input format (json, yaml, toml)"
)
@click.option(
    "--to",
    "-t",
    "to_format",
    required=True,
    help="Output format (json, yaml, toml)",
)
@click.option("--indent", "-i", default=2, help="Indentation spaces (0 for compact)")
@click.option("--quiet", "-q", is_flag=True, help="Minimal output")
def convert(
    input_file: str,
    output_file: str,
    from_format: Optional[str],
    to_format: str,
    indent: int,
    quiet: bool,
) -> None:
    """Convert a file from one format to another.

    INPUT_FILE: Input file path (or - for stdin)

    OUTPUT_FILE: Output file path (or - for stdout)
    """
    try:
        if to_format not in SUPPORTED_FORMATS:
            raise click.BadParameter(
                f"Unsupported format: {to_format}. Supported: {', '.join(SUPPORTED_FORMATS)}",
                param_hint="--to",
            )

        if input_file == "-":
            import sys

            content = sys.stdin.read()
            input_format = from_format or "json"
            data = load_data(content, input_format)
        else:
            input_format = resolve_format(from_format, input_file)
            data = load_data(input_file, input_format)

        output_format = to_format
        result = dump_data(data, output_format, indent=indent if indent > 0 else None)

        if output_file == "-":
            import sys

            sys.stdout.write(result)
        else:
            with open(output_file, "w", encoding="utf-8") as f:
                f.write(result)

        if not quiet:
            click.echo(f"Successfully converted {input_file} to {output_file}")

    except Exception as e:
        raise click.ClickException(str(e))


@click.command()
@click.argument("input_files", nargs=-1, type=click.Path(exists=True))
@click.option(
    "--from", "-f", "from_format", help="Input format (json, yaml, toml)"
)
@click.option(
    "--to",
    "-t",
    "to_format",
    required=True,
    help="Output format (json, yaml, toml)",
)
@click.option("--output-dir", "-o", default=".", help="Output directory for converted files")
@click.option("--indent", "-i", default=2, help="Indentation spaces")
@click.option(
    "--pattern",
    "-p",
    default="*.{json,yaml,yml,toml}",
    help="File pattern for batch processing",
)
@click.option("--recursive", "-r", is_flag=True, help="Search recursively")
@click.option("--quiet", "-q", is_flag=True, help="Minimal output")
def batch_convert(
    input_files: tuple,
    from_format: Optional[str],
    to_format: str,
    output_dir: str,
    indent: int,
    pattern: str,
    recursive: bool,
    quiet: bool,
) -> None:
    """Convert multiple files from one format to another.

    INPUT_FILES: Input file paths (optional, uses --pattern if not provided)
    """
    try:
        if to_format not in SUPPORTED_FORMATS:
            raise click.BadParameter(
                f"Unsupported format: {to_format}. Supported: {', '.join(SUPPORTED_FORMATS)}",
                param_hint="--to",
            )

        os.makedirs(output_dir, exist_ok=True)

        files = list(input_files) if input_files else find_files(pattern, recursive)

        if not files:
            if not quiet:
                click.echo("No files found matching the pattern")
            return

        converted = 0
        errors = 0

        for file_path in files:
            try:
                input_format = resolve_format(from_format, file_path)
                data = load_data(file_path, input_format)

                output_filename = Path(file_path).stem + f".{to_format}"
                output_file = os.path.join(output_dir, output_filename)
                dump_data(data, to_format, output_file, indent=indent if indent > 0 else None)
                converted += 1
            except Exception as e:
                errors += 1
                if not quiet:
                    click.echo(f"Error converting {file_path}: {e}")

        if not quiet:
            click.echo(f"Converted {converted} files, {errors} errors")
        elif errors > 0:
            click.echo(f"Converted {converted}, errors: {errors}", err=True)

    except Exception as e:
        raise click.ClickException(str(e))


@click.command()
@click.argument("input_file", type=click.Path(exists=True))
@click.option("--schema", "-s", "schema_file", help="Path to JSON Schema file")
@click.option("--strict", is_flag=True, help="Strict validation mode")
@click.option("--quiet", "-q", is_flag=True, help="Minimal output")
def validate(
    input_file: str,
    schema_file: Optional[str],
    strict: bool,
    quiet: bool,
) -> None:
    """Validate a file against a JSON Schema.

    INPUT_FILE: Input file path (or - for stdin)
    """
    try:
        if input_file == "-":
            import sys

            content = sys.stdin.read()
            input_format = "json"
            data = load_data(content, input_format)
        else:
            input_format = detect_format(input_file)
            data = load_data(input_file, input_format)

        if schema_file:
            validator = SchemaValidator(schema_file=schema_file)
            errors = validator.validate(data)

            if errors:
                if not quiet:
                    for error in validator.get_error_messages(errors):
                        click.echo(error)
                raise click.ClickException(
                    f"Validation failed with {len(errors)} error(s)"
                )
            else:
                if not quiet:
                    click.echo("Validation passed")
        else:
            if not quiet:
                click.echo(f"File is valid {input_format}")

    except Exception as e:
        raise click.ClickException(str(e))


@click.command()
@click.argument("input_files", nargs=-1, type=click.Path(exists=True))
@click.option(
    "--schema", "-s", "schema_file", required=True, help="Path to JSON Schema file"
)
@click.option(
    "--pattern",
    "-p",
    default="*.{json,yaml,yml,toml}",
    help="File pattern for batch processing",
)
@click.option("--recursive", "-r", is_flag=True, help="Search recursively")
@click.option("--quiet", "-q", is_flag=True, help="Minimal output")
def batch_validate(
    input_files: tuple,
    schema_file: str,
    pattern: str,
    recursive: bool,
    quiet: bool,
) -> None:
    """Validate multiple files against a JSON Schema.

    INPUT_FILES: Input file paths (optional, uses --pattern if not provided)
    """
    try:
        files = list(input_files) if input_files else find_files(pattern, recursive)

        if not files:
            if not quiet:
                click.echo("No files found matching the pattern")
            return

        validator = SchemaValidator(schema_file=schema_file)
        valid_count = 0
        invalid_count = 0
        invalid_files = []

        for file_path in files:
            try:
                input_format = detect_format(file_path)
                data = load_data(file_path, input_format)
                errors = validator.validate(data)

                if errors:
                    invalid_count += 1
                    invalid_files.append(file_path)
                    if not quiet:
                        click.echo(f"Invalid: {file_path}")
                        for error in validator.get_error_messages(errors)[:3]:
                            click.echo(f"  {error}")
                else:
                    valid_count += 1
                    if not quiet:
                        click.echo(f"Valid: {file_path}")
            except Exception as e:
                invalid_count += 1
                invalid_files.append(file_path)
                if not quiet:
                    click.echo(f"Error: {file_path} - {e}")

        if not quiet:
            click.echo(f"\nSummary: {valid_count} valid, {invalid_count} invalid")
        elif invalid_count > 0:
            click.echo(f"Valid: {valid_count}, Invalid: {invalid_count}", err=True)

        if invalid_count > 0:
            raise click.ClickException(
                f"Validation failed for {invalid_count} file(s)"
            )

    except Exception as e:
        raise click.ClickException(str(e))


@click.command()
@click.argument("input_file", type=click.Path(exists=True))
@click.option("--infer", is_flag=True, help="Infer schema from data")
@click.option("--quiet", "-q", is_flag=True, help="Minimal output")
def typecheck(
    input_file: str,
    infer: bool,
    quiet: bool,
) -> None:
    """Check types in a data file.

    INPUT_FILE: Input file path (or - for stdin)
    """
    try:
        if input_file == "-":
            import sys

            content = sys.stdin.read()
            input_format = "json"
            data = load_data(content, input_format)
        else:
            input_format = detect_format(input_file)
            data = load_data(input_file, input_format)

        if infer:
            schema = infer_schema_from_data(data)
            if not quiet:
                import json

                click.echo(json.dumps(schema, indent=2))
        else:
            if not quiet:
                inferred_type = type(data).__name__
                if isinstance(data, dict):
                    click.echo(f"Type: object with {len(data)} keys")
                elif isinstance(data, list):
                    click.echo(f"Type: array with {len(data)} items")
                else:
                    click.echo(f"Type: {inferred_type}")

    except Exception as e:
        raise click.ClickException(str(e))