dsyoon
3c9d9283bd
Enable scheduled backups of ncue.net databases and git.ncue.net repository mirrors via .env-driven shell scripts. Co-authored-by: Cursor <cursoragent@cursor.com>
131 lines
4.0 KiB
Python
131 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
|
"""MariaDB logical dump fallback when mysqldump is unavailable."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import gzip
|
|
import sys
|
|
from typing import Iterable
|
|
|
|
import pymysql
|
|
|
|
|
|
SYSTEM_SCHEMAS = {"information_schema", "performance_schema", "sys", "mysql"}
|
|
|
|
|
|
def sql_quote(value) -> str:
|
|
"""Return a SQL literal for a Python value."""
|
|
if value is None:
|
|
return "NULL"
|
|
if isinstance(value, (int, float)):
|
|
return str(value)
|
|
if isinstance(value, (bytes, bytearray)):
|
|
return "0x" + value.hex()
|
|
escaped = (
|
|
str(value)
|
|
.replace("\\", "\\\\")
|
|
.replace("\0", "\\0")
|
|
.replace("\n", "\\n")
|
|
.replace("\r", "\\r")
|
|
.replace("\t", "\\t")
|
|
.replace("'", "\\'")
|
|
.replace('"', '\\"')
|
|
)
|
|
return f"'{escaped}'"
|
|
|
|
|
|
def list_databases(conn) -> list[str]:
|
|
"""List user databases excluding system schemas."""
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT schema_name
|
|
FROM information_schema.schemata
|
|
WHERE schema_name NOT IN (%s, %s, %s, %s)
|
|
ORDER BY schema_name
|
|
""",
|
|
tuple(SYSTEM_SCHEMAS),
|
|
)
|
|
return [row[0] for row in cur.fetchall()]
|
|
|
|
|
|
def dump_database(host: str, port: int, user: str, password: str, database: str, output_path: str) -> None:
|
|
"""Dump one database to a gzip SQL file."""
|
|
conn = pymysql.connect(
|
|
host=host,
|
|
port=port,
|
|
user=user,
|
|
password=password,
|
|
database=database,
|
|
charset="utf8mb4",
|
|
connect_timeout=30,
|
|
)
|
|
try:
|
|
lines: list[str] = [
|
|
"-- MariaDB dump via ncue_backup Python fallback",
|
|
"SET NAMES utf8mb4;",
|
|
"SET FOREIGN_KEY_CHECKS=0;",
|
|
f"CREATE DATABASE IF NOT EXISTS `{database}`;",
|
|
f"USE `{database}`;",
|
|
]
|
|
|
|
with conn.cursor() as cur:
|
|
cur.execute("SHOW FULL TABLES WHERE Table_type = 'BASE TABLE'")
|
|
tables = [row[0] for row in cur.fetchall()]
|
|
|
|
for table in tables:
|
|
cur.execute(f"SHOW CREATE TABLE `{table}`")
|
|
create_sql = cur.fetchone()[1]
|
|
lines.append(f"DROP TABLE IF EXISTS `{table}`;")
|
|
lines.append(f"{create_sql};")
|
|
|
|
cur.execute(f"SELECT * FROM `{table}`")
|
|
columns = [desc[0] for desc in cur.description]
|
|
col_list = ", ".join(f"`{col}`" for col in columns)
|
|
|
|
while True:
|
|
rows = cur.fetchmany(500)
|
|
if not rows:
|
|
break
|
|
for row in rows:
|
|
values = ", ".join(sql_quote(value) for value in row)
|
|
lines.append(f"INSERT INTO `{table}` ({col_list}) VALUES ({values});")
|
|
|
|
lines.append("SET FOREIGN_KEY_CHECKS=1;")
|
|
payload = "\n".join(lines) + "\n"
|
|
with gzip.open(output_path, "wt", encoding="utf-8") as fh:
|
|
fh.write(payload)
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def main(argv: Iterable[str] | None = None) -> int:
|
|
"""CLI entrypoint."""
|
|
parser = argparse.ArgumentParser(description="MariaDB dump fallback")
|
|
parser.add_argument("--host", required=True)
|
|
parser.add_argument("--port", type=int, default=3306)
|
|
parser.add_argument("--user", required=True)
|
|
parser.add_argument("--password", required=True)
|
|
parser.add_argument("--database", required=True)
|
|
parser.add_argument("--output", required=True)
|
|
args = parser.parse_args(list(argv) if argv is not None else None)
|
|
|
|
dump_database(
|
|
host=args.host,
|
|
port=args.port,
|
|
user=args.user,
|
|
password=args.password,
|
|
database=args.database,
|
|
output_path=args.output,
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
raise SystemExit(main())
|
|
except pymysql.MySQLError as exc:
|
|
print(f"mr_dump.py: database error: {exc}", file=sys.stderr)
|
|
raise SystemExit(1) from exc
|