Restore web bằng Restic và Rclone với Google Drive lưu trữ

Restore web bằng Restic

Trong phần 1 và phần 2, mình đã hướng dẫn các bạn các để sử dụng Restic backup web trên DirectAdmin và gửi backup lên Google Drive. Ở phần này, mình sẽ tạo script restore web bằng python, chi tiết script:

#!/usr/bin/env python3
import subprocess
import os
import logging
import json
from pathlib import Path
from datetime import datetime
import requests
import time
import shutil

# Cấu hình giống script backup
CONFIG = {
'restic_bin': '/usr/local/bin/restic',
'rclone_bin': '/usr/local/bin/rclone',
'repository': 'rclone:gd:/restic-backups',
'password': 'i2bdi2oa(23J',
'rclone_config': '/root/.config/rclone/rclone.conf',
'mysql': {
'user': 'da_admin',
'password': 'qcqw832hjasUancis',
'host': 'localhost',
'port': 3306,
'dump_dir': '/tmp/mysql_restores'
},
'telegram': {
'enabled': True,
'bot_token': 'CHATBOT_TOKEN',
'chat_id': 'CHAT_ID',
'api_timeout': 10
}
}

logging.basicConfig(
filename='/var/log/file_level_restore.log',
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
filemode='a'
)

class EnhancedRestore:
def __init__(self):
self.start_time = datetime.now()
self.env = os.environ.copy()
self.env.update({
'RESTIC_PASSWORD': CONFIG['password'],
'RCLONE_CONFIG': CONFIG['rclone_config']
})
self.restored_items = []
self.setup_path()
self.ensure_restore_dir()

def setup_path(self):
essential_paths = ['/usr/bin', '/usr/local/bin', '/bin']
self.env['PATH'] = ":".join(essential_paths + self.env.get('PATH', '').split(':'))

def ensure_restore_dir(self):
restore_dir = Path(CONFIG['mysql']['dump_dir'])
restore_dir.mkdir(parents=True, exist_ok=True)

def run_cmd(self, cmd, context="Command"):
logging.info(f"Executing: {' '.join(cmd)}")
try:
result = subprocess.run(
cmd,
env=self.env,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=True,
check=True
)
logging.info(f"{context} succeeded: {result.stdout[:200]}...")
return result.stdout
except subprocess.CalledProcessError as e:
error_msg = f"{context} failed: {e.stderr}"
logging.error(error_msg)
raise Exception(error_msg)

def list_snapshots(self, tag_filter=None):
cmd = [
CONFIG['restic_bin'],
'snapshots',
'--repo', CONFIG['repository'],
'--json',
'--option', f"rclone.program={CONFIG['rclone_bin']}",
'--option', f"rclone.args=serve restic --stdio --config {CONFIG['rclone_config']}"
]
if tag_filter:
cmd.extend(['--tag', tag_filter])
output = self.run_cmd(cmd, "List Snapshots")
return json.loads(output)

def select_snapshot(self, snapshots):
print("\nAvailable Snapshots:")
for i, snapshot in enumerate(snapshots):
tags = ",".join(snapshot.get('tags', []))
time = snapshot.get('time', 'N/A')
print(f"{i}: ID={snapshot['short_id']}, Time={time}, Tags={tags}")
while True:
try:
choice = int(input("\nSelect snapshot number to restore (0-{}): ".format(len(snapshots)-1)))
if 0 <= choice < len(snapshots):
return snapshots[choice]['short_id']
print("Invalid choice. Try again.")
except ValueError:
print("Please enter a valid number.")

def restore_domain(self, user, domain):
domain_path = Path(f"/home/{user}/domains/{domain}")
domain_path.mkdir(parents=True, exist_ok=True)
temp_restore_path = Path(f"/tmp/restore_{user}_{domain}_{int(time.time())}")
temp_restore_path.mkdir(parents=True, exist_ok=True)

snapshots = self.list_snapshots(f"user:{user},domain:{domain}")
if not snapshots:
logging.error(f"No snapshots found for user:{user}, domain:{domain}")
print(f"No backups found for {domain}")
return False

