| """CSV document processor""" |
|
|
| import pandas as pd |
| import logging |
| from typing import Dict, Any |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| class CSVProcessor: |
| """Processor for CSV documents""" |
|
|
| @staticmethod |
| def extract_text(file_path: str) -> str: |
| """Extract text from CSV file""" |
| try: |
| df = pd.read_csv(file_path) |
| text = df.to_string() |
| logger.info(f"Extracted text from CSV: {len(df)} rows") |
| return text |
| except Exception as e: |
| logger.error(f"Failed to extract text from CSV: {e}") |
| raise |
|
|
| @staticmethod |
| def get_metadata(file_path: str) -> Dict[str, Any]: |
| """Extract metadata from CSV""" |
| try: |
| df = pd.read_csv(file_path) |
| return { |
| "num_rows": len(df), |
| "num_columns": len(df.columns), |
| "columns": list(df.columns), |
| } |
| except Exception as e: |
| logger.error(f"Failed to extract metadata from CSV: {e}") |
| return {"num_rows": 0} |
|
|
