utils/storage_runtime_util.py

from __future__ import annotations

import json
import os
from datetime import datetime
from pathlib import Path
from typing import Any, Literal


BackupTarget = Literal['postgres', 'redis', 'all']
VALID_BACKUP_TARGETS: set[str] = {'postgres', 'redis', 'all'}


def get_backup_workdir() -> Path:
    return Path(os.environ.get('HF_BACKUP_WORKDIR', '/data/backup-work'))


def get_backup_status_file() -> Path:
    return Path(os.environ.get('HF_BACKUP_STATUS_FILE', str(get_backup_workdir() / 'status' / 'runtime-state.json')))


def get_backup_control_dir() -> Path:
    return Path(os.environ.get('HF_BACKUP_CONTROL_DIR', str(get_backup_workdir() / 'control')))


def get_storage_monitor_page_size_max() -> int:
    raw_value = os.environ.get('HF_STORAGE_MONITOR_PAGE_SIZE_MAX', '200')
    try:
        return max(1, int(raw_value))
    except (TypeError, ValueError):
        return 200


def load_backup_status() -> dict[str, Any]:
    status_file = get_backup_status_file()
    if not status_file.exists():
        return {
            'updated_at': None,
            'backup_enabled': False,
            'status_file': str(status_file),
            'targets': {
                'postgres': {'enabled': False, 'dirty': False, 'last_error': ''},
                'redis': {'enabled': False, 'dirty': False, 'last_error': ''},
            },
        }
    try:
        return json.loads(status_file.read_text(encoding='utf-8'))
    except (OSError, json.JSONDecodeError) as exc:
        return {
            'updated_at': datetime.now().isoformat(),
            'backup_enabled': False,
            'status_file': str(status_file),
            'targets': {
                'postgres': {'enabled': False, 'dirty': False, 'last_error': f'读取状态文件失败: {exc}'},
                'redis': {'enabled': False, 'dirty': False, 'last_error': f'读取状态文件失败: {exc}'},
            },
        }


def request_backup_sync(target: BackupTarget) -> Path:
    if target not in VALID_BACKUP_TARGETS:
        raise ValueError(f'不支持的备份目标: {target}')
    control_dir = get_backup_control_dir()
    control_dir.mkdir(parents=True, exist_ok=True)
    request_file = control_dir / f'{target}.sync'
    request_file.write_text(datetime.now().isoformat(), encoding='utf-8')
    return request_file