feat(07-03): convert export_prod.sh to Python with YAML config parsing
This commit is contained in:
127
scripts/export_prod.py
Executable file
127
scripts/export_prod.py
Executable file
@@ -0,0 +1,127 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""
|
||||||
|
TFM aInventory - Production Export/Backup Script (v1.12.0)
|
||||||
|
Converted from export_prod.sh (as a backup tool) to Python per Decision D-05.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
import yaml
|
||||||
|
import argparse
|
||||||
|
import tarfile
|
||||||
|
import logging
|
||||||
|
from datetime import datetime
|
||||||
|
from typing import Dict, Any, List
|
||||||
|
|
||||||
|
# Color codes
|
||||||
|
class Colors:
|
||||||
|
RED = '\033[0;31m'
|
||||||
|
GREEN = '\033[0;32m'
|
||||||
|
YELLOW = '\033[1;33m'
|
||||||
|
BLUE = '\033[0;34m'
|
||||||
|
NC = '\033[0m'
|
||||||
|
|
||||||
|
# Setup logging
|
||||||
|
logging.basicConfig(level=logging.INFO, format=f"{Colors.BLUE}[INFO]{Colors.NC} %(message)s")
|
||||||
|
logger = logging.getLogger("export_prod")
|
||||||
|
|
||||||
|
def load_yaml(file_path: str) -> Dict[str, Any]:
|
||||||
|
if not os.path.exists(file_path):
|
||||||
|
return {}
|
||||||
|
try:
|
||||||
|
with open(file_path, 'r') as f:
|
||||||
|
return yaml.safe_load(f) or {}
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(f"Failed to parse {file_path}: {e}")
|
||||||
|
return {}
|
||||||
|
|
||||||
|
def main():
|
||||||
|
parser = argparse.ArgumentParser(description="TFM aInventory Production Export/Backup Tool")
|
||||||
|
parser.add_argument("--output", help="Path to the output tar.gz file")
|
||||||
|
parser.add_argument("--include-logs", action="store_true", help="Include log files in the backup")
|
||||||
|
|
||||||
|
args = parser.parse_args()
|
||||||
|
|
||||||
|
# Load config to find data and logs directories
|
||||||
|
backend_cfg = load_yaml("config/backend.yaml")
|
||||||
|
app_cfg = backend_cfg.get("application", {})
|
||||||
|
|
||||||
|
data_dir = app_cfg.get("data_dir", "./data")
|
||||||
|
logs_dir = app_cfg.get("logs_dir", "./logs")
|
||||||
|
|
||||||
|
timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
|
||||||
|
|
||||||
|
if args.output:
|
||||||
|
output_path = args.output
|
||||||
|
else:
|
||||||
|
os.makedirs("backups", exist_ok=True)
|
||||||
|
output_path = f"backups/ainventory_{timestamp}.tar.gz"
|
||||||
|
|
||||||
|
logger.info(f"Creating production backup: {output_path}")
|
||||||
|
|
||||||
|
# Files and directories to include
|
||||||
|
to_include = []
|
||||||
|
|
||||||
|
# Data directory (contains database)
|
||||||
|
if os.path.exists(data_dir):
|
||||||
|
to_include.append(data_dir)
|
||||||
|
else:
|
||||||
|
logger.warning(f"Data directory {data_dir} not found. Skipping.")
|
||||||
|
|
||||||
|
# Config files (specific ones)
|
||||||
|
config_files = [
|
||||||
|
"config/backend.yaml",
|
||||||
|
"config/frontend.yaml",
|
||||||
|
"config/network.yaml",
|
||||||
|
"config/backend.yaml.example",
|
||||||
|
"config/frontend.yaml.example",
|
||||||
|
"config/network.yaml.example",
|
||||||
|
"config/docker.yaml.example",
|
||||||
|
"config/secrets.yaml.example"
|
||||||
|
]
|
||||||
|
for cf in config_files:
|
||||||
|
if os.path.exists(cf):
|
||||||
|
to_include.append(cf)
|
||||||
|
|
||||||
|
# Optional logs
|
||||||
|
if args.include_logs and os.path.exists(logs_dir):
|
||||||
|
to_include.append(logs_dir)
|
||||||
|
|
||||||
|
# Exclude patterns
|
||||||
|
exclude_files = ["config/secrets.yaml"]
|
||||||
|
exclude_dirs = ["node_modules", "__pycache__", ".git", ".venv", ".next", ".pytest_cache"]
|
||||||
|
|
||||||
|
def filter_tar(tarinfo):
|
||||||
|
# Exclude specific files
|
||||||
|
if tarinfo.name in exclude_files:
|
||||||
|
logger.info(f" Excluding sensitive file: {tarinfo.name}")
|
||||||
|
return None
|
||||||
|
|
||||||
|
# Exclude directories by name
|
||||||
|
for d in exclude_dirs:
|
||||||
|
if f"/{d}/" in f"/{tarinfo.name}/":
|
||||||
|
return None
|
||||||
|
|
||||||
|
return tarinfo
|
||||||
|
|
||||||
|
try:
|
||||||
|
with tarfile.open(output_path, "w:gz") as tar:
|
||||||
|
for item in to_include:
|
||||||
|
logger.info(f" Adding {item}...")
|
||||||
|
tar.add(item, filter=filter_tar)
|
||||||
|
|
||||||
|
# Check size
|
||||||
|
size_bytes = os.path.getsize(output_path)
|
||||||
|
size_mb = size_bytes / (1024 * 1024)
|
||||||
|
|
||||||
|
logger.info(f"{Colors.GREEN}✓ Backup created successfully!{Colors.NC}")
|
||||||
|
logger.info(f" Archive: {output_path}")
|
||||||
|
logger.info(f" Size: {size_mb:.2f} MB")
|
||||||
|
logger.info(f" Contents: Data and config (secrets excluded)")
|
||||||
|
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(f"Failed to create backup: {e}")
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
Reference in New Issue
Block a user