"""Резервные копии по расписанию. Настройка лежит в CONFIG_DIR/backup_schedule.json, итог последнего запуска — в CONFIG_DIR/backup_schedule_state.json. Копии делает не панель, а таймер virtuality-backup.timer раз в 15 минут: `/opt/virtuality/venv/bin/python /opt/virtuality/web/features/backup_schedule.py`. Так ночной перезапуск панели при обновлении не обрывает копию на середине и не оставляет машину выключенной. Автокопии помечены в meta.json (auto: true); после каждой удачной копии у машины остаются последние N автокопий. Ручные копии расписание не трогает. """ import fcntl import json import re import sys from datetime import datetime, timedelta from pathlib import Path from typing import Any if __name__ == "__main__": # запуск как скрипта: нужен web/ в sys.path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from fastapi import APIRouter, Form, Request # noqa: E402 import core # noqa: E402 from features import backups # noqa: E402 router = APIRouter() TIME_RE = re.compile(r"([01]\d|2[0-3]):([0-5]\d)") CATCH_UP = timedelta(hours=2) # сервер был выключен в назначенное время: догоняем не позже чем через 2 часа, иначе ждём следующего раза KEEP_MIN, KEEP_MAX = 1, 60 WEEKDAYS = ("по понедельникам", "по вторникам", "по средам", "по четвергам", "по пятницам", "по субботам", "по воскресеньям") WEEKDAY_NAMES = ("Понедельник", "Вторник", "Среда", "Четверг", "Пятница", "Суббота", "Воскресенье") DEFAULT: dict[str, Any] = {"enabled": False, "frequency": "daily", "weekday": 6, "time": "03:00", "all_vms": True, "vms": [], "keep": 7} NOTE = "По расписанию" # ---------------------------------------------------------------- хранение def schedule_path() -> Path: return core.CONFIG_DIR / "backup_schedule.json" def state_path() -> Path: return core.CONFIG_DIR / "backup_schedule_state.json" def lock_path() -> Path: return core.CONFIG_DIR / "backup_schedule.lock" def read_json(path: Path) -> dict[str, Any]: try: data = json.loads(path.read_text()) except (OSError, ValueError): return {} return data if isinstance(data, dict) else {} def write_json(path: Path, data: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) tmp_path = path.with_suffix(".tmp") tmp_path.write_text(json.dumps(data, ensure_ascii=False, indent=2)) tmp_path.replace(path) def is_int(value: Any) -> bool: return isinstance(value, int) and not isinstance(value, bool) def clean_schedule(raw: dict[str, Any]) -> dict[str, Any]: """Настройка всегда полная и корректная: испорченные значения заменяются значениями по умолчанию.""" schedule = dict(DEFAULT) schedule["enabled"] = raw.get("enabled") is True if raw.get("frequency") in ("daily", "weekly"): schedule["frequency"] = raw["frequency"] if is_int(raw.get("weekday")) and 0 <= raw["weekday"] <= 6: schedule["weekday"] = raw["weekday"] if isinstance(raw.get("time"), str) and TIME_RE.fullmatch(raw["time"]): schedule["time"] = raw["time"] schedule["all_vms"] = raw.get("all_vms") is not False vms = raw.get("vms") schedule["vms"] = sorted({vm for vm in vms if isinstance(vm, str) and core.valid_vm_name(vm)}) if isinstance(vms, list) else [] if is_int(raw.get("keep")) and KEEP_MIN <= raw["keep"] <= KEEP_MAX: schedule["keep"] = raw["keep"] return schedule def load_schedule() -> dict[str, Any]: return clean_schedule(read_json(schedule_path())) # ---------------------------------------------------------------- когда копировать def step(schedule: dict[str, Any]) -> timedelta: return timedelta(days=7 if schedule["frequency"] == "weekly" else 1) def last_slot(schedule: dict[str, Any], now: datetime) -> datetime: """Последний назначенный момент копирования, не позже now (время сервера).""" hour, minute = map(int, schedule["time"].split(":")) slot = now.replace(hour=hour, minute=minute, second=0, microsecond=0) if schedule["frequency"] == "weekly": slot -= timedelta(days=(slot.weekday() - schedule["weekday"]) % 7) if slot > now: slot -= step(schedule) return slot def next_slot(schedule: dict[str, Any], now: datetime) -> datetime: return last_slot(schedule, now) + step(schedule) def slot_key(slot: datetime) -> str: return slot.strftime("%Y-%m-%d %H:%M") def due(schedule: dict[str, Any], state: dict[str, Any], now: datetime) -> tuple[str, datetime] | None: """("run", момент) — пора копировать; ("missed", момент) — время прошло, пока сервер был выключен; None — нечего делать.""" if not schedule["enabled"]: return None slot = last_slot(schedule, now) if state.get("slot") == slot_key(slot): return None return ("run" if now - slot <= CATCH_UP else "missed"), slot def selected_vms(schedule: dict[str, Any]) -> list[str]: existing = backups.existing_vm_names() return sorted(existing if schedule["all_vms"] else existing & set(schedule["vms"])) # ---------------------------------------------------------------- запуск def prune(vm: str, keep: int) -> list[str]: """Удалить автокопии машины сверх последних keep. Ручные копии и копии, с которыми идёт работа, остаются.""" autos = [item for item in backups.list_backups(vm) if item["auto"] and item["complete"]] # новые сверху removed = [] for item in autos[keep:]: try: backups.delete_backup(vm, item["id"]) except backups.BackupError: continue removed.append(item["id"]) return removed def run_due(now: datetime | None = None) -> dict[str, Any] | None: """Сделать копии, если подошло время. Возвращает итог запуска или None, если делать нечего.""" lock_path().parent.mkdir(parents=True, exist_ok=True) with lock_path().open("w") as lock: try: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) except OSError: return None # предыдущий запуск ещё копирует now = now or datetime.now() schedule = load_schedule() state = read_json(state_path()) found = due(schedule, state, now) if not found: return None kind, slot = found if kind == "missed": state.update(slot=slot_key(slot), missed=slot_key(slot)) write_json(state_path(), state) return None state = {"slot": slot_key(slot), "status": "running", "started_at": core.utc_now(), "results": []} write_json(state_path(), state) for vm in selected_vms(schedule): result: dict[str, Any] = {"vm": vm} try: operation = backups.create_backup(vm, NOTE, wait=True, auto=True) except Exception as exc: # noqa: BLE001 — одна машина не должна сорвать копии остальных result.update(ok=False, message=str(exc)[:240]) else: result.update(ok=operation.get("status") == "success", message=str(operation.get("message") or "")[:240], operation_id=operation["id"]) if result["ok"]: result["removed"] = prune(vm, schedule["keep"]) state["results"].append(result) write_json(state_path(), state) state.update(status="done", finished_at=core.utc_now()) write_json(state_path(), state) return state # ---------------------------------------------------------------- как показать def format_slot(slot: datetime) -> str: return f"{slot.day} {backups.MONTHS[slot.month - 1]}, {slot:%H:%M}" def format_key(key: str) -> str: try: return format_slot(datetime.strptime(key, "%Y-%m-%d %H:%M")) except (TypeError, ValueError): return "" def when_text(schedule: dict[str, Any]) -> str: day = "каждый день" if schedule["frequency"] == "daily" else WEEKDAYS[schedule["weekday"]] return f"{day} в {schedule['time']}" def overview(now: datetime | None = None) -> dict[str, Any]: """Всё для карточки на странице «Резервные копии» и на странице машины.""" now = now or datetime.now() schedule = load_schedule() state = read_json(state_path()) results = [item for item in state.get("results") or [] if isinstance(item, dict)] failed = [item for item in results if not item.get("ok")] last = None if state.get("started_at"): last = { "when": format_key(str(state.get("slot") or "")), "running": state.get("status") == "running", "count": len(results), "ok": len(results) - len(failed), "failed": failed, } return { **schedule, "when": when_text(schedule), "next": format_slot(next_slot(schedule, now)) if schedule["enabled"] else "", "last": last, "missed": format_key(str(state.get("missed") or "")) if state.get("missed") and state.get("missed") == state.get("slot") else "", } def covers(schedule: dict[str, Any], vm: str) -> bool: return schedule["enabled"] and (schedule["all_vms"] or vm in schedule["vms"]) # ---------------------------------------------------------------- страницы def form_page(request: Request, form: dict[str, Any], error: str = "", status_code: int = 200): names = sorted(backups.existing_vm_names() | set(form["vms"])) context = {"page_title": "Расписание копий", "form": form, "vm_names": names, "weekdays": WEEKDAY_NAMES, "keep_max": KEEP_MAX, "error": error} return core.render(request, "backup_schedule.html", context, status_code=status_code) @router.get("/backups/schedule") def schedule_page(request: Request): auth = core.require_auth(request) if auth: return auth return form_page(request, load_schedule()) def read_form(enabled: str, frequency: str, weekday: str, time: str, machines: str, vms: list[str], keep: str) -> dict[str, Any]: return { "enabled": enabled == "1", "frequency": frequency if frequency in ("daily", "weekly") else "daily", "weekday": int(weekday) if weekday.isdigit() and int(weekday) <= 6 else DEFAULT["weekday"], "time": time.strip(), "all_vms": machines != "chosen", "vms": sorted({vm for vm in vms if core.valid_vm_name(vm)}), "keep": int(keep) if keep.strip().isdigit() else keep.strip(), } def form_error(form: dict[str, Any]) -> str: if not TIME_RE.fullmatch(form["time"]): return "Укажите время в виде ЧЧ:ММ, например 03:00" if not is_int(form["keep"]) or not KEEP_MIN <= form["keep"] <= KEEP_MAX: return f"Сколько копий хранить: число от {KEEP_MIN} до {KEEP_MAX}" if form["enabled"] and not form["all_vms"] and not form["vms"]: return "Отметьте хотя бы одну машину или выберите «Все машины»" return "" def same_timing(old: dict[str, Any], new: dict[str, Any]) -> bool: return old["enabled"] and new["enabled"] and all(old[key] == new[key] for key in ("frequency", "weekday", "time")) @router.post("/backups/schedule") def schedule_save( request: Request, enabled: str = Form(""), frequency: str = Form("daily"), weekday: str = Form("6"), time: str = Form("03:00"), machines: str = Form("all"), vms: list[str] = Form([]), keep: str = Form("7"), ): auth = core.require_auth(request) if auth: return auth form = read_form(enabled, frequency, weekday, time, machines, vms, keep) error = form_error(form) if error: return form_page(request, form, error, status_code=400) old, schedule = load_schedule(), clean_schedule(form) write_json(schedule_path(), schedule) now = datetime.now() if not same_timing(old, schedule): # Уже прошедший момент нового расписания не догоняем: первая копия — в следующее назначенное время. state = read_json(state_path()) state["slot"] = slot_key(last_slot(schedule, now)) write_json(state_path(), state) if not schedule["enabled"]: return core.redirect_with_message("/backups", "backup_message", "Автоматические копии выключены") message = f"Расписание сохранено: {when_text(schedule)}, следующая копия — {format_slot(next_slot(schedule, now))}" return core.redirect_with_message("/backups", "backup_message", message) def main() -> None: """Точка входа virtuality-backup.service: итог пишется в журнал службы.""" outcome = run_due() if outcome: done = sum(1 for item in outcome["results"] if item.get("ok")) print(f"Копии по расписанию {outcome['slot']}: {done} из {len(outcome['results'])} успешно") for item in outcome["results"]: if not item.get("ok"): print(f" {item['vm']}: {item.get('message')}") if __name__ == "__main__": # запускает virtuality-backup.timer main()