From a88b2604a1baad4895a9bdad24aa1d5f7bf8099e Mon Sep 17 00:00:00 2001 From: Roko Date: Tue, 7 Jul 2026 11:27:02 +0700 Subject: [PATCH] =?UTF-8?q?Initial:=20Roko=20Backup=20Script=20v2=20?= =?UTF-8?q?=E2=80=94=20Python=20rewrite=20with=20streaming=20progress=20&?= =?UTF-8?q?=20log=20rotation?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Current deployed version (roko-backup.py) after refactoring from bash to Python. Generated via OpenCode (kimi-k2.7-code). Includes: - rclone streaming progress (log every 5% or 30s) - RotatingFileHandler (max 1MB, 5 backups) - LockFile for concurrent run prevention - Yandex Cloud upload with NO_PROXY bypass - Also includes comparison versions from mimo-v2.5-pro and kimi-k2.7-code --- README.md | 118 +++++++ TECH_SPEC.md | 94 ++++++ roko-backup-kimi.py | 741 ++++++++++++++++++++++++++++++++++++++++++++ roko-backup-mimo.py | 682 ++++++++++++++++++++++++++++++++++++++++ roko-backup.py | 727 +++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 2362 insertions(+) create mode 100644 README.md create mode 100644 TECH_SPEC.md create mode 100644 roko-backup-kimi.py create mode 100644 roko-backup-mimo.py create mode 100755 roko-backup.py diff --git a/README.md b/README.md new file mode 100644 index 0000000..9c46017 --- /dev/null +++ b/README.md @@ -0,0 +1,118 @@ +# Roko Backup Script + +Скрипт для автоматического резервного копирования данных OpenClaw в Yandex Object Storage. + +## Назначение + +Автоматизированный бэкап рабочих файлов, конфигурации, медиа и баз данных OpenClaw с загрузкой в облако (Yandex Object Storage через rclone). Включает верификацию, ротацию старых бэкапов и формирование отчётов. + +## Архитектура + +Скрипт построен по принципу конвейера (pipeline) с разделением на классы: + +- **Config** — конфигурация: пути, лимиты, настройки. Загружается из env-переменных или JSON-файла. +- **LockFile** — предотвращает параллельный запуск через PID-файл. +- **Collector** — собирает файлы для бэкапа в staging-директорию с учётом исключений. +- **Archiver** — создаёт tar.gz-архив из staging-директории. +- **Uploader** — загружает архив в облако через rclone с retry-логикой (10s, 30s, 60s). +- **Verifier** — проверяет наличие архива в облаке. +- **Cleaner** — удаляет старые локальные и облачные бэкапы сверх лимита. +- **Reporter** — пишет JSON-отчёт о результате. +- **BackupPipeline** — оркестрирует все шаги: collect → archive → upload → verify → clean → report. + +## Схема работы + +1. **Блокировка** — создаётся lock-файл `/tmp/roko-backup.lock` для предотвращения параллельных запусков. +2. **Сбор** — файлы копируются во временную staging-директорию с учётом исключений. +3. **Архивация** — staging упаковывается в `roko-YYYYMMDD-HHMMSS.tar.gz` в `~/backups/`. +4. **Загрузка** — архив загружается в `yabackup:for-backup/roko/` через rclone (3 попытки с backoff). +5. **Верификация** — проверяется наличие архива в облаке через `rclone ls`. +6. **Очистка** — удаляются старые локальные (>3) и облачные (>14) бэкапы. +7. **Отчёт** — пишется JSON-отчёт в `~/.openclaw/workspace/roko-backup-report.json`. +8. **Разблокировка** — lock-файл удаляется. + +## Запуск + +### Вручную + +```bash +python3 ~/.openclaw/scripts/roko-backup.py +``` + +### Из cron (ежедневно в 3:00) + +```bash +crontab -e +``` + +Добавить строку: + +```cron +0 3 * * * /usr/bin/python3 /home/openclaw/.openclaw/scripts/roko-backup.py >> /home/openclaw/backups/roko-backup.log 2>&1 +``` + +### Dry-run (проверить что будет бэкапиться) + +```bash +python3 ~/.openclaw/scripts/roko-backup.py --dry-run +``` + +## Аргументы командной строки + +- `--config PATH` — путь к JSON-файлу конфигурации (по умолчанию: env-переменные или built-in defaults). +- `--dry-run` — только собрать файлы, показать что будет бэкапиться, без архивации и загрузки. +- `--verbose` — подробный лог (DEBUG уровень вместо INFO). + +## Логи и отчёты + +- **Логи** — выводятся в stdout (перенаправляйте в файл при запуске из cron). +- **Отчёт** — JSON-файл `~/.openclaw/workspace/roko-backup-report.json` с полями: + - `timestamp` — время бэкапа + - `status` — `success`, `error`, `dry-run` + - `name` — имя архива + - `size_bytes` — размер в байтах + - `error` — сообщение об ошибке (если есть) + +## Что бэкапится + +- `~/.openclaw/workspace/` — рабочая директория +- `~/.openclaw/openclaw.json` — конфигурация +- `~/.openclaw/scripts/` — скрипты +- `~/.openclaw/media/` — медиа-файлы +- `~/.openclaw/plugin-skills/` — навыки плагинов +- `~/.openclaw/agents/main/agent/openclaw-agent.sqlite` — база данных агента + +### Исключения + +- `node_modules/` +- `__pycache__/` +- `*.pyc` +- `.git/` +- `roko-pulse.jsonl` +- `cache/` +- `*.tmp` +- `.cache/` + +## Проксирование (NO_PROXY) + +Для работы с Yandex Object Storage используется переменная окружения: + +```bash +NO_PROXY="storage.yandexcloud.net,yandexcloud.net" +``` + +Это гарантирует, что запросы к `storage.yandexcloud.net` и `yandexcloud.net` идут напрямую, минуя HTTP-прокси (если он настроен в системе). Применяется автоматически во всех вызовах rclone (upload, verify, clean). + +## Конфигурация через env-переменные + +- `ROKO_REPORT_FILE` — путь к файлу отчёта (по умолчанию: `~/.openclaw/workspace/roko-backup-report.json`) +- `ROKO_ARCHIVE_DIR` — директория для локальных архивов (по умолчанию: `~/backups`) +- `ROKO_RCLONE_REMOTE` — rclone remote (по умолчанию: `yabackup:for-backup/roko`) +- `ROKO_MAX_CLOUD` — макс. облачных бэкапов (по умолчанию: 14) +- `ROKO_MAX_LOCAL` — макс. локальных бэкапов (по умолчанию: 3) + +## Требования + +- Python 3.10+ +- rclone (должен быть в PATH) +- Настроенный rclone remote `yabackup` для Yandex Object Storage diff --git a/TECH_SPEC.md b/TECH_SPEC.md new file mode 100644 index 0000000..6a6c302 --- /dev/null +++ b/TECH_SPEC.md @@ -0,0 +1,94 @@ +# Refactoring №2: roko-backup.py — улучшенная версия + +## Исходный код +`/home/openclaw/.openclaw/scripts/roko-backup.py` (текущая версия — 8 классов, 380 строк) + +## ✅ Сохранить +- Архитектуру (конвейер: Config → LockFile → Collector → Archiver → Uploader → Verifier → Cleaner → Reporter → BackupPipeline) +- Python 3.10+, только stdlib + rclone subprocess +- argparse (--config, --dry-run, --verbose) +- LockFile (защита от параллельных запусков) +- NO_PROXY для Yandex Cloud в subprocess env +- Exponential backoff (10s, 30s, 60s) на retry + +## 🆕 Новый функционал + +### 1. Streaming rclone progress +- Сейчас: `subprocess.run(capture_output=True)` — ждём завершения, потом показываем вывод +- Надо: `subprocess.Popen` с построчным чтением stderr в реальном времени +- rclone с `--progress` пишет progress строки в stderr +- **Формат вывода в логах:** + ``` + 2026-07-07 10:30:15 INFO Upload: 15% (95 MiB / 622 MiB) @ 2.1 MiB/s + ``` +- Парсить `Transferred: ... MiB / ... MiB, ...%` из rclone stderr +- Не flood-ить лог — логировать прогресс раз в 30 секунд или при каждом новом проценте (кратном 5%) +- По завершении показать финальную статистику (скорость, время) + +### 2. Ротация логов +- Лог-файл: `/home/openclaw/backups/roko-backup.log` (НА ДИСКЕ, не в /tmp!) +- Максимальный размер: 1 МБ (после этого ротация) +- Хранить: 5 последних ротированных логов (`roko-backup.log.1`, `.2`, ... `.5`) +- Использовать `logging.handlers.RotatingFileHandler` из stdlib +- При `--verbose` — дублировать в stdout тоже (для отладки вручную) + +### 3. Улучшения кода +- **Type hints** полные, включая `collections.abc.Generator`, `ContextManager` +- **Меньше повторений** — RCLONE_ENV дублируется в Uploader/Verifier/Cleaner, вынести в конфиг или хелпер +- **Документация** — docstrings по PEP 257 на английском (краткие, по делу) +- **Error handling** — отдельный класс BackupError(Exception) вместо голых return 3 +- **Cleanup гарантированный** — даже при падении архиватор должен подчищать staging + +## Структура (сохраняем конвейер, улучшаем детали) + +```python +class BackupError(Exception): ... +class LockError(BackupError): ... + +@dataclass(frozen=True) +class Config: + # добавляем log_file: Path, max_log_size: int, max_log_backups: int + ... + +class LockFile(ContextManager): # добавить __enter__/__exit__ + ... + +class Collector: + EXCLUDE_PATTERNS = (...) # можно сделать конфигурируемым в Config + ... + +class Archiver: + # архивирует через tarfile (не subprocess tar) + ... + +class Uploader: + # subprocess.Popen + streaming stderr для progress + UPLOAD_PROGRESS_INTERVAL = 30 # секунд между логами прогресса + PROGRESS_PERCENT_STEP = 5 # логировать каждые N процентов + ... + +class Verifier: ... +class Cleaner: ... +class Reporter: ... +class BackupPipeline: ... + +def setup_logging(verbose, log_file, max_size, max_backups): ... +def main(): ... +``` + +## Требования к качеству +- `python3 -c "import ast; ast.parse(open('roko-backup.py').read())"` — OK +- `--dry-run --verbose` — работает, показывает все источники +- Без аргументов — полный цикл без ошибок +- `--config X` — работает +- `python3 -c "from roko_backup import Config, BackupPipeline"` — импортируемый модуль + +## Файлы +- Пишем: `/home/openclaw/.openclaw/scripts/roko-backup.py` (перезаписать) +- Сохраняем: `roko-backup.py.bak` (текущая версия) +- README: обновить `/home/openclaw/.openclaw/scripts/README.md` (добавить про логи и streaming) + +## Модели для OpenCode +- **Основная:** mimo-v2.5-pro (OpenCode Go) +- **Сравнение:** kimi-k2.7-code (OpenCode Go) +- **Запасные при зависании (до 3 попыток):** minimax-m3 (OpenCode Go), deepseek-v4-pro (OpenCode Go) diff --git a/roko-backup-kimi.py b/roko-backup-kimi.py new file mode 100644 index 0000000..25efe6c --- /dev/null +++ b/roko-backup-kimi.py @@ -0,0 +1,741 @@ +#!/usr/bin/env python3 +"""Roko Backup Script — refactored Python rewrite. + +Collect -> archive -> upload -> verify -> clean -> report. +Runs from system crontab, standalone Python 3.10+. +""" + +from __future__ import annotations + +import abc +import argparse +import datetime +import json +import logging +import logging.handlers +import os +import re +import signal +import shutil +import subprocess +import sys +import tarfile +import tempfile +import time +from collections.abc import Generator +from contextlib import contextmanager +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, ContextManager + +log = logging.getLogger("roko-backup") + + +# ─── Exceptions ────────────────────────────────────────────────────────────── + + +class BackupError(Exception): + """Base exception for backup pipeline failures.""" + + +class LockError(BackupError): + """Raised when the lock file cannot be acquired.""" + + +# ─── Config ────────────────────────────────────────────────────────────────── + + +@dataclass(frozen=True) +class Config: + """All paths and limits for the backup pipeline.""" + + home: Path = field(default_factory=lambda: Path.home()) + report_file: Path = field( + default_factory=lambda: Path.home() + / ".openclaw" + / "workspace" + / "roko-backup-report.json" + ) + archive_dir: Path = field(default_factory=lambda: Path.home() / "backups") + log_file: Path = field(default_factory=lambda: Path.home() / "backups" / "roko-backup.log") + rclone_remote: str = "yabackup:for-backup/roko" + max_cloud_backups: int = 14 + max_local_backups: int = 3 + max_log_size: int = 1_048_576 # 1 MiB + max_log_backups: int = 5 + lock_path: Path = Path("/tmp/roko-backup.lock") + + @property + def workspace(self) -> Path: + """Path to the OpenCode workspace directory.""" + return self.home / ".openclaw" / "workspace" + + @property + def config_file(self) -> Path: + """Path to the main OpenCode config file.""" + return self.home / ".openclaw" / "openclaw.json" + + @property + def scripts_dir(self) -> Path: + """Path to the OpenCode scripts directory.""" + return self.home / ".openclaw" / "scripts" + + @property + def media_dir(self) -> Path: + """Path to the OpenCode media directory.""" + return self.home / ".openclaw" / "media" + + @property + def plugin_skills_dir(self) -> Path: + """Path to the OpenCode plugin-skills directory.""" + return self.home / ".openclaw" / "plugin-skills" + + @property + def agent_db(self) -> Path: + """Path to the main agent SQLite database.""" + return self.home / ".openclaw" / "agents" / "main" / "agent" / "openclaw-agent.sqlite" + + @property + def rclone_env(self) -> dict[str, str]: + """Environment variables for rclone subprocesses.""" + return { + **os.environ, + "NO_PROXY": "storage.yandexcloud.net,yandexcloud.net", + } + + @classmethod + def from_env(cls) -> Config: + """Build config from environment variables with built-in defaults.""" + home = Path.home() + return cls( + home=home, + report_file=Path( + os.environ.get( + "ROKO_REPORT_FILE", str(home / ".openclaw/workspace/roko-backup-report.json") + ) + ), + archive_dir=Path(os.environ.get("ROKO_ARCHIVE_DIR", str(home / "backups"))), + log_file=Path(os.environ.get("ROKO_LOG_FILE", str(home / "backups/roko-backup.log"))), + rclone_remote=os.environ.get("ROKO_RCLONE_REMOTE", "yabackup:for-backup/roko"), + max_cloud_backups=int(os.environ.get("ROKO_MAX_CLOUD", "14")), + max_local_backups=int(os.environ.get("ROKO_MAX_LOCAL", "3")), + max_log_size=int(os.environ.get("ROKO_MAX_LOG_SIZE", str(1_048_576))), + max_log_backups=int(os.environ.get("ROKO_MAX_LOG_BACKUPS", "5")), + ) + + @classmethod + def from_file(cls, path: Path) -> Config: + """Load config from a JSON file, falling back to defaults.""" + data: dict[str, Any] = {} + if path.exists(): + data = json.loads(path.read_text()) + home = Path(data.get("home", str(Path.home()))) + return cls( + home=home, + report_file=Path( + data.get( + "report_file", str(home / ".openclaw/workspace/roko-backup-report.json") + ) + ), + archive_dir=Path(data.get("archive_dir", str(home / "backups"))), + log_file=Path(data.get("log_file", str(home / "backups/roko-backup.log"))), + rclone_remote=data.get("rclone_remote", "yabackup:for-backup/roko"), + max_cloud_backups=data.get("max_cloud_backups", 14), + max_local_backups=data.get("max_local_backups", 3), + max_log_size=data.get("max_log_size", 1_048_576), + max_log_backups=data.get("max_log_backups", 5), + ) + + +# ─── LockFile ──────────────────────────────────────────────────────────────── + + +class LockFile(ContextManager["LockFile"]): + """Prevents concurrent backup runs via a PID file.""" + + def __init__(self, path: Path) -> None: + self.path = path + self._acquired = False + + def acquire(self) -> None: + """Acquire the lock. Raises LockError if another live process holds it.""" + if self.path.exists(): + try: + pid = int(self.path.read_text().strip()) + os.kill(pid, 0) + raise LockError(f"Another backup is already running (lock: {self.path})") + except (ValueError, ProcessLookupError, PermissionError, OSError): + log.warning("Stale lock file found (PID dead), removing") + self.path.unlink(missing_ok=True) + + self.path.write_text(str(os.getpid())) + self._acquired = True + signal.signal(signal.SIGTERM, self._signal_handler) + signal.signal(signal.SIGINT, self._signal_handler) + + def release(self) -> None: + """Release the lock and remove the file.""" + if self._acquired: + self.path.unlink(missing_ok=True) + self._acquired = False + + def _signal_handler(self, signum: int, _frame: Any) -> None: + self.release() + sys.exit(128 + signum) + + def __enter__(self) -> LockFile: + self.acquire() + return self + + def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: + self.release() + + +# ─── Collector ─────────────────────────────────────────────────────────────── + + +class Collector: + """Gathers files to back up into a staging directory.""" + + EXCLUDE_PATTERNS = ( + "node_modules", + "__pycache__", + "*.pyc", + ".git", + "roko-pulse.jsonl", + "cache", + "*.tmp", + ".cache", + ) + + def __init__(self, config: Config, exclude_patterns: tuple[str, ...] | None = None) -> None: + self.config = config + self.exclude_patterns = exclude_patterns or self.EXCLUDE_PATTERNS + + def collect(self, staging: Path) -> tuple[list[tuple[Path, Path]], int]: + """Copy workspace/config/scripts/media/plugin-skills/agent-db into staging. + + Returns: + A tuple of (list of (source, dest) pairs, total bytes collected). + """ + collected: list[tuple[Path, Path]] = [] + total_bytes = 0 + ignore = shutil.ignore_patterns(*self.exclude_patterns) + + targets: list[tuple[Path, str]] = [ + (self.config.workspace, "openclaw/workspace"), + (self.config.config_file, "openclaw/openclaw.json"), + (self.config.scripts_dir, "openclaw/scripts"), + (self.config.media_dir, "openclaw/media"), + (self.config.plugin_skills_dir, "openclaw/plugin-skills"), + (self.config.agent_db, "openclaw/openclaw-agent.sqlite"), + ] + + for src, dest_rel in targets: + dest = staging / dest_rel + if not src.exists(): + log.warning("Source not found, skipping: %s", src) + continue + + dest.parent.mkdir(parents=True, exist_ok=True) + + try: + if src.is_dir(): + shutil.copytree(src, dest, ignore=ignore, dirs_exist_ok=True) + else: + shutil.copy2(src, dest) + except Exception as exc: + log.warning("Failed to copy %s: %s", src, exc) + continue + + size = self._tree_size(dest) + collected.append((src, dest)) + total_bytes += size + log.info("Collected %s -> %s (%d bytes)", src.name, dest_rel, size) + + return collected, total_bytes + + @staticmethod + def _tree_size(path: Path) -> int: + """Return total size in bytes for a file or directory.""" + if path.is_file(): + return path.stat().st_size + return sum(f.stat().st_size for f in path.rglob("*") if f.is_file()) + + +# ─── Archiver ──────────────────────────────────────────────────────────────── + + +class Archiver: + """Creates a tar.gz archive from a staging directory.""" + + def __init__(self, config: Config) -> None: + self.config = config + + def archive(self, staging: Path, backup_name: str) -> tuple[Path, int]: + """Create tar.gz archive. Returns (archive_path, size_bytes).""" + self.config.archive_dir.mkdir(parents=True, exist_ok=True) + archive_path = self.config.archive_dir / f"{backup_name}.tar.gz" + + log.info("Creating archive: %s", archive_path) + with tarfile.open(archive_path, mode="w:gz") as tar: + tar.add(staging, arcname=".") + + size = archive_path.stat().st_size + log.info("Archive created: %s (%d bytes)", archive_path.name, size) + return archive_path, size + + +# ─── Uploader ──────────────────────────────────────────────────────────────── + + +class Uploader: + """Uploads archive via rclone to Yandex Object Storage with retry.""" + + UPLOAD_PROGRESS_INTERVAL = 30 # seconds between progress log lines + PROGRESS_PERCENT_STEP = 5 # log every N percent + + BACKOFF_SCHEDULE = (10, 30, 60) + + _PROGRESS_RE = re.compile( + r"Transferred:\s*([\d.]+)\s*(\w*)\s*/\s*([\d.]+)\s*(\w*),\s*([\d.]+)%" + ) + + def __init__(self, config: Config) -> None: + self.config = config + + def upload(self, archive_path: Path) -> bool: + """Upload archive to remote. Returns True on success.""" + max_attempts = len(self.BACKOFF_SCHEDULE) + 1 + + for attempt in range(1, max_attempts + 1): + log.info("Upload attempt %d/%d", attempt, max_attempts) + if self._upload_once(archive_path): + log.info("Upload succeeded") + return True + + if attempt < max_attempts: + delay = self.BACKOFF_SCHEDULE[min(attempt - 1, len(self.BACKOFF_SCHEDULE) - 1)] + log.info("Retrying in %ds...", delay) + time.sleep(delay) + + log.error("Upload failed after %d attempts", max_attempts) + return False + + def _upload_once(self, archive_path: Path) -> bool: + """Run a single upload attempt with streaming progress.""" + cmd = [ + "rclone", + "copy", + str(archive_path), + f"{self.config.rclone_remote}/", + "--progress", + ] + log.debug("Running: %s", " ".join(cmd)) + + process = subprocess.Popen( + cmd, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + env=self.config.rclone_env, + ) + + last_logged_percent = -1 + last_log_time = 0.0 + final_stats: dict[str, str | float] = {} + + if process.stderr is None: + log.warning("rclone stderr not available") + process.wait() + return process.returncode == 0 + + for line in process.stderr: + line = line.strip() + if not line: + continue + + match = self._PROGRESS_RE.search(line) + if match: + current, cur_unit, total, tot_unit, percent = match.groups() + pct = float(percent) + should_log = ( + pct - last_logged_percent >= self.PROGRESS_PERCENT_STEP + or time.monotonic() - last_log_time >= self.UPLOAD_PROGRESS_INTERVAL + ) + if should_log: + log.info( + "Upload: %.0f%% (%s %s / %s %s)", + pct, + current, + cur_unit or "B", + total, + tot_unit or "B", + ) + last_logged_percent = pct + last_log_time = time.monotonic() + final_stats = { + "current": current, + "current_unit": cur_unit or "B", + "total": total, + "total_unit": tot_unit or "B", + "percent": pct, + } + elif "Transferred:" in line and "Elapsed time:" in line: + final_stats["summary"] = line + + process.wait() + if process.returncode == 0: + if final_stats: + summary = final_stats.get("summary") + if summary: + log.info("Upload stats: %s", summary) + else: + log.info( + "Upload complete: %.0f%% (%s %s / %s %s)", + final_stats.get("percent", 0.0), + final_stats.get("current", "0"), + final_stats.get("current_unit", "B"), + final_stats.get("total", "0"), + final_stats.get("total_unit", "B"), + ) + return True + + log.warning("Upload failed (rc=%d)", process.returncode) + return False + + +# ─── Verifier ──────────────────────────────────────────────────────────────── + + +class Verifier: + """Confirms the uploaded file exists in the remote.""" + + def __init__(self, config: Config) -> None: + self.config = config + + def verify(self, backup_name: str) -> tuple[bool, str]: + """Check that backup_name.tar.gz is listed in the remote. + + Returns: + A tuple of (success, details). + """ + result = subprocess.run( + ["rclone", "ls", self.config.rclone_remote], + capture_output=True, + text=True, + check=False, + env=self.config.rclone_env, + ) + + if result.returncode != 0: + return False, f"rclone ls failed (rc={result.returncode}): {result.stderr[:200]}" + + target = f"{backup_name}.tar.gz" + if target in result.stdout: + return True, f"Verified: {target} found in remote" + + return False, f"Not found in remote: {target}" + + +# ─── Cleaner ───────────────────────────────────────────────────────────────── + + +class Cleaner: + """Manages retention of local and cloud backups.""" + + def __init__(self, config: Config) -> None: + self.config = config + + def clean_local(self) -> int: + """Delete local archives beyond MAX_LOCAL_BACKUPS. Returns count deleted.""" + archives = sorted( + self.config.archive_dir.glob("roko-*.tar.gz"), + key=lambda p: p.stat().st_mtime, + reverse=True, + ) + deleted = 0 + for old in archives[self.config.max_local_backups :]: + log.info("Deleting local: %s", old.name) + old.unlink() + deleted += 1 + return deleted + + def clean_cloud(self) -> int: + """Delete cloud archives beyond MAX_CLOUD_BACKUPS. Returns count deleted.""" + result = subprocess.run( + ["rclone", "ls", self.config.rclone_remote], + capture_output=True, + text=True, + check=False, + env=self.config.rclone_env, + ) + if result.returncode != 0: + log.warning("Cannot list cloud backups for cleanup: %s", result.stderr[:200]) + return 0 + + entries: list[tuple[int, str]] = [] + for line in result.stdout.strip().splitlines(): + parts = line.split(None, 1) + if len(parts) == 2: + try: + entries.append((int(parts[0]), parts[1])) + except ValueError: + continue + + entries.sort(key=lambda e: e[1]) + deleted = 0 + to_remove = entries[: max(0, len(entries) - self.config.max_cloud_backups)] + + for _size, name in to_remove: + log.info("Deleting cloud: %s", name) + subprocess.run( + ["rclone", "delete", f"{self.config.rclone_remote}/{name}"], + capture_output=True, + text=True, + check=False, + env=self.config.rclone_env, + ) + deleted += 1 + + return deleted + + +# ─── Reporter ──────────────────────────────────────────────────────────────── + + +class Reporter: + """Writes JSON backup report to the workspace.""" + + def __init__(self, config: Config) -> None: + self.config = config + + def write( + self, + status: str, + name: str | None = None, + size_bytes: int | None = None, + error: str | None = None, + ) -> None: + """Write the report JSON file.""" + report = { + "timestamp": datetime.datetime.now().astimezone().isoformat(), + "status": status, + "name": name, + "size_bytes": size_bytes, + "error": error, + } + self.config.report_file.parent.mkdir(parents=True, exist_ok=True) + self.config.report_file.write_text(json.dumps(report, indent=2) + "\n") + log.info("Report written: %s (%s)", self.config.report_file, status) + + +# ─── Pipeline ──────────────────────────────────────────────────────────────── + + +class BackupPipeline: + """Orchestrates the full backup: collect -> archive -> upload -> verify -> clean -> report.""" + + def __init__(self, config: Config, *, dry_run: bool = False) -> None: + self.config = config + self.dry_run = dry_run + self.collector = Collector(config) + self.archiver = Archiver(config) + self.uploader = Uploader(config) + self.verifier = Verifier(config) + self.cleaner = Cleaner(config) + self.reporter = Reporter(config) + + def run(self) -> int: + """Execute the pipeline. Returns exit code.""" + backup_name = f"roko-{datetime.datetime.now().strftime('%Y%m%d-%H%M%S')}" + log.info("Starting backup: %s", backup_name) + + free = shutil.disk_usage(self.config.archive_dir).free + log.info("Free disk: %s", self._fmt_bytes(free)) + + archive_path: Path | None = None + archive_size: int | None = None + + with self._managed_staging() as staging: + # 1. Collect + t0 = time.monotonic() + collected, total_bytes = self.collector.collect(staging) + log.info( + "Collection done in %.1fs: %d items, %s", + time.monotonic() - t0, + len(collected), + self._fmt_bytes(total_bytes), + ) + + if self.dry_run: + log.info( + "[DRY RUN] Would archive and upload %d items (%s)", + len(collected), + self._fmt_bytes(total_bytes), + ) + for src, dest in collected: + log.info(" %s -> %s", src, dest.relative_to(staging)) + self.reporter.write("dry-run", name=backup_name, size_bytes=total_bytes) + return 0 + + # 2. Archive + t0 = time.monotonic() + archive_path, archive_size = self.archiver.archive(staging, backup_name) + log.info( + "Archiving done in %.1fs: %s", + time.monotonic() - t0, + self._fmt_bytes(archive_size), + ) + + # 3. Upload + if archive_path is None or archive_size is None: + msg = "Archive was not created" + log.error(msg) + self.reporter.write("error", name=backup_name, error=msg) + return 3 + + t0 = time.monotonic() + if not self.uploader.upload(archive_path): + msg = "Upload failed after retries" + log.error(msg) + self.reporter.write("error", name=backup_name, size_bytes=archive_size, error=msg) + return 3 + log.info("Upload done in %.1fs", time.monotonic() - t0) + + # 4. Verify + ok, detail = self.verifier.verify(backup_name) + if not ok: + log.error("Verification failed: %s", detail) + self.reporter.write("error", name=backup_name, size_bytes=archive_size, error=detail) + return 3 + log.info("Verification: %s", detail) + + # 5. Clean + local_deleted = self.cleaner.clean_local() + cloud_deleted = self.cleaner.clean_cloud() + log.info("Cleanup: %d local, %d cloud removed", local_deleted, cloud_deleted) + + # 6. Report + self.reporter.write("success", name=backup_name, size_bytes=archive_size) + log.info("Backup complete: %s (%s)", backup_name, self._fmt_bytes(archive_size)) + return 0 + + @contextmanager + def _managed_staging(self) -> Generator[Path, None, None]: + """Create and always clean up a temporary staging directory.""" + tmp_dir = tempfile.mkdtemp(prefix="roko-backup-") + staging = Path(tmp_dir) + try: + yield staging + finally: + try: + shutil.rmtree(staging, ignore_errors=True) + except Exception as exc: + log.warning("Failed to remove staging directory %s: %s", staging, exc) + + @staticmethod + def _fmt_bytes(n: int) -> str: + """Human-readable byte size.""" + value = float(n) + for unit in ("B", "KB", "MB", "GB"): + if abs(value) < 1024: + return f"{value:.1f} {unit}" + value /= 1024 + return f"{value:.1f} TB" + + +# ─── Main ──────────────────────────────────────────────────────────────────── + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + """Parse command-line arguments.""" + parser = argparse.ArgumentParser( + prog="roko-backup.py", + description="Roko backup: collect -> archive -> upload -> verify -> clean -> report", + ) + parser.add_argument( + "--config", + type=Path, + default=None, + help="Config file path (default: env vars / built-in defaults)", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Collect and print what would be backed up, don't archive/upload", + ) + parser.add_argument( + "--verbose", + action="store_true", + help="Detailed logging output (DEBUG level, also to stdout)", + ) + return parser.parse_args(argv) + + +def setup_logging( + verbose: bool, + log_file: Path, + max_size: int, + max_backups: int, +) -> None: + """Configure rotating file logging and optional stdout output.""" + level = logging.DEBUG if verbose else logging.INFO + formatter = logging.Formatter( + "%(asctime)s %(levelname)-5s %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + + log_file.parent.mkdir(parents=True, exist_ok=True) + file_handler = logging.handlers.RotatingFileHandler( + str(log_file), + maxBytes=max_size, + backupCount=max_backups, + ) + file_handler.setFormatter(formatter) + + root = logging.getLogger() + root.setLevel(level) + root.addHandler(file_handler) + + if verbose: + stream_handler = logging.StreamHandler(sys.stdout) + stream_handler.setFormatter(formatter) + root.addHandler(stream_handler) + + +def main(argv: list[str] | None = None) -> int: + """Entry point.""" + args = parse_args(argv) + + if args.config: + config = Config.from_file(args.config) + else: + config = Config.from_env() + + setup_logging( + args.verbose, + config.log_file, + config.max_log_size, + config.max_log_backups, + ) + + if not shutil.which("rclone"): + log.error("rclone not found in PATH") + return 1 + + try: + with LockFile(config.lock_path): + pipeline = BackupPipeline(config, dry_run=args.dry_run) + return pipeline.run() + except LockError as exc: + log.error("%s", exc) + return 2 + except Exception as exc: + log.exception("Unexpected error: %s", exc) + Reporter(config).write("error", error=str(exc)) + return 3 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/roko-backup-mimo.py b/roko-backup-mimo.py new file mode 100644 index 0000000..6ad74fd --- /dev/null +++ b/roko-backup-mimo.py @@ -0,0 +1,682 @@ +#!/usr/bin/env python3 +"""Roko Backup Script — refactored v2. + +Collect -> archive -> upload (streaming progress) -> verify -> clean -> report. +Runs from system crontab, standalone Python 3.10+. +""" + +from __future__ import annotations + +import abc +import argparse +import collections.abc +import contextlib +import datetime +import json +import logging +import logging.handlers +import os +import re +import signal +import shutil +import subprocess +import sys +import tarfile +import tempfile +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +log = logging.getLogger("roko-backup") + + +# --- Exceptions --------------------------------------------------------------- + + +class BackupError(Exception): + """Base exception for backup pipeline failures.""" + + +class LockError(BackupError): + """Raised when the lock cannot be acquired.""" + + +# --- Shared rclone environment ------------------------------------------------ + + +def _rclone_env() -> dict[str, str]: + """Return environment dict with NO_PROXY set for Yandex Cloud.""" + return { + **os.environ, + "NO_PROXY": "storage.yandexcloud.net,yandexcloud.net", + } + + +# --- Config ------------------------------------------------------------------- + + +@dataclass(frozen=True) +class Config: + """All paths and limits for the backup pipeline.""" + + home: Path = field(default_factory=lambda: Path.home()) + report_file: Path = field( + default_factory=lambda: Path.home() / ".openclaw" / "workspace" / "roko-backup-report.json" + ) + archive_dir: Path = field(default_factory=lambda: Path.home() / "backups") + log_file: Path = field(default_factory=lambda: Path.home() / "backups" / "roko-backup.log") + rclone_remote: str = "yabackup:for-backup/roko" + max_cloud_backups: int = 14 + max_local_backups: int = 3 + max_log_size: int = 1_048_576 # 1 MiB + max_log_backups: int = 5 + lock_path: Path = field(default_factory=lambda: Path("/tmp/roko-backup.lock")) + + @property + def workspace(self) -> Path: + """Return the workspace directory.""" + return self.home / ".openclaw" / "workspace" + + @property + def config_file(self) -> Path: + """Return the main openclaw config file.""" + return self.home / ".openclaw" / "openclaw.json" + + @property + def scripts_dir(self) -> Path: + """Return the scripts directory.""" + return self.home / ".openclaw" / "scripts" + + @property + def media_dir(self) -> Path: + """Return the media directory.""" + return self.home / ".openclaw" / "media" + + @property + def plugin_skills_dir(self) -> Path: + """Return the plugin-skills directory.""" + return self.home / ".openclaw" / "plugin-skills" + + @property + def agent_db(self) -> Path: + """Return the agent SQLite database path.""" + return self.home / ".openclaw" / "agents" / "main" / "agent" / "openclaw-agent.sqlite" + + @classmethod + def from_env(cls) -> Config: + """Build config from environment variables with built-in defaults.""" + home = Path.home() + return cls( + home=home, + report_file=Path(os.environ.get("ROKO_REPORT_FILE", str(home / ".openclaw/workspace/roko-backup-report.json"))), + archive_dir=Path(os.environ.get("ROKO_ARCHIVE_DIR", str(home / "backups"))), + log_file=Path(os.environ.get("ROKO_LOG_FILE", str(home / "backups/roko-backup.log"))), + rclone_remote=os.environ.get("ROKO_RCLONE_REMOTE", "yabackup:for-backup/roko"), + max_cloud_backups=int(os.environ.get("ROKO_MAX_CLOUD", "14")), + max_local_backups=int(os.environ.get("ROKO_MAX_LOCAL", "3")), + ) + + @classmethod + def from_file(cls, path: Path) -> Config: + """Load config from a JSON file, falling back to defaults.""" + data: dict[str, Any] = {} + if path.exists(): + data = json.loads(path.read_text()) + home = Path(data.get("home", str(Path.home()))) + return cls( + home=home, + report_file=Path(data.get("report_file", str(home / ".openclaw/workspace/roko-backup-report.json"))), + archive_dir=Path(data.get("archive_dir", str(home / "backups"))), + log_file=Path(data.get("log_file", str(home / "backups/roko-backup.log"))), + rclone_remote=data.get("rclone_remote", "yabackup:for-backup/roko"), + max_cloud_backups=data.get("max_cloud_backups", 14), + max_local_backups=data.get("max_local_backups", 3), + max_log_size=data.get("max_log_size", 1_048_576), + max_log_backups=data.get("max_log_backups", 5), + ) + + +# --- LockFile ----------------------------------------------------------------- + + +class LockFile: + """Context manager that prevents concurrent backup runs via a PID file.""" + + def __init__(self, path: Path) -> None: + self.path = path + self._acquired = False + + def __enter__(self) -> LockFile: + if not self.acquire(): + raise LockError(f"Another backup is already running (lock: {self.path})") + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_val: BaseException | None, + exc_tb: Any, + ) -> None: + self.release() + + def acquire(self) -> bool: + """Acquire the lock. Returns False if another live process holds it.""" + if self.path.exists(): + try: + pid = int(self.path.read_text().strip()) + os.kill(pid, 0) + return False + except (ValueError, ProcessLookupError, PermissionError, OSError): + log.warning("Stale lock file found (PID dead), removing") + self.path.unlink(missing_ok=True) + + self.path.write_text(str(os.getpid())) + self._acquired = True + signal.signal(signal.SIGTERM, self._signal_handler) + signal.signal(signal.SIGINT, self._signal_handler) + return True + + def release(self) -> None: + """Release the lock and remove the file.""" + if self._acquired: + self.path.unlink(missing_ok=True) + self._acquired = False + + def _signal_handler(self, signum: int, _frame: Any) -> None: + self.release() + sys.exit(128 + signum) + + +# --- Collector ---------------------------------------------------------------- + + +class Collector: + """Gathers files to back up into a staging directory.""" + + EXCLUDE_PATTERNS: tuple[str, ...] = ( + "node_modules", + "__pycache__", + "*.pyc", + ".git", + "roko-pulse.jsonl", + "cache", + "*.tmp", + ".cache", + ) + + def __init__(self, config: Config) -> None: + self.config = config + + def collect(self, staging: Path) -> tuple[list[tuple[Path, Path]], int]: + """Copy workspace/config/scripts/media/plugin-skills/agent-db into staging. + + Returns (list of (source, dest) pairs, total bytes collected). + """ + collected: list[tuple[Path, Path]] = [] + total_bytes = 0 + ignore = shutil.ignore_patterns(*self.EXCLUDE_PATTERNS) + + targets: list[tuple[Path, str]] = [ + (self.config.workspace, "openclaw/workspace"), + (self.config.config_file, "openclaw/openclaw.json"), + (self.config.scripts_dir, "openclaw/scripts"), + (self.config.media_dir, "openclaw/media"), + (self.config.plugin_skills_dir, "openclaw/plugin-skills"), + (self.config.agent_db, "openclaw/openclaw-agent.sqlite"), + ] + + for src, dest_rel in targets: + dest = staging / dest_rel + if not src.exists(): + log.warning("Source not found, skipping: %s", src) + continue + + dest.parent.mkdir(parents=True, exist_ok=True) + + try: + if src.is_dir(): + shutil.copytree(src, dest, ignore=ignore, dirs_exist_ok=True) + else: + shutil.copy2(src, dest) + except Exception as exc: + log.warning("Failed to copy %s: %s", src, exc) + continue + + size = self._tree_size(dest) + collected.append((src, dest)) + total_bytes += size + log.info("Collected %s -> %s (%d bytes)", src.name, dest_rel, size) + + return collected, total_bytes + + @staticmethod + def _tree_size(path: Path) -> int: + """Return total size in bytes for a file or directory.""" + if path.is_file(): + return path.stat().st_size + return sum(f.stat().st_size for f in path.rglob("*") if f.is_file()) + + +# --- Archiver ----------------------------------------------------------------- + + +class Archiver: + """Creates a tar.gz archive from a staging directory.""" + + def __init__(self, config: Config) -> None: + self.config = config + + def archive(self, staging: Path, backup_name: str) -> tuple[Path, int]: + """Create tar.gz archive. Returns (archive_path, size_bytes).""" + self.config.archive_dir.mkdir(parents=True, exist_ok=True) + archive_path = self.config.archive_dir / f"{backup_name}.tar.gz" + + log.info("Creating archive: %s", archive_path) + with tarfile.open(archive_path, mode="w:gz") as tar: + tar.add(staging, arcname=".") + + size = archive_path.stat().st_size + log.info("Archive created: %s (%d bytes)", archive_path.name, size) + return archive_path, size + + +# --- Uploader ----------------------------------------------------------------- + + +class Uploader: + """Uploads archive via rclone with streaming progress and retry.""" + + BACKOFF_SCHEDULE: tuple[int, ...] = (10, 30, 60) + UPLOAD_PROGRESS_INTERVAL: int = 30 # seconds between progress logs + PROGRESS_PERCENT_STEP: int = 5 # log every N percent + + _RE_TRANSFERRED = re.compile( + r"Transferred:\s+[\w.]+\s+([\d.]+\s*\w+)\s+/\s+([\d.]+\s*\w+),\s+(\d+)%" + ) + + def __init__(self, config: Config) -> None: + self.config = config + + def upload(self, archive_path: Path) -> bool: + """Upload archive to remote with streaming progress. Returns True on success.""" + max_attempts = len(self.BACKOFF_SCHEDULE) + 1 + + for attempt in range(1, max_attempts + 1): + log.info("Upload attempt %d/%d", attempt, max_attempts) + + if self._upload_once(archive_path): + return True + + if attempt < max_attempts: + delay = self.BACKOFF_SCHEDULE[min(attempt - 1, len(self.BACKOFF_SCHEDULE) - 1)] + log.info("Retrying in %ds...", delay) + time.sleep(delay) + + log.error("Upload failed after %d attempts", max_attempts) + return False + + def _upload_once(self, archive_path: Path) -> bool: + """Run a single rclone copy with streaming stderr. + + Parses progress lines from rclone and logs periodically. + """ + cmd = [ + "rclone", "copy", + str(archive_path), + f"{self.config.rclone_remote}/", + "--progress", + ] + last_log_ts: float = 0.0 + last_pct: int = -1 + + with subprocess.Popen( + cmd, + stderr=subprocess.PIPE, + stdout=subprocess.DEVNULL, + text=True, + env=_rclone_env(), + ) as proc: + assert proc.stderr is not None + for line in proc.stderr: + line = line.rstrip("\n") + if not line: + continue + + m = self._RE_TRANSFERRED.search(line) + if m: + transferred, total, pct_str = m.group(1), m.group(2), int(m.group(3)) + now = time.monotonic() + new_pct_bucket = pct_str // self.PROGRESS_PERCENT_STEP * self.PROGRESS_PERCENT_STEP + if now - last_log_ts >= self.UPLOAD_PROGRESS_INTERVAL or new_pct_bucket > last_pct: + log.info("Upload: %d%% (%s / %s)", pct_str, transferred, total) + last_log_ts = now + last_pct = new_pct_bucket + else: + log.debug("rclone: %s", line) + + proc.wait() + + if proc.returncode == 0: + log.info("Upload succeeded") + return True + + log.warning("Upload failed (rc=%d)", proc.returncode) + return False + + +# --- Verifier ----------------------------------------------------------------- + + +class Verifier: + """Confirms the uploaded file exists in the remote.""" + + def __init__(self, config: Config) -> None: + self.config = config + + def verify(self, backup_name: str) -> tuple[bool, str]: + """Check that backup_name.tar.gz is listed in the remote. + + Returns (success, details). + """ + result = subprocess.run( + ["rclone", "ls", self.config.rclone_remote], + capture_output=True, + text=True, + check=False, + env=_rclone_env(), + ) + + if result.returncode != 0: + return False, f"rclone ls failed (rc={result.returncode}): {result.stderr[:200]}" + + target = f"{backup_name}.tar.gz" + if target in result.stdout: + return True, f"Verified: {target} found in remote" + + return False, f"Not found in remote: {target}" + + +# --- Cleaner ------------------------------------------------------------------ + + +class Cleaner: + """Manages retention of local and cloud backups.""" + + def __init__(self, config: Config) -> None: + self.config = config + + def clean_local(self) -> int: + """Delete local archives beyond max_local_backups. Returns count deleted.""" + archives = sorted( + self.config.archive_dir.glob("roko-*.tar.gz"), + key=lambda p: p.stat().st_mtime, + reverse=True, + ) + deleted = 0 + for old in archives[self.config.max_local_backups :]: + log.info("Deleting local: %s", old.name) + old.unlink() + deleted += 1 + return deleted + + def clean_cloud(self) -> int: + """Delete cloud archives beyond max_cloud_backups. Returns count deleted.""" + result = subprocess.run( + ["rclone", "ls", self.config.rclone_remote], + capture_output=True, + text=True, + check=False, + env=_rclone_env(), + ) + if result.returncode != 0: + log.warning("Cannot list cloud backups for cleanup: %s", result.stderr[:200]) + return 0 + + entries: list[tuple[int, str]] = [] + for line in result.stdout.strip().splitlines(): + parts = line.split(None, 1) + if len(parts) == 2: + try: + entries.append((int(parts[0]), parts[1])) + except ValueError: + continue + + entries.sort(key=lambda e: e[1]) + deleted = 0 + to_remove = entries[: max(0, len(entries) - self.config.max_cloud_backups)] + + for _size, name in to_remove: + log.info("Deleting cloud: %s", name) + subprocess.run( + ["rclone", "delete", f"{self.config.rclone_remote}/{name}"], + capture_output=True, + text=True, + check=False, + env=_rclone_env(), + ) + deleted += 1 + + return deleted + + +# --- Reporter ----------------------------------------------------------------- + + +class Reporter: + """Writes JSON backup report to the workspace.""" + + def __init__(self, config: Config) -> None: + self.config = config + + def write( + self, + status: str, + name: str | None = None, + size_bytes: int | None = None, + error: str | None = None, + ) -> None: + """Write the report JSON file.""" + report = { + "timestamp": datetime.datetime.now().astimezone().isoformat(), + "status": status, + "name": name, + "size_bytes": size_bytes, + "error": error, + } + self.config.report_file.parent.mkdir(parents=True, exist_ok=True) + self.config.report_file.write_text(json.dumps(report, indent=2) + "\n") + log.info("Report written: %s (%s)", self.config.report_file, status) + + +# --- Pipeline ----------------------------------------------------------------- + + +class BackupPipeline: + """Orchestrates the full backup: collect -> archive -> upload -> verify -> clean -> report.""" + + def __init__(self, config: Config, *, dry_run: bool = False) -> None: + self.config = config + self.dry_run = dry_run + self.collector = Collector(config) + self.archiver = Archiver(config) + self.uploader = Uploader(config) + self.verifier = Verifier(config) + self.cleaner = Cleaner(config) + self.reporter = Reporter(config) + + def run(self) -> int: + """Execute the pipeline. Returns exit code.""" + backup_name = f"roko-{datetime.datetime.now().strftime('%Y%m%d-%H%M%S')}" + log.info("Starting backup: %s", backup_name) + + free = shutil.disk_usage(self.config.archive_dir).free + log.info("Free disk: %s", self._fmt_bytes(free)) + + try: + with tempfile.TemporaryDirectory(prefix="roko-backup-") as tmp: + staging = Path(tmp) + + # 1. Collect + t0 = time.monotonic() + collected, total_bytes = self.collector.collect(staging) + log.info( + "Collection done in %.1fs: %d items, %s", + time.monotonic() - t0, + len(collected), + self._fmt_bytes(total_bytes), + ) + + if self.dry_run: + log.info( + "[DRY RUN] Would archive and upload %d items (%s)", + len(collected), + self._fmt_bytes(total_bytes), + ) + for src, dest in collected: + log.info(" %s -> %s", src, dest.relative_to(staging)) + self.reporter.write("dry-run", name=backup_name, size_bytes=total_bytes) + return 0 + + # 2. Archive + t0 = time.monotonic() + archive_path, archive_size = self.archiver.archive(staging, backup_name) + log.info( + "Archiving done in %.1fs: %s", + time.monotonic() - t0, + self._fmt_bytes(archive_size), + ) + + # 3. Upload + t0 = time.monotonic() + if not self.uploader.upload(archive_path): + msg = "Upload failed after retries" + log.error(msg) + self.reporter.write("error", name=backup_name, size_bytes=archive_size, error=msg) + return 3 + log.info("Upload done in %.1fs", time.monotonic() - t0) + + # 4. Verify + ok, detail = self.verifier.verify(backup_name) + if not ok: + log.error("Verification failed: %s", detail) + self.reporter.write("error", name=backup_name, size_bytes=archive_size, error=detail) + return 3 + log.info("Verification: %s", detail) + except Exception: + raise + + # 5. Clean (outside tempdir so archive still exists for local cleanup) + local_deleted = self.cleaner.clean_local() + cloud_deleted = self.cleaner.clean_cloud() + log.info("Cleanup: %d local, %d cloud removed", local_deleted, cloud_deleted) + + # 6. Report + self.reporter.write("success", name=backup_name, size_bytes=archive_size) + log.info("Backup complete: %s (%s)", backup_name, self._fmt_bytes(archive_size)) + return 0 + + @staticmethod + def _fmt_bytes(n: int) -> str: + """Return human-readable byte size.""" + for unit in ("B", "KB", "MB", "GB"): + if abs(n) < 1024: + return f"{n:.1f} {unit}" + n /= 1024 # type: ignore[assignment] + return f"{n:.1f} TB" + + +# --- Logging setup ------------------------------------------------------------ + + +def setup_logging( + *, + verbose: bool, + log_file: Path, + max_size: int, + max_backups: int, +) -> None: + """Configure logging with rotating file handler and optional stdout.""" + root = logging.getLogger() + root.setLevel(logging.DEBUG) + + fmt = logging.Formatter("%(asctime)s %(levelname)-5s %(message)s", datefmt="%Y-%m-%d %H:%M:%S") + + log_file.parent.mkdir(parents=True, exist_ok=True) + file_handler = logging.handlers.RotatingFileHandler( + log_file, + maxBytes=max_size, + backupCount=max_backups, + encoding="utf-8", + ) + file_handler.setLevel(logging.INFO) + file_handler.setFormatter(fmt) + root.addHandler(file_handler) + + if verbose: + stdout_handler = logging.StreamHandler(sys.stdout) + stdout_handler.setLevel(logging.DEBUG) + stdout_handler.setFormatter(fmt) + root.addHandler(stdout_handler) + + +# --- Main --------------------------------------------------------------------- + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + """Parse command-line arguments.""" + parser = argparse.ArgumentParser( + prog="roko-backup.py", + description="Roko backup: collect -> archive -> upload -> verify -> clean -> report", + ) + parser.add_argument( + "--config", type=Path, default=None, + help="Config file path (default: env vars / built-in defaults)", + ) + parser.add_argument( + "--dry-run", action="store_true", + help="Collect and print what would be backed up, don't archive/upload", + ) + parser.add_argument( + "--verbose", action="store_true", + help="Detailed logging output (DEBUG level, also enables stdout)", + ) + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + """Entry point.""" + args = parse_args(argv) + + if args.config: + config = Config.from_file(args.config) + else: + config = Config.from_env() + + setup_logging( + verbose=args.verbose, + log_file=config.log_file, + max_size=config.max_log_size, + max_backups=config.max_log_backups, + ) + + if not shutil.which("rclone"): + log.error("rclone not found in PATH") + return 1 + + lock = LockFile(config.lock_path) + try: + with lock: + pipeline = BackupPipeline(config, dry_run=args.dry_run) + return pipeline.run() + except LockError as exc: + log.error(str(exc)) + return 2 + except Exception as exc: + log.exception("Unexpected error: %s", exc) + Reporter(config).write("error", error=str(exc)) + return 3 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/roko-backup.py b/roko-backup.py new file mode 100755 index 0000000..58ed4a3 --- /dev/null +++ b/roko-backup.py @@ -0,0 +1,727 @@ +#!/usr/bin/env python3 +"""Roko Backup Script — refactored Python rewrite. + +Collect -> archive -> upload -> verify -> clean -> report. +Runs from system crontab, standalone Python 3.10+. +""" + +from __future__ import annotations + +import abc +import argparse +import datetime +import json +import logging +import logging.handlers +import os +import re +import signal +import shutil +import subprocess +import sys +import tarfile +import tempfile +import time +from collections.abc import Generator +from contextlib import contextmanager +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, ContextManager, TypeVar + +log = logging.getLogger("roko-backup") + + +T = TypeVar("T") + + +def assert_not_none(value: T | None) -> T: + """Assert a value is not None at runtime, for type narrowing.""" + assert value is not None # pragma: no cover + return value + + +# ─── Exceptions ────────────────────────────────────────────────────────────── + + +class BackupError(Exception): + """Base exception for backup pipeline failures.""" + + +class LockError(BackupError): + """Raised when the lock file cannot be acquired.""" + + +# ─── Config ────────────────────────────────────────────────────────────────── + + +@dataclass(frozen=True) +class Config: + """All paths and limits for the backup pipeline.""" + + home: Path = field(default_factory=lambda: Path.home()) + report_file: Path = field( + default_factory=lambda: Path.home() + / ".openclaw" + / "workspace" + / "roko-backup-report.json" + ) + archive_dir: Path = field(default_factory=lambda: Path.home() / "backups") + log_file: Path = field(default_factory=lambda: Path.home() / "backups" / "roko-backup.log") + rclone_remote: str = "yabackup:for-backup/roko" + max_cloud_backups: int = 14 + max_local_backups: int = 3 + max_log_size: int = 1_048_576 # 1 MiB + max_log_backups: int = 5 + lock_path: Path = Path("/tmp/roko-backup.lock") + + @property + def workspace(self) -> Path: + """Path to the OpenCode workspace directory.""" + return self.home / ".openclaw" / "workspace" + + @property + def config_file(self) -> Path: + """Path to the main OpenCode config file.""" + return self.home / ".openclaw" / "openclaw.json" + + @property + def scripts_dir(self) -> Path: + """Path to the OpenCode scripts directory.""" + return self.home / ".openclaw" / "scripts" + + @property + def media_dir(self) -> Path: + """Path to the OpenCode media directory.""" + return self.home / ".openclaw" / "media" + + @property + def plugin_skills_dir(self) -> Path: + """Path to the OpenCode plugin-skills directory.""" + return self.home / ".openclaw" / "plugin-skills" + + @property + def agent_db(self) -> Path: + """Path to the main agent SQLite database.""" + return self.home / ".openclaw" / "agents" / "main" / "agent" / "openclaw-agent.sqlite" + + @property + def rclone_env(self) -> dict[str, str]: + """Environment variables for rclone subprocesses.""" + return { + **os.environ, + "NO_PROXY": "storage.yandexcloud.net,yandexcloud.net", + } + + @classmethod + def from_env(cls) -> Config: + """Build config from environment variables with built-in defaults.""" + home = Path.home() + return cls( + home=home, + report_file=Path( + os.environ.get( + "ROKO_REPORT_FILE", str(home / ".openclaw/workspace/roko-backup-report.json") + ) + ), + archive_dir=Path(os.environ.get("ROKO_ARCHIVE_DIR", str(home / "backups"))), + log_file=Path(os.environ.get("ROKO_LOG_FILE", str(home / "backups/roko-backup.log"))), + rclone_remote=os.environ.get("ROKO_RCLONE_REMOTE", "yabackup:for-backup/roko"), + max_cloud_backups=int(os.environ.get("ROKO_MAX_CLOUD", "14")), + max_local_backups=int(os.environ.get("ROKO_MAX_LOCAL", "3")), + max_log_size=int(os.environ.get("ROKO_MAX_LOG_SIZE", str(1_048_576))), + max_log_backups=int(os.environ.get("ROKO_MAX_LOG_BACKUPS", "5")), + ) + + @classmethod + def from_file(cls, path: Path) -> Config: + """Load config from a JSON file, falling back to defaults.""" + data: dict[str, Any] = {} + if path.exists(): + data = json.loads(path.read_text()) + home = Path(data.get("home", str(Path.home()))) + return cls( + home=home, + report_file=Path( + data.get( + "report_file", str(home / ".openclaw/workspace/roko-backup-report.json") + ) + ), + archive_dir=Path(data.get("archive_dir", str(home / "backups"))), + log_file=Path(data.get("log_file", str(home / "backups/roko-backup.log"))), + rclone_remote=data.get("rclone_remote", "yabackup:for-backup/roko"), + max_cloud_backups=data.get("max_cloud_backups", 14), + max_local_backups=data.get("max_local_backups", 3), + max_log_size=data.get("max_log_size", 1_048_576), + max_log_backups=data.get("max_log_backups", 5), + ) + + +# ─── LockFile ──────────────────────────────────────────────────────────────── + + +class LockFile(ContextManager["LockFile"]): + """Prevents concurrent backup runs via a PID file.""" + + def __init__(self, path: Path) -> None: + self.path = path + self._acquired = False + + def acquire(self) -> None: + """Acquire the lock. Raises LockError if another live process holds it.""" + if self.path.exists(): + try: + pid = int(self.path.read_text().strip()) + os.kill(pid, 0) + raise LockError(f"Another backup is already running (lock: {self.path})") + except (ValueError, ProcessLookupError, PermissionError, OSError): + log.warning("Stale lock file found (PID dead), removing") + self.path.unlink(missing_ok=True) + + self.path.write_text(str(os.getpid())) + self._acquired = True + signal.signal(signal.SIGTERM, self._signal_handler) + signal.signal(signal.SIGINT, self._signal_handler) + + def release(self) -> None: + """Release the lock and remove the file.""" + if self._acquired: + self.path.unlink(missing_ok=True) + self._acquired = False + + def _signal_handler(self, signum: int, _frame: Any) -> None: + self.release() + sys.exit(128 + signum) + + def __enter__(self) -> LockFile: + self.acquire() + return self + + def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: + self.release() + + +# ─── Collector ─────────────────────────────────────────────────────────────── + + +class Collector: + """Gathers files to back up into a staging directory.""" + + EXCLUDE_PATTERNS = ( + "node_modules", + "__pycache__", + "*.pyc", + ".git", + "roko-pulse.jsonl", + "cache", + "*.tmp", + ".cache", + ) + + def __init__(self, config: Config, exclude_patterns: tuple[str, ...] | None = None) -> None: + self.config = config + self.exclude_patterns = exclude_patterns or self.EXCLUDE_PATTERNS + + def collect(self, staging: Path) -> tuple[list[tuple[Path, Path]], int]: + """Copy workspace/config/scripts/media/plugin-skills/agent-db into staging. + + Returns: + A tuple of (list of (source, dest) pairs, total bytes collected). + """ + collected: list[tuple[Path, Path]] = [] + total_bytes = 0 + ignore = shutil.ignore_patterns(*self.exclude_patterns) + + targets: list[tuple[Path, str]] = [ + (self.config.workspace, "openclaw/workspace"), + (self.config.config_file, "openclaw/openclaw.json"), + (self.config.scripts_dir, "openclaw/scripts"), + (self.config.media_dir, "openclaw/media"), + (self.config.plugin_skills_dir, "openclaw/plugin-skills"), + (self.config.agent_db, "openclaw/openclaw-agent.sqlite"), + ] + + for src, dest_rel in targets: + dest = staging / dest_rel + if not src.exists(): + log.warning("Source not found, skipping: %s", src) + continue + + dest.parent.mkdir(parents=True, exist_ok=True) + + try: + if src.is_dir(): + shutil.copytree(src, dest, ignore=ignore, dirs_exist_ok=True) + else: + shutil.copy2(src, dest) + except Exception as exc: + log.warning("Failed to copy %s: %s", src, exc) + continue + + size = self._tree_size(dest) + collected.append((src, dest)) + total_bytes += size + log.info("Collected %s -> %s (%d bytes)", src.name, dest_rel, size) + + return collected, total_bytes + + @staticmethod + def _tree_size(path: Path) -> int: + """Return total size in bytes for a file or directory.""" + if path.is_file(): + return path.stat().st_size + return sum(f.stat().st_size for f in path.rglob("*") if f.is_file()) + + +# ─── Archiver ──────────────────────────────────────────────────────────────── + + +class Archiver: + """Creates a tar.gz archive from a staging directory.""" + + def __init__(self, config: Config) -> None: + self.config = config + + def archive(self, staging: Path, backup_name: str) -> tuple[Path, int]: + """Create tar.gz archive. Returns (archive_path, size_bytes).""" + self.config.archive_dir.mkdir(parents=True, exist_ok=True) + archive_path = self.config.archive_dir / f"{backup_name}.tar.gz" + + log.info("Creating archive: %s", archive_path) + with tarfile.open(archive_path, mode="w:gz") as tar: + tar.add(staging, arcname=".") + + size = archive_path.stat().st_size + log.info("Archive created: %s (%d bytes)", archive_path.name, size) + return archive_path, size + + +# ─── Uploader ──────────────────────────────────────────────────────────────── + + +class Uploader: + """Uploads archive via rclone to Yandex Object Storage with retry.""" + + UPLOAD_PROGRESS_INTERVAL = 30 # seconds between progress log lines + PROGRESS_PERCENT_STEP = 5 # log every N percent + + BACKOFF_SCHEDULE = (10, 30, 60) + + _PROGRESS_RE = re.compile( + r"Transferred:\s*([\d.]+)\s*(\w*)\s*/\s*([\d.]+)\s*(\w*),\s*([\d.]+)%" + ) + + def __init__(self, config: Config) -> None: + self.config = config + + def upload(self, archive_path: Path) -> bool: + """Upload archive to remote. Returns True on success.""" + max_attempts = len(self.BACKOFF_SCHEDULE) + 1 + + for attempt in range(1, max_attempts + 1): + log.info("Upload attempt %d/%d", attempt, max_attempts) + if self._upload_once(archive_path): + log.info("Upload succeeded") + return True + + if attempt < max_attempts: + delay = self.BACKOFF_SCHEDULE[min(attempt - 1, len(self.BACKOFF_SCHEDULE) - 1)] + log.info("Retrying in %ds...", delay) + time.sleep(delay) + + log.error("Upload failed after %d attempts", max_attempts) + return False + + def _upload_once(self, archive_path: Path) -> bool: + """Run a single upload attempt. Uses rclone --progress, reads stderr + with a timeout to handle carriage-return-based progress lines.""" + cmd = [ + "rclone", + "copy", + str(archive_path), + f"{self.config.rclone_remote}/", + "--stats", "30s", + "--progress", + ] + log.debug("Running: %s", " ".join(cmd)) + + process = subprocess.Popen( + cmd, + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + text=True, + bufsize=1, + env=self.config.rclone_env, + ) + + last_logged_percent = -1 + last_log_time = 0.0 + progress_lines: list[str] = [] + stderr = assert_not_none(process.stderr) + + while True: + chunk = stderr.read(4096) + if not chunk: + break + for line in chunk.splitlines(): + line = line.strip() + if not line: + continue + progress_lines.append(line) + match = self._PROGRESS_RE.search(line) + if match: + parts = match.groups() + pct = float(parts[4]) + bytes_done = f"{parts[0]} {parts[1] or 'B'}" + bytes_total = f"{parts[2]} {parts[3] or 'B'}" + now = time.monotonic() + new_bucket = int(pct) // self.PROGRESS_PERCENT_STEP * self.PROGRESS_PERCENT_STEP + if now - last_log_time >= self.UPLOAD_PROGRESS_INTERVAL or new_bucket > last_logged_percent: + log.info("Upload: %.0f%% (%s / %s)", pct, bytes_done, bytes_total) + last_logged_percent = new_bucket + last_log_time = now + if process.poll() is not None: + break + + process.wait() + if process.returncode == 0: + log.info("Upload succeeded") + return True + + log.warning("Upload failed (rc=%d)", process.returncode) + return False + + +# ─── Verifier ──────────────────────────────────────────────────────────────── + + +class Verifier: + """Confirms the uploaded file exists in the remote.""" + + def __init__(self, config: Config) -> None: + self.config = config + + def verify(self, backup_name: str) -> tuple[bool, str]: + """Check that backup_name.tar.gz is listed in the remote. + + Returns: + A tuple of (success, details). + """ + result = subprocess.run( + ["rclone", "ls", self.config.rclone_remote], + capture_output=True, + text=True, + check=False, + env=self.config.rclone_env, + ) + + if result.returncode != 0: + return False, f"rclone ls failed (rc={result.returncode}): {result.stderr[:200]}" + + target = f"{backup_name}.tar.gz" + if target in result.stdout: + return True, f"Verified: {target} found in remote" + + return False, f"Not found in remote: {target}" + + +# ─── Cleaner ───────────────────────────────────────────────────────────────── + + +class Cleaner: + """Manages retention of local and cloud backups.""" + + def __init__(self, config: Config) -> None: + self.config = config + + def clean_local(self) -> int: + """Delete local archives beyond MAX_LOCAL_BACKUPS. Returns count deleted.""" + archives = sorted( + self.config.archive_dir.glob("roko-*.tar.gz"), + key=lambda p: p.stat().st_mtime, + reverse=True, + ) + deleted = 0 + for old in archives[self.config.max_local_backups :]: + log.info("Deleting local: %s", old.name) + old.unlink() + deleted += 1 + return deleted + + def clean_cloud(self) -> int: + """Delete cloud archives beyond MAX_CLOUD_BACKUPS. Returns count deleted.""" + result = subprocess.run( + ["rclone", "ls", self.config.rclone_remote], + capture_output=True, + text=True, + check=False, + env=self.config.rclone_env, + ) + if result.returncode != 0: + log.warning("Cannot list cloud backups for cleanup: %s", result.stderr[:200]) + return 0 + + entries: list[tuple[int, str]] = [] + for line in result.stdout.strip().splitlines(): + parts = line.split(None, 1) + if len(parts) == 2: + try: + entries.append((int(parts[0]), parts[1])) + except ValueError: + continue + + entries.sort(key=lambda e: e[1]) + deleted = 0 + to_remove = entries[: max(0, len(entries) - self.config.max_cloud_backups)] + + for _size, name in to_remove: + log.info("Deleting cloud: %s", name) + subprocess.run( + ["rclone", "delete", f"{self.config.rclone_remote}/{name}"], + capture_output=True, + text=True, + check=False, + env=self.config.rclone_env, + ) + deleted += 1 + + return deleted + + +# ─── Reporter ──────────────────────────────────────────────────────────────── + + +class Reporter: + """Writes JSON backup report to the workspace.""" + + def __init__(self, config: Config) -> None: + self.config = config + + def write( + self, + status: str, + name: str | None = None, + size_bytes: int | None = None, + error: str | None = None, + ) -> None: + """Write the report JSON file.""" + report = { + "timestamp": datetime.datetime.now().astimezone().isoformat(), + "status": status, + "name": name, + "size_bytes": size_bytes, + "error": error, + } + self.config.report_file.parent.mkdir(parents=True, exist_ok=True) + self.config.report_file.write_text(json.dumps(report, indent=2) + "\n") + log.info("Report written: %s (%s)", self.config.report_file, status) + + +# ─── Pipeline ──────────────────────────────────────────────────────────────── + + +class BackupPipeline: + """Orchestrates the full backup: collect -> archive -> upload -> verify -> clean -> report.""" + + def __init__(self, config: Config, *, dry_run: bool = False) -> None: + self.config = config + self.dry_run = dry_run + self.collector = Collector(config) + self.archiver = Archiver(config) + self.uploader = Uploader(config) + self.verifier = Verifier(config) + self.cleaner = Cleaner(config) + self.reporter = Reporter(config) + + def run(self) -> int: + """Execute the pipeline. Returns exit code.""" + backup_name = f"roko-{datetime.datetime.now().strftime('%Y%m%d-%H%M%S')}" + log.info("Starting backup: %s", backup_name) + + free = shutil.disk_usage(self.config.archive_dir).free + log.info("Free disk: %s", self._fmt_bytes(free)) + + archive_path: Path | None = None + archive_size: int | None = None + + with self._managed_staging() as staging: + # 1. Collect + t0 = time.monotonic() + collected, total_bytes = self.collector.collect(staging) + log.info( + "Collection done in %.1fs: %d items, %s", + time.monotonic() - t0, + len(collected), + self._fmt_bytes(total_bytes), + ) + + if self.dry_run: + log.info( + "[DRY RUN] Would archive and upload %d items (%s)", + len(collected), + self._fmt_bytes(total_bytes), + ) + for src, dest in collected: + log.info(" %s -> %s", src, dest.relative_to(staging)) + self.reporter.write("dry-run", name=backup_name, size_bytes=total_bytes) + return 0 + + # 2. Archive + t0 = time.monotonic() + archive_path, archive_size = self.archiver.archive(staging, backup_name) + log.info( + "Archiving done in %.1fs: %s", + time.monotonic() - t0, + self._fmt_bytes(archive_size), + ) + + # 3. Upload + if archive_path is None or archive_size is None: + msg = "Archive was not created" + log.error(msg) + self.reporter.write("error", name=backup_name, error=msg) + return 3 + + t0 = time.monotonic() + if not self.uploader.upload(archive_path): + msg = "Upload failed after retries" + log.error(msg) + self.reporter.write("error", name=backup_name, size_bytes=archive_size, error=msg) + return 3 + log.info("Upload done in %.1fs", time.monotonic() - t0) + + # 4. Verify + ok, detail = self.verifier.verify(backup_name) + if not ok: + log.error("Verification failed: %s", detail) + self.reporter.write("error", name=backup_name, size_bytes=archive_size, error=detail) + return 3 + log.info("Verification: %s", detail) + + # 5. Clean + local_deleted = self.cleaner.clean_local() + cloud_deleted = self.cleaner.clean_cloud() + log.info("Cleanup: %d local, %d cloud removed", local_deleted, cloud_deleted) + + # 6. Report + self.reporter.write("success", name=backup_name, size_bytes=archive_size) + log.info("Backup complete: %s (%s)", backup_name, self._fmt_bytes(archive_size)) + return 0 + + @contextmanager + def _managed_staging(self) -> Generator[Path, None, None]: + """Create and always clean up a temporary staging directory.""" + tmp_dir = tempfile.mkdtemp(prefix="roko-backup-") + staging = Path(tmp_dir) + try: + yield staging + finally: + try: + shutil.rmtree(staging, ignore_errors=True) + except Exception as exc: + log.warning("Failed to remove staging directory %s: %s", staging, exc) + + @staticmethod + def _fmt_bytes(n: int) -> str: + """Human-readable byte size.""" + value = float(n) + for unit in ("B", "KB", "MB", "GB"): + if abs(value) < 1024: + return f"{value:.1f} {unit}" + value /= 1024 + return f"{value:.1f} TB" + + +# ─── Main ──────────────────────────────────────────────────────────────────── + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + """Parse command-line arguments.""" + parser = argparse.ArgumentParser( + prog="roko-backup.py", + description="Roko backup: collect -> archive -> upload -> verify -> clean -> report", + ) + parser.add_argument( + "--config", + type=Path, + default=None, + help="Config file path (default: env vars / built-in defaults)", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Collect and print what would be backed up, don't archive/upload", + ) + parser.add_argument( + "--verbose", + action="store_true", + help="Detailed logging output (DEBUG level, also to stdout)", + ) + return parser.parse_args(argv) + + +def setup_logging( + verbose: bool, + log_file: Path, + max_size: int, + max_backups: int, +) -> None: + """Configure rotating file logging and optional stdout output.""" + level = logging.DEBUG if verbose else logging.INFO + formatter = logging.Formatter( + "%(asctime)s %(levelname)-5s %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + + log_file.parent.mkdir(parents=True, exist_ok=True) + file_handler = logging.handlers.RotatingFileHandler( + str(log_file), + maxBytes=max_size, + backupCount=max_backups, + ) + file_handler.setFormatter(formatter) + + root = logging.getLogger() + root.setLevel(level) + root.addHandler(file_handler) + + if verbose: + stream_handler = logging.StreamHandler(sys.stdout) + stream_handler.setFormatter(formatter) + root.addHandler(stream_handler) + + +def main(argv: list[str] | None = None) -> int: + """Entry point.""" + args = parse_args(argv) + + if args.config: + config = Config.from_file(args.config) + else: + config = Config.from_env() + + setup_logging( + args.verbose, + config.log_file, + config.max_log_size, + config.max_log_backups, + ) + + if not shutil.which("rclone"): + log.error("rclone not found in PATH") + return 1 + + try: + with LockFile(config.lock_path): + pipeline = BackupPipeline(config, dry_run=args.dry_run) + return pipeline.run() + except LockError as exc: + log.error("%s", exc) + return 2 + except Exception as exc: + log.exception("Unexpected error: %s", exc) + Reporter(config).write("error", error=str(exc)) + return 3 + + +if __name__ == "__main__": + sys.exit(main())