Hidosi/roko-backup-script
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial: Roko Backup Script v2 — Python rewrite with streaming progress & log rotation

Browse Source 
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
This commit is contained in:
Roko
2026-07-07 11:27:02 +07:00
commit a88b2604a1
5 changed files with 2362 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+118
View File
@@ -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
TECH_SPEC.md
+94
View File
@@ -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)
roko-backup-kimi.py
+741
View File
@@ -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())
roko-backup-mimo.py
+682
View File
@@ -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())
roko-backup.py
Executable
+727
View File
@@ -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())