snapshot_id = self.select_snapshot(snapshots)

cmd = [
CONFIG['restic_bin'],
'restore',
snapshot_id,
'--repo', CONFIG['repository'],
'--target', str(temp_restore_path),
'--option', f"rclone.program={CONFIG['rclone_bin']}",
'--option', f"rclone.args=serve restic --stdio --config {CONFIG['rclone_config']}"
]

try:
self.run_cmd(cmd, f"Restore domain {domain} for user {user} to temp dir")
restored_domain_path = temp_restore_path / "home" / user / "domains" / domain
if not restored_domain_path.exists():
raise Exception(f"Expected restored path {restored_domain_path} not found")

for item in restored_domain_path.iterdir():
dest_path = domain_path / item.name
if dest_path.exists():
if dest_path.is_symlink():
dest_path.unlink()
elif dest_path.is_dir():
shutil.rmtree(dest_path)
else:
dest_path.unlink()
shutil.move(str(item), str(domain_path))

self.set_permissions(domain_path, user)
self.restored_items.append(f"Domain: {domain} (User: {user})")
logging.info(f"Successfully restored domain {domain} for user {user}")
print(f"Domain {domain} restored successfully.")
return True

except Exception as e:
logging.error(f"Restore failed for domain {domain}: {str(e)}")
print(f"Failed to restore domain {domain}: {str(e)}")
return False

finally:
if temp_restore_path.exists():
shutil.rmtree(temp_restore_path, ignore_errors=True)

def restore_database(self, db_name):
snapshots = self.list_snapshots(f"mysql,database:{db_name}")
if not snapshots:
logging.error(f"No snapshots found for database:{db_name}")
print(f"No backups found for database {db_name}")
return False

snapshot_id = self.select_snapshot(snapshots)
temp_restore_path = Path(f"/tmp/restore_db_{db_name}_{int(time.time())}")
temp_restore_path.mkdir(parents=True, exist_ok=True)
dump_file = Path(CONFIG['mysql']['dump_dir']) / f"{db_name}_restore_{snapshot_id}.sql.gz"

cmd = [
CONFIG['restic_bin'],
'restore',
snapshot_id,
'--repo', CONFIG['repository'],
'--target', str(temp_restore_path),
'--option', f"rclone.program={CONFIG['rclone_bin']}",
'--option', f"rclone.args=serve restic --stdio --config {CONFIG['rclone_config']}"
]

try:
self.run_cmd(cmd, f"Restore dump for database {db_name}")

gz_files = list(temp_restore_path.rglob(f"{db_name}_*.sql.gz"))
if not gz_files:
raise Exception(f"No .sql.gz file found in restored snapshot for {db_name}")
if len(gz_files) > 1:
logging.warning(f"Multiple .sql.gz files found, using the first one: {gz_files[0]}")

restored_gz_file = gz_files[0]
shutil.move(str(restored_gz_file), str(dump_file)) # Di chuyển file đến vị trí mong muốn

with subprocess.Popen(['gunzip', str(dump_file)], stdout=subprocess.PIPE, stderr=subprocess.PIPE) as gunzip:
gunzip.wait()
if gunzip.returncode != 0:
raise Exception(f"Gunzip failed: {gunzip.stderr.read().decode()}")

sql_file = dump_file.with_suffix('')
self.import_database(db_name, sql_file)
sql_file.unlink() # Dọn dẹp file SQL
self.restored_items.append(f"Database: {db_name}") # Ghi nhận restore thành công
logging.info(f"Successfully restored database {db_name}")
print(f"Database {db_name} restored successfully.")
return True

except Exception as e:
logging.error(f"Restore failed for database {db_name}: {str(e)}")
print(f"Failed to restore database {db_name}: {str(e)}")
return False

finally:
if temp_restore_path.exists():
shutil.rmtree(temp_restore_path, ignore_errors=True)

