| """ |
| Data Manager Service |
| |
| Manages the active data file path for model training. |
| Persists configuration to a JSON file so it survives restarts. |
| """ |
|
|
| import json |
| import logging |
| from datetime import datetime |
| from pathlib import Path |
|
|
| from app.core.config import get_settings |
|
|
| logger = logging.getLogger(__name__) |
|
|
| CONFIG_FILENAME = "data_config.json" |
|
|
|
|
| def _config_path() -> Path: |
| """Get path to the data config file.""" |
| settings = get_settings() |
| return settings.artifacts_dir / CONFIG_FILENAME |
|
|
|
|
| def get_active_data_path() -> Path: |
| """ |
| Get the currently active data file path. |
| Returns custom path if set, otherwise returns default. |
| """ |
| settings = get_settings() |
| config_file = _config_path() |
|
|
| if config_file.exists(): |
| try: |
| config = json.loads(config_file.read_text(encoding="utf-8")) |
| custom_path = config.get("active_data_path") |
| if custom_path: |
| custom_path = Path(custom_path) |
| if custom_path.exists(): |
| return custom_path |
| else: |
| logger.warning("Custom data path no longer exists: %s", custom_path) |
| except (json.JSONDecodeError, KeyError) as e: |
| logger.warning("Failed to read data config: %s", e) |
|
|
| return settings.data_path |
|
|
|
|
| def set_active_data_path(path: Path) -> None: |
| """Set the active data file path.""" |
| config_file = _config_path() |
| config_file.parent.mkdir(parents=True, exist_ok=True) |
|
|
| config = { |
| "active_data_path": str(path), |
| "updated_at": datetime.now().isoformat(), |
| } |
|
|
| config_file.write_text(json.dumps(config, indent=2), encoding="utf-8") |
| logger.info("Updated active data path to: %s", path) |
|
|
|
|
| def reset_to_default() -> None: |
| """Reset to use the default data file.""" |
| config_file = _config_path() |
| if config_file.exists(): |
| config_file.unlink() |
| logger.info("Reset to default data path") |
|
|
|
|
| def save_uploaded_file(content: bytes, original_filename: str) -> Path: |
| """ |
| Save uploaded file content to the data/raw directory. |
| Returns the path to the saved file. |
| """ |
| settings = get_settings() |
| upload_dir = settings.root_dir / "data" / "raw" |
| upload_dir.mkdir(parents=True, exist_ok=True) |
|
|
| |
| timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") |
| |
| suffix = Path(original_filename).suffix or ".csv" |
| filename = f"uploaded_data_{timestamp}{suffix}" |
|
|
| file_path = upload_dir / filename |
| file_path.write_bytes(content) |
|
|
| logger.info("Saved uploaded file to: %s", file_path) |
| return file_path |
|
|
|
|
| def get_data_info() -> dict: |
| """Get information about the current data configuration.""" |
| settings = get_settings() |
| active_path = get_active_data_path() |
| default_path = settings.data_path |
|
|
| return { |
| "active_path": str(active_path), |
| "active_filename": active_path.name, |
| "active_exists": active_path.exists(), |
| "is_custom": active_path != default_path, |
| "default_path": str(default_path), |
| "default_filename": default_path.name, |
| "default_exists": default_path.exists(), |
| } |
|
|
