Initial upload: config-converter-cli v1.0.0
This commit is contained in:
180
configconverter/converters.py
Normal file
180
configconverter/converters.py
Normal file
@@ -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()
|
||||
Reference in New Issue
Block a user