ttjson2md/cli.py - master - dictator/ragflow_sync

ttjson2md/cli.py
316 строк10 KB
Developer
feat: v2 enrichment for wiki children, 2 new customfields, docs update
22 июн 2026, 16:37
22 июн 2026, 16:37bc836e8
"""
CLI interface for TaskTracker JSON to Markdown Converter.

Provides subcommands for converting and validating JSON files.
"""

import argparse
import json
import sys
from pathlib import Path

from .core import (
    convert_json_to_markdown,
    convert_list_to_markdown,
    validate_json_structure,
)
from .attributes import ATTRIBUTES_CONFIG


def _print_available_attributes() -> None:
    """Print all available attributes."""
    print("Available attributes:")
    print()
    for code, config in ATTRIBUTES_CONFIG.items():
        if code == "wiki_page_body":
            continue  # Skip wiki page body
        status = "enabled" if config.get("enabled", True) else "disabled"
        print(f"  {code:40s} {config.get('name', ''):30s} ({status})")


def _read_json(input_path: str | None) -> dict | list:
    """Read JSON from file or stdin."""
    try:
        if input_path:
            path = Path(input_path)
            if not path.exists():
                print(f"Error: File not found: {input_path}", file=sys.stderr)
                sys.exit(1)
            with open(path, encoding="utf-8") as f:
                return json.load(f)
        else:
            return json.load(sys.stdin)
    except json.JSONDecodeError as e:
        print(f"Error: Invalid JSON: {e}", file=sys.stderr)
        sys.exit(1)
    except OSError as e:
        print(f"Error: Could not read file: {e}", file=sys.stderr)
        sys.exit(1)


def _write_output(content: str, output_path: str | None) -> None:
    """Write content to file or stdout."""
    try:
        if output_path:
            path = Path(output_path)
            path.parent.mkdir(parents=True, exist_ok=True)
            with open(path, "w", encoding="utf-8") as f:
                f.write(content)
        else:
            print(content)
    except OSError as e:
        print(f"Error: Could not write file: {e}", file=sys.stderr)
        sys.exit(1)


def cmd_convert(args: argparse.Namespace) -> None:
    """Handle the 'convert' subcommand."""
    input_files = args.files or []
    if args.input:
        input_files.insert(0, args.input)

    output_path = args.output
    verbose = args.verbose
    batch = args.batch

    # Parse attribute filters
    include_attrs = set(args.include_attrs.split(",")) if args.include_attrs else None
    exclude_attrs = set(args.exclude_attrs.split(",")) if args.exclude_attrs else None

    # Handle --list-attrs
    if args.list_attrs:
        _print_available_attributes()
        return

    if not input_files and sys.stdin.isatty():
        print(
            "Error: No input files specified.",
            "Use -i <file> or pipe from stdin.",
            file=sys.stderr,
        )
        print("Run 'ttjson2md convert --help'", "for usage.", file=sys.stderr)
        sys.exit(1)

    if not input_files:
        # Single stdin mode - auto-detect dict vs list
        data = _read_json(None)
        if verbose:
            print(
                f"Reading from stdin... (type: {type(data).__name__})",
                file=sys.stderr,
            )
        markdown = convert_json_to_markdown(data, include_attrs, exclude_attrs)
        _write_output(markdown, output_path)
        return

    # Multiple files mode
    if batch:
        # Batch mode: read all files into list and convert together
        all_data = []
        for input_file in input_files:
            if verbose:
                print(f"Reading: {input_file}", file=sys.stderr)
            data = _read_json(input_file)
            if isinstance(data, list):
                all_data.extend(data)
            else:
                all_data.append(data)
        markdown = convert_list_to_markdown(all_data, include_attrs, exclude_attrs)
        _write_output(markdown, output_path)
    else:
        # Individual mode: convert each file separately
        for i, input_file in enumerate(input_files):
            if verbose:
                print(f"Processing: {input_file}", file=sys.stderr)

            data = _read_json(input_file)
            markdown = convert_json_to_markdown(data, include_attrs, exclude_attrs)

            if len(input_files) > 1:
                if output_path:
                    if i > 0:
                        _write_output(
                            f"\n--- File: {input_file} ---\n\n{markdown}",
                            output_path,
                        )
                    else:
                        _write_output(f"# {input_file}\n\n{markdown}", output_path)
                else:
                    if verbose:
                        print(f"--- {input_file} ---", file=sys.stderr)
                    _write_output(markdown, None)
            else:
                _write_output(markdown, output_path)