def import_database(self, db_name, sql_file):
drop_cmd = [
'mysql',
f"--user={CONFIG['mysql']['user']}",
f"--password={CONFIG['mysql']['password']}",
f"--host={CONFIG['mysql']['host']}",
f"--port={CONFIG['mysql']['port']}",
'-e', f"DROP DATABASE IF EXISTS {db_name}; CREATE DATABASE {db_name};"
]
self.run_cmd(drop_cmd, f"Recreate database {db_name}")

import_cmd = [
'mysql',
f"--user={CONFIG['mysql']['user']}",
f"--password={CONFIG['mysql']['password']}",
f"--host={CONFIG['mysql']['host']}",
f"--port={CONFIG['mysql']['port']}",
db_name
]
with open(sql_file, 'r') as f:
result = subprocess.run(
import_cmd,
stdin=f,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env=self.env,
universal_newlines=True
)
if result.returncode != 0:
raise Exception(f"Database import failed: {result.stderr}")

def set_permissions(self, domain_path, user):
import pwd
import grp
try:
uid = pwd.getpwnam(user).pw_uid
gid = grp.getgrnam(user).gr_gid
os.chown(str(domain_path), uid, gid)
os.chmod(str(domain_path), 0o751)
for root, dirs, files in os.walk(str(domain_path)):
for d in dirs:
d_path = os.path.join(root, d)
os.chown(d_path, uid, gid)
os.chmod(d_path, 0o755)
for f in files:
f_path = os.path.join(root, f)
os.chown(f_path, uid, gid)
os.chmod(f_path, 0o644)
logging.info(f"Permissions set successfully for {domain_path}")
except Exception as e:
logging.error(f"Failed to set permissions for {domain_path}: {str(e)}")
raise

def send_telegram_report(self, success=True, error_msg=None):
if not CONFIG['telegram']['enabled']:
return
report = self.generate_report(success, error_msg)
url = f"https://api.telegram.org/bot{CONFIG['telegram']['bot_token']}/sendMessage"
payload = {'chat_id': CONFIG['telegram']['chat_id'], 'text': report, 'parse_mode': 'HTML'}
requests.post(url, data=payload, timeout=CONFIG['telegram']['api_timeout'])

def generate_report(self, success, error_msg=None):
status_emoji = "✅" if success else "❌"
status_text = "THÀNH CÔNG" if success else "THẤT BẠI"
duration = datetime.now() - self.start_time
duration_str = str(duration).split('.')[0]

report = f"""
<b>{status_emoji} RESTORE REPORT {status_emoji}</b>
━━━━━━━━━━━━━━━━━━━
<b>• Trạng thái:</b> {status_text}
<b>• Thời gian bắt đầu:</b> {self.start_time.strftime('%Y-%m-%d %H:%M:%S')}
<b>• Thời lượng:</b> {duration_str}
<b>• Repository:</b> {CONFIG['repository']}
"""
if self.restored_items: # Thêm danh sách các đối tượng restore thành công
report += "<b>• Restored successfully:</b>\n" + "\n".join([f" - {item}" for item in self.restored_items])
if error_msg:
report += f"\n<b>• Lỗi chính:</b> <code>{error_msg[:500]}</code>"
return report

def main():
restore = EnhancedRestore()
success = False
error_message = None

try:
print("Restore Options:")
print("1. Restore a domain")
print("2. Restore a database")
choice = input("Select an option (1-2): ")

if choice == '1':
user = input("Enter username: ")
domain = input("Enter domain name (e.g., example.com): ")
success = restore.restore_domain(user, domain)
elif choice == '2':
db_name = input("Enter database name: ")
success = restore.restore_database(db_name)
else:
raise Exception("Invalid option selected")

if not success:
raise Exception("Restore process failed")

except Exception as e:
error_message = str(e)
logging.critical(f"Restore failed: {error_message}")
success = False

finally:
restore.send_telegram_report(success, error_message)
restore_dir = Path(CONFIG['mysql']['dump_dir'])
for f in restore_dir.glob("*.sql.gz"):
f.unlink(missing_ok=True)
return 0 if success else 1

if __name__ == "__main__":
exit(main())