config-converter-cli/configconverter/converters.py

"""Core conversion logic for config-converter-cli."""

import io
import json
from typing import Any, Dict, Optional

import tomlkit
from ruamel.yaml import YAML

from configconverter.exceptions import (
    InvalidFormatError,
    ParseError,
    UnsupportedConversionError,
)


def _toml_to_dict(obj: Any) -> Any:
    """Convert TOMLDocument to dict recursively."""
    if hasattr(obj, 'unwrap'):
        return _toml_to_dict(obj.unwrap())
    elif isinstance(obj, tomlkit.items.Array):
        return [_toml_to_dict(item) for item in obj]
    elif isinstance(obj, dict):
        return {key: _toml_to_dict(value) for key, value in obj.items()}
    return obj


class Converter:
    """Bidirectional converter between JSON, YAML, and TOML formats."""

    SUPPORTED_FORMATS = ("json", "yaml", "toml")

    def __init__(self):
        self.yaml_parser = YAML()
        self.yaml_parser.preserve_quotes = True
        self.yaml_parser.allow_duplicate_keys = False

    def detect_format(self, content: str) -> str:
        """Detect the format of the input content."""
        content = content.strip()

        if not content:
            raise InvalidFormatError("Empty content")

        first_char = content[0]

        if first_char in ("{", "["):
            if self._is_valid_json(content):
                return "json"

        try:
            parsed = tomlkit.parse(content)
            if parsed:
                return "toml"
        except Exception:
            pass

        try:
            self.yaml_parser.load(content)
            return "yaml"
        except Exception:
            pass

        raise InvalidFormatError("Could not determine input format")

    def _is_valid_json(self, content: str) -> bool:
        """Check if content is valid JSON."""
        try:
            json.loads(content)
            return True
        except json.JSONDecodeError:
            return False

    def convert(self, content: str, from_format: str, to_format: str) -> str:
        """Convert content from one format to another."""
        from_format = from_format.lower()
        if from_format == "auto":
            from_format = self.detect_format(content)

        to_format = to_format.lower()

        if from_format not in self.SUPPORTED_FORMATS:
            raise UnsupportedConversionError(f"Unsupported source format: {from_format}")
        if to_format not in self.SUPPORTED_FORMATS:
            raise UnsupportedConversionError(f"Unsupported target format: {to_format}")

        data = self._parse(content, from_format)

        return self._serialize(data, to_format)

    def _parse(self, content: str, format: str) -> Any:
        """Parse content based on format."""
        if format == "json":
            return self._parse_json(content)
        elif format == "yaml":
            return self._parse_yaml(content)
        elif format == "toml":
            return self._parse_toml(content)
        else:
            raise UnsupportedConversionError(f"Unsupported format: {format}")

    def _parse_json(self, content: str) -> Any:
        """Parse JSON content."""
        try:
            return json.loads(content)
        except json.JSONDecodeError as e:
            raise ParseError(
                f"Invalid JSON syntax: {e.msg}",
                line_number=e.lineno,
                column=e.colno,
                context=e.doc,
            )

    def _parse_yaml(self, content: str) -> Any:
        """Parse YAML content."""
        try:
            return self.yaml_parser.load(content)
        except Exception as e:
            error_msg = str(e)
            line_info = ""
            if hasattr(e, "problem_mark") and e.problem_mark:
                line_info = f" (line {e.problem_mark.line + 1}, column {e.problem_mark.column + 1})"
            raise ParseError(f"Invalid YAML syntax: {error_msg}{line_info}")

    def _parse_toml(self, content: str) -> Any:
        """Parse TOML content."""
        try:
            parsed = tomlkit.parse(content)
            return _toml_to_dict(parsed)
        except tomlkit.exceptions.ParseError as e:
            line_info = ""
            if e.line is not None:
                line_info = f" (line {e.line})"
            raise ParseError(f"Invalid TOML syntax: {e}{line_info}")

    def _serialize(self, data: Any, format: str) -> str:
        """Serialize data to the specified format."""
        if format == "json":
            return self._to_json(data)
        elif format == "yaml":
            return self._to_yaml(data)
        elif format == "toml":
            return self._to_toml(data)
        else:
            raise UnsupportedConversionError(f"Unsupported format: {format}")

    def _to_json(self, data: Any, indent: int = 2) -> str:
        """Convert data to JSON string."""
        return json.dumps(data, indent=indent, ensure_ascii=False)

    def _to_yaml(self, data: Any) -> str:
        """Convert data to YAML string."""
        stream = io.StringIO()
        self.yaml_parser.dump(data, stream)
        return stream.getvalue()

    def _to_toml(self, data: Any) -> str:
        """Convert data to TOML string."""
        return tomlkit.dumps(data)

    def serialize_with_indent(
        self, data: Any, format: str, indent: int = 2, compact: bool = False
    ) -> str:
        """Serialize data with custom indentation."""
        if format == "json":
            if compact:
                return json.dumps(data, separators=(',', ':'), ensure_ascii=False)
            return json.dumps(data, indent=indent, ensure_ascii=False)
        elif format == "yaml":
            stream = io.StringIO()
            self.yaml_parser.indent(mapping=indent, sequence=indent, offset=2)
            self.yaml_parser.dump(data, stream)
            return stream.getvalue()
        elif format == "toml":
            return tomlkit.dumps(data)
        else:
            raise UnsupportedConversionError(f"Unsupported format: {format}")


converter = Converter()