def cmd_validate(args: argparse.Namespace) -> None:
    """Handle the 'validate' subcommand."""
    input_files = args.files or []
    if args.input:
        input_files.insert(0, args.input)
    verbose = args.verbose

    if not input_files and sys.stdin.isatty():
        print(
            "Error: No input files specified. Use -i <file> or pipe from stdin.",
            file=sys.stderr,
        )
        print("Run 'ttjson validate --help' for usage information.", file=sys.stderr)
        sys.exit(1)

    if not input_files:
        # Single stdin mode
        data = _read_json(None)
        result = validate_json_structure(data)
        _print_validation_result(result, "<stdin>", verbose)
        return

    all_valid = True
    for input_file in input_files:
        if verbose:
            print(f"Validating: {input_file}", file=sys.stderr)

        data = _read_json(input_file)
        result = validate_json_structure(data)
        if not _print_validation_result(result, input_file, verbose):
            all_valid = False

    if not all_valid:
        sys.exit(1)


def _print_validation_result(result: dict, filename: str, verbose: bool) -> bool:
    """Print validation result and return True if valid."""
    is_valid = result["valid"]
    fmt = result["format"]
    issues = result["issues"]

    status = "VALID" if is_valid else "INVALID"
    print(f"{filename}: {status} ({fmt} format)")

    if issues:
        for issue in issues:
            print(f"  - {issue}")

    if verbose:
        fields = result.get("fields", {})
        if fields:
            print("  Detected fields:")
            for key, value in fields.items():
                if value:
                    print(f"    {key}: {value}")

    return is_valid


def build_parser() -> argparse.ArgumentParser:
    """Build the argument parser."""
    parser = argparse.ArgumentParser(
        prog="ttjson2md",
        description="TaskTracker JSON to Markdown Converter CLI",
        epilog="Examples:\n"
        "  ttjson2md convert -i input.json -o output.md\n"
        "  ttjson2md convert file1.json file2.json\n"
        "  cat input.json | ttjson2md convert\n"
        "  ttjson2md validate -i input.json\n",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )

    parser.add_argument(
        "-v",
        "--verbose",
        action="store_true",
        help="Enable verbose output",
    )

    subparsers = parser.add_subparsers(dest="command", help="Available commands")

    # Convert subcommand
    convert_parser = subparsers.add_parser(
        "convert",
        help="Convert JSON files to Markdown",
        description="Convert TaskTracker JSON documents to Markdown format",
    )
    convert_parser.add_argument(
        "files",
        nargs="*",
        help="Input JSON files to convert",
    )
    convert_parser.add_argument(
        "-i",
        "--input",
        help="Single input file (or use stdin)",
    )
    convert_parser.add_argument(
        "-o",
        "--output",
        help="Output file (default: stdout)",
    )
    convert_parser.add_argument(
        "--batch",
        action="store_true",
        help=(
            "Batch mode: treat input as JSON array"
            " or combine multiple files into single output"
        ),
    )
    convert_parser.add_argument(
        "--include-attrs",
        help=(
            "Comma-separated list of attribute codes to include (default: all). "
            "Run --list-attrs to see available codes. "
            "See ttjson2md/attributes.py ATTRIBUTES_CONFIG for mapping details."
        ),
    )
    convert_parser.add_argument(
        "--exclude-attrs",
        help=(
            "Comma-separated list of attribute codes to exclude. "
            "Use with --list-attrs to find codes to filter."
        ),
    )
    convert_parser.add_argument(
        "--list-attrs",
        action="store_true",
        help=(
            "List all 67+ known attribute codes, display names, and enabled/disabled status. "
            "Add new codes to ATTRIBUTES_CONFIG in ttjson2md/attributes.py to map unknown customfields."
        ),
    )
    convert_parser.set_defaults(func=cmd_convert)

    # Validate subcommand
    validate_parser = subparsers.add_parser(
        "validate",
        help="Validate JSON file structure",
        description="Validate TaskTracker JSON file structure",
    )
    validate_parser.add_argument(
        "files",
        nargs="*",
        help="Input JSON files to validate",
    )
    validate_parser.add_argument(
        "-i",
        "--input",
        help="Single input file (or use stdin)",
    )
    validate_parser.set_defaults(func=cmd_validate)

    return parser


def main() -> None:
    """Main entry point for the CLI."""
    parser = build_parser()
    args = parser.parse_args()

    if not args.command:
        parser.print_help()
        sys.exit(1)

    args.func(args)


if __name__ == "__main__":
    main()