145 lines
4.4 KiB
Python
145 lines
4.4 KiB
Python
#!/usr/bin/env -S uv run --script
|
|
# /// script
|
|
# dependencies = []
|
|
# ///
|
|
|
|
import argparse
|
|
import logging
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import zipfile
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Backup MariaDB database and website files", formatter_class=argparse.ArgumentDefaultsHelpFormatter)
|
|
parser.add_argument("--db-url", default=os.getenv("DB_URL"), help="Database URL (default: from DB_URL env var)")
|
|
parser.add_argument("--site-dir", type=Path, required=True, help="Website directory to backup")
|
|
parser.add_argument("--backup-dir", type=Path, default="/mnt/box/backup/droplet", help="Backup destination directory")
|
|
return parser.parse_args()
|
|
|
|
|
|
def parse_db_url(db_url: str) -> dict:
|
|
if not db_url:
|
|
raise ValueError("Database URL is required")
|
|
|
|
parsed = urlparse(db_url)
|
|
|
|
if parsed.scheme != "mysql":
|
|
raise ValueError("Unsupported database URL format")
|
|
|
|
return {
|
|
"host": parsed.hostname or "localhost",
|
|
"port": str(parsed.port or 3306),
|
|
"user": parsed.username or "root",
|
|
"password": parsed.password or "",
|
|
"database": parsed.path.lstrip("/") if parsed.path else "",
|
|
}
|
|
|
|
|
|
def dump_database(db_params: dict, dump_path: Path) -> None:
|
|
logger.info(f"Dumping database to {dump_path}")
|
|
|
|
cmd = [
|
|
"mysqldump",
|
|
"--host",
|
|
db_params["host"],
|
|
"--port",
|
|
db_params["port"],
|
|
"--user",
|
|
db_params["user"],
|
|
"--single-transaction",
|
|
"--routines",
|
|
"--triggers",
|
|
db_params["database"],
|
|
]
|
|
|
|
if db_params["password"]:
|
|
cmd.append(f"--password={db_params['password']}")
|
|
|
|
try:
|
|
with dump_path.open("w") as f:
|
|
subprocess.run(cmd, stdout=f, stderr=subprocess.PIPE, text=True, check=True)
|
|
logger.info("Database dump completed successfully")
|
|
except subprocess.CalledProcessError as e:
|
|
logger.error(f"Database dump failed: {e.stderr}")
|
|
raise
|
|
|
|
|
|
def zip_directory(source_dir: Path, zip_path: Path) -> None:
|
|
"""Zip a directory"""
|
|
logger.info(f"Zipping directory {source_dir} to {zip_path}")
|
|
|
|
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zipf:
|
|
for file_path in source_dir.rglob("*"):
|
|
if file_path.is_file():
|
|
# Use relative path within the zip
|
|
arcname = file_path.relative_to(source_dir.parent)
|
|
zipf.write(file_path, arcname)
|
|
|
|
logger.info(f"Directory zipped successfully: {zip_path}")
|
|
|
|
|
|
def create_final_backup(temp_dir: Path, backup_dir: Path, timestamp: str) -> Path:
|
|
final_backup_name = f"{timestamp}_ucsuzkalem.zip"
|
|
final_backup_path = backup_dir / final_backup_name
|
|
|
|
logger.info(f"Creating final backup: {final_backup_path}")
|
|
zip_directory(temp_dir, final_backup_path)
|
|
|
|
return final_backup_path
|
|
|
|
|
|
def main():
|
|
args = parse_args()
|
|
|
|
if not args.db_url:
|
|
raise ValueError("Database URL is required (set DB_URL env var or use --db-url)")
|
|
|
|
# Parse database URL
|
|
db_params = parse_db_url(args.db_url)
|
|
logger.info(f"Connecting to database: {db_params['host']}:{db_params['port']}/{db_params['database']}")
|
|
|
|
# Create timestamp
|
|
timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
|
|
|
|
# Setup paths
|
|
site_dir = Path(args.site_dir)
|
|
backup_dir = Path(args.backup_dir)
|
|
temp_dir = Path("/tmp") / f"backup_{timestamp}"
|
|
|
|
if not site_dir.exists():
|
|
raise FileNotFoundError(f"Site directory does not exist: {site_dir}")
|
|
|
|
# Create directories
|
|
backup_dir.mkdir(parents=True, exist_ok=True)
|
|
temp_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Dump database
|
|
db_dump_path = temp_dir / "database.sql"
|
|
dump_database(db_params, db_dump_path)
|
|
|
|
# Zip website files
|
|
|
|
files_zip_path = temp_dir / "files.zip"
|
|
zip_directory(site_dir, files_zip_path)
|
|
|
|
# Create final backup
|
|
final_backup_path = create_final_backup(temp_dir, backup_dir, timestamp)
|
|
|
|
# Cleanup temp directory
|
|
shutil.rmtree(temp_dir)
|
|
|
|
logger.info(f"Backup completed successfully: {final_backup_path}")
|
|
logger.info(f"Backup size: {final_backup_path.stat().st_size / (1024 * 1024):.1f} MB")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
exit(main())
|