From e3d4dd03b72383421d5bfd0a51607abd64f95133 Mon Sep 17 00:00:00 2001 From: "vladimir.buzalka" Date: Wed, 28 Jan 2026 15:20:25 +0100 Subject: [PATCH] z230 --- 10 Backup/20 KanBoardDestroyAndRestore.py | 152 ++++++++++++++++++++++ 10 Backup/30 MySQLInteractiveBackup.py | 148 +++++++++++++++++++++ 2 files changed, 300 insertions(+) create mode 100644 10 Backup/20 KanBoardDestroyAndRestore.py create mode 100644 10 Backup/30 MySQLInteractiveBackup.py diff --git a/10 Backup/20 KanBoardDestroyAndRestore.py b/10 Backup/20 KanBoardDestroyAndRestore.py new file mode 100644 index 0000000..267282e --- /dev/null +++ b/10 Backup/20 KanBoardDestroyAndRestore.py @@ -0,0 +1,152 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +import os +import gzip +import subprocess +from pathlib import Path + +# ============================================================ +# CESTY K BINÁRKÁM (PORTABLE) +# ============================================================ + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +MYSQL = PROJECT_ROOT / "bin" / "mysql.exe" + +# ============================================================ +# 🔧 KONFIGURACE – UPRAV SI JEN TADY +# ============================================================ + +DUMP_FILE = Path( + r"U:\MySQLBackup\Kanboard\kanboard_kanboard_2026-01-28_09-53-50.sql.gz" +) + +# ============================================================ + + +def load_dotenv(dotenv_path: Path) -> None: + if not dotenv_path.exists(): + return + for line in dotenv_path.read_text(encoding="utf-8").splitlines(): + line = line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + k, v = line.split("=", 1) + os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'")) + + +def require_env(name: str) -> str: + v = os.getenv(name) + if not v: + raise SystemExit(f"Chybí proměnná {name} (dej ji do .env nebo env).") + return v + + +def extract_db_name_from_dump(dump_file: Path) -> str: + """ + Najde první 'CREATE DATABASE `xxx`' v dumpu + """ + with gzip.open(dump_file, "rb") as f: + for raw in f: + line = raw.decode("utf-8", errors="ignore").strip() + if line.upper().startswith("CREATE DATABASE"): + return line.split("`")[1] + raise SystemExit("Nepodařilo se zjistit jméno DB z dumpu.") + + +def database_exists(cnf_path: Path, db_name: str) -> bool: + """ + Ověří existenci DB v MySQL + """ + cmd = [ + str(MYSQL), + f"--defaults-extra-file={cnf_path}", + "-N", + "-B", + "-e", + f"SHOW DATABASES LIKE '{db_name}';", + ] + p = subprocess.run(cmd, capture_output=True, text=True) + return db_name in (p.stdout or "") + + +def main() -> None: + if not DUMP_FILE.exists(): + raise SystemExit(f"Dump neexistuje: {DUMP_FILE}") + + if not DUMP_FILE.name.endswith(".sql.gz"): + raise SystemExit("Dump musí končit na .sql.gz") + + here = Path(__file__).resolve().parent + load_dotenv(here / ".env") + + host = require_env("KB_MYSQL_HOST") + port = require_env("KB_MYSQL_PORT") + user = require_env("KB_MYSQL_USER") + password = require_env("KB_MYSQL_PASSWORD") + + # dočasný mysql client config (bez hesla v CLI) + cnf_path = here / ".mysql_restore.cnf" + cnf_path.write_text( + "[client]\n" + f"host={host}\n" + f"port={port}\n" + f"user={user}\n" + f"password={password}\n", + encoding="utf-8" + ) + + tmp_sql = DUMP_FILE.with_suffix(".restore.tmp.sql") + + try: + try: + os.chmod(cnf_path, 0o600) + except Exception: + pass + + # 🔎 zjistíme jméno DB z dumpu + db_name = extract_db_name_from_dump(DUMP_FILE) + print(f"[INFO] Dump obsahuje databázi: {db_name}") + + # 🛑 POJISTKA – DB už existuje + if database_exists(cnf_path, db_name): + raise SystemExit( + f"❌ STOP: Databáze '{db_name}' už v MySQL existuje.\n" + f" Musíš ji NEJDŘÍVE ručně smazat (DROP DATABASE {db_name};)\n" + f" a teprve pak znovu spustit restore." + ) + + print("[OK] Databáze v MySQL neexistuje → pokračuji v obnově") + + # ==================================================== + # 1) ROZBALENÍ DUMPU DO DOČASNÉHO SQL + # ==================================================== + print("[INFO] Rozbaluji dump do dočasného SQL souboru") + with gzip.open(DUMP_FILE, "rb") as f_in, open(tmp_sql, "wb") as f_out: + f_out.write(f_in.read()) + + # ==================================================== + # 2) RESTORE Z ČISTÉHO SQL + # ==================================================== + mysql_cmd = [ + str(MYSQL), + f"--defaults-extra-file={cnf_path}", + "--binary-mode=1", + ] + + print("[OK] Spouštím restore z dočasného SQL") + with open(tmp_sql, "rb") as f_sql: + p = subprocess.run(mysql_cmd, stdin=f_sql) + + if p.returncode != 0: + raise SystemExit(f"mysql import selhal (exit={p.returncode})") + + print(f"[OK] Obnova databáze '{db_name}' dokončena.") + + finally: + tmp_sql.unlink(missing_ok=True) + cnf_path.unlink(missing_ok=True) + + +if __name__ == "__main__": + main() diff --git a/10 Backup/30 MySQLInteractiveBackup.py b/10 Backup/30 MySQLInteractiveBackup.py new file mode 100644 index 0000000..07f5a7a --- /dev/null +++ b/10 Backup/30 MySQLInteractiveBackup.py @@ -0,0 +1,148 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +import os +import gzip +import subprocess +from datetime import datetime +from pathlib import Path + +# ============================================================ +# CESTY K BINÁRKÁM (PORTABLE) +# ============================================================ + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +MYSQL = PROJECT_ROOT / "bin" / "mysql.exe" +MYSQLDUMP = PROJECT_ROOT / "bin" / "mysqldump.exe" + +# ============================================================ +# KONFIGURACE +# ============================================================ + +BACKUP_ROOT = Path(r"U:\mysqlbackup") + +# ============================================================ + + +def load_dotenv(dotenv_path: Path) -> None: + if not dotenv_path.exists(): + return + for line in dotenv_path.read_text(encoding="utf-8").splitlines(): + line = line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + k, v = line.split("=", 1) + os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'")) + + +def require_env(name: str) -> str: + v = os.getenv(name) + if not v: + raise SystemExit(f"Chybí proměnná {name} (dej ji do .env nebo env).") + return v + + +def get_mysql_cnf(here: Path) -> Path: + cnf = here / ".mysql_backup.cnf" + cnf.write_text( + "[client]\n" + f"host={require_env('KB_MYSQL_HOST')}\n" + f"port={require_env('KB_MYSQL_PORT')}\n" + f"user={require_env('KB_MYSQL_USER')}\n" + f"password={require_env('KB_MYSQL_PASSWORD')}\n", + encoding="utf-8" + ) + try: + os.chmod(cnf, 0o600) + except Exception: + pass + return cnf + + +def list_databases(cnf_path: Path) -> list[str]: + cmd = [ + str(MYSQL), + f"--defaults-extra-file={cnf_path}", + "-N", + "-B", + "-e", + "SHOW DATABASES;", + ] + p = subprocess.run(cmd, capture_output=True, text=True) + if p.returncode != 0: + raise SystemExit("Nelze vylistovat databáze.") + return [ + db.strip() + for db in p.stdout.splitlines() + if db and db not in ("information_schema", "mysql", "performance_schema", "sys") + ] + + +def choose_database(dbs: list[str]) -> str: + print("\nDOSTUPNÉ DATABÁZE:\n") + for i, db in enumerate(dbs, start=1): + print(f" {i:2d}) {db}") + print() + + while True: + choice = input("Vyber číslo databáze k záloze: ").strip() + if not choice.isdigit(): + continue + idx = int(choice) + if 1 <= idx <= len(dbs): + return dbs[idx - 1] + + +def main() -> None: + here = Path(__file__).resolve().parent + load_dotenv(here / ".env") + + cnf_path = get_mysql_cnf(here) + + try: + dbs = list_databases(cnf_path) + if not dbs: + raise SystemExit("Nenalezena žádná databáze k záloze.") + + db_name = choose_database(dbs) + print(f"\n[OK] Vybraná databáze: {db_name}") + + # cílový adresář + target_dir = BACKUP_ROOT / db_name + target_dir.mkdir(parents=True, exist_ok=True) + + ts = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") + dump_path = target_dir / f"{db_name}_{ts}.sql.gz" + + dump_cmd = [ + str(MYSQLDUMP), + f"--defaults-extra-file={cnf_path}", + "--single-transaction", + "--routines", + "--triggers", + "--events", + "--hex-blob", + "--default-character-set=utf8mb4", + "--set-gtid-purged=OFF", + db_name, + ] + + print(f"[OK] Spouštím backup → {dump_path}") + + with subprocess.Popen(dump_cmd, stdout=subprocess.PIPE) as proc: + with gzip.open(dump_path, "wb", compresslevel=6) as gz: + gz.write(proc.stdout.read()) + + rc = proc.wait() + if rc != 0: + raise SystemExit(f"mysqldump selhal (exit={rc})") + + size_mb = dump_path.stat().st_size / 1024 / 1024 + print(f"[OK] Hotovo ({size_mb:.2f} MB)") + + finally: + cnf_path.unlink(missing_ok=True) + + +if __name__ == "__main__": + main()