"""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()}")