diff --git a/configconverter/converters.py b/configconverter/converters.py new file mode 100644 index 0000000..e7529bc --- /dev/null +++ b/configconverter/converters.py @@ -0,0 +1,180 @@ +"""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()