utils/file_utils.py

"""File I/O utilities for Myanmar Ghost project."""

import json
import os
from pathlib import Path
from typing import Any, Dict, List, Optional

import pandas as pd
import yaml


def load_json(path: str) -> Any:
    """Load JSON file."""
    with open(path, "r", encoding="utf-8") as f:
        return json.load(f)


def save_json(data: Any, path: str, indent: int = 2) -> None:
    """Save data to JSON file."""
    Path(path).parent.mkdir(parents=True, exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=indent, ensure_ascii=False)


def load_yaml(path: str) -> Dict:
    """Load YAML file."""
    with open(path, "r", encoding="utf-8") as f:
        return yaml.safe_load(f)


def save_yaml(data: Dict, path: str) -> None:
    """Save data to YAML file."""
    Path(path).parent.mkdir(parents=True, exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        yaml.dump(data, f, allow_unicode=True, default_flow_style=False)


def load_jsonl(path: str) -> List[Dict]:
    """Load JSONL file (one JSON object per line)."""
    data = []
    with open(path, "r", encoding="utf-8") as f:
        for line in f:
            if line.strip():
                data.append(json.loads(line))
    return data


def save_jsonl(data: List[Dict], path: str) -> None:
    """Save data to JSONL file."""
    Path(path).parent.mkdir(parents=True, exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        for item in data:
            f.write(json.dumps(item, ensure_ascii=False) + "\n")


def load_csv(path: str) -> pd.DataFrame:
    """Load CSV file as DataFrame."""
    return pd.read_csv(path)


def save_csv(df: pd.DataFrame, path: str, index: bool = False) -> None:
    """Save DataFrame to CSV file."""
    Path(path).parent.mkdir(parents=True, exist_ok=True)
    df.to_csv(path, index=index)


def ensure_dir(path: str) -> Path:
    """Ensure directory exists."""
    p = Path(path)
    p.mkdir(parents=True, exist_ok=True)
    return p


def list_files(
    directory: str,
    pattern: str = "*",
    recursive: bool = False,
) -> List[Path]:
    """List files in directory matching pattern."""
    p = Path(directory)
    if recursive:
        return list(p.rglob(pattern))
    return list(p.glob(pattern))


def get_file_size(path: str) -> int:
    """Get file size in bytes."""
    return os.path.getsize(path)


def copy_file(src: str, dst: str) -> None:
    """Copy file from src to dst."""
    import shutil
    Path(dst).parent.mkdir(parents=True, exist_ok=True)
    shutil.copy2(src, dst)


def move_file(src: str, dst: str) -> None:
    """Move file from src to dst."""
    import shutil
    Path(dst).parent.mkdir(parents=True, exist_ok=True)
    shutil.move(src, dst)


def delete_file(path: str) -> None:
    """Delete file."""
    Path(path).unlink(missing_ok=True)


class ConfigManager:
    """Manage configuration files."""
    
    def __init__(self, config_dir: str = "configs"):
        self.config_dir = Path(config_dir)
    
    def load(self, name: str, config_type: str = "yaml") -> Dict:
        """Load configuration by name."""
        path = self.config_dir / f"{name}.{config_type}"
        
        if config_type == "yaml":
            return load_yaml(str(path))
        elif config_type == "json":
            return load_json(str(path))
        else:
            raise ValueError(f"Unsupported config type: {config_type}")
    
    def save(self, name: str, config: Dict, config_type: str = "yaml") -> None:
        """Save configuration by name."""
        path = self.config_dir / f"{name}.{config_type}"
        
        if config_type == "yaml":
            save_yaml(config, str(path))
        elif config_type == "json":
            save_json(config, str(path))
        else:
            raise ValueError(f"Unsupported config type: {config_type}")


if __name__ == "__main__":
    # Test file utilities
    print("File utilities loaded")
    print(f"Current directory: {Path.cwd()}")