This commit is contained in:
98
src/converters/csv_converter.py
Normal file
98
src/converters/csv_converter.py
Normal file
@@ -0,0 +1,98 @@
|
|||||||
|
"""CSV converter module."""
|
||||||
|
|
||||||
|
from typing import Any
|
||||||
|
import pandas as pd
|
||||||
|
from io import StringIO
|
||||||
|
|
||||||
|
from .base import BaseConverter, ConversionResult
|
||||||
|
|
||||||
|
|
||||||
|
class CSVConverter(BaseConverter):
|
||||||
|
"""Converter for CSV format."""
|
||||||
|
|
||||||
|
format_name = 'csv'
|
||||||
|
extensions = ('.csv',)
|
||||||
|
|
||||||
|
def loads(self, content: str) -> Any:
|
||||||
|
"""Parse CSV string to pandas DataFrame.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
content: CSV string content
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
pandas DataFrame
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
pd.errors.EmptyDataError: If content is empty
|
||||||
|
pd.errors.ParserError: If content is not valid CSV
|
||||||
|
"""
|
||||||
|
return pd.read_csv(StringIO(content))
|
||||||
|
|
||||||
|
def dumps(self, data: Any, indent: int = 2) -> str:
|
||||||
|
"""Serialize data to CSV string.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
data: DataFrame or list of dicts to serialize
|
||||||
|
indent: Ignored for CSV (no indentation)
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
CSV string representation
|
||||||
|
"""
|
||||||
|
if isinstance(data, pd.DataFrame):
|
||||||
|
return data.to_csv(index=False)
|
||||||
|
elif isinstance(data, list) and data:
|
||||||
|
df = pd.DataFrame(data)
|
||||||
|
return df.to_csv(index=False)
|
||||||
|
else:
|
||||||
|
return ""
|
||||||
|
|
||||||
|
def convert(self, content: str, target_converter: BaseConverter, indent: int = 2) -> ConversionResult:
|
||||||
|
"""Convert content from CSV to another format.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
content: CSV source content
|
||||||
|
target_converter: Target format converter
|
||||||
|
indent: Indentation for output
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
ConversionResult with converted data or error
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
df = self.loads(content)
|
||||||
|
data = df.to_dict(orient='records')
|
||||||
|
result = target_converter.dumps(data, indent=indent)
|
||||||
|
return ConversionResult(
|
||||||
|
success=True,
|
||||||
|
data=result,
|
||||||
|
source_format=self.format_name,
|
||||||
|
target_format=target_converter.format_name
|
||||||
|
)
|
||||||
|
except Exception as e:
|
||||||
|
return ConversionResult(
|
||||||
|
success=False,
|
||||||
|
error=str(e),
|
||||||
|
source_format=self.format_name,
|
||||||
|
target_format=target_converter.format_name
|
||||||
|
)
|
||||||
|
|
||||||
|
def read_file(self, filepath: str) -> str:
|
||||||
|
"""Read CSV content from file.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
filepath: Path to CSV file
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
File content as string
|
||||||
|
"""
|
||||||
|
with open(filepath, 'r', encoding='utf-8') as f:
|
||||||
|
return f.read()
|
||||||
|
|
||||||
|
def write_file(self, filepath: str, content: str) -> None:
|
||||||
|
"""Write CSV content to file.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
filepath: Path to file
|
||||||
|
content: CSV content to write
|
||||||
|
"""
|
||||||
|
with open(filepath, 'w', encoding='utf-8') as f:
|
||||||
|
f.write(content)
|
||||||
Reference in New Issue
Block a user