z230

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()