
halo自动备份到alist-s3脚本
Halo 备份和 S3 上传教程 v2.0
更新日志 v2.0
新增功能
- 分段上传支持:添加了分段上传开关,可处理大文件上传
- 进度显示功能:实时显示上传进度百分比和传输量
- 彩色终端输出:不同状态信息使用不同颜色区分
- 增强的错误处理:更详细的错误提示和排查建议
- 兼容性优化:更好的支持各类 S3 兼容服务
优化改进
- 简化配置流程
- 增加日志记录功能
- 改进上传稳定性
- 添加详细的注释说明
环境要求
- Python 3.6+
boto3
库:用于与 S3 兼容服务交互requests
库:用于发送 HTTP 请求
前提要求
如果你还没有安装alist或者任何s3服务的,你可以参考我的这篇文章
(Alist/S3/阿里ossz制作随机图片api | 枫の屋 (6wd.cn))
注:如果你完成了alist的s3配置或者任何s3服务,您可以继续往下看了
最重要的,如果你的halo版本>=2.20请务必完成这一步操作!!
如果你是docker run
启动,那么需要你在命令中间加一句
-e HALO_SECURITY_BASIC_AUTH_DISABLED=false
例:
docker run -d --name halo -p 8090:8090 -v ~/.halo2:/root/.halo2 -e HALO_SECURITY_BASIC_AUTH_DISABLED=false halohub/halo:2.20
2,如果你是docker-compose
部署的,那么你需要在docker-compose.yaml中添加
environment:
- HALO_SECURITY_BASIC_AUTH_DISABLED=false
例:
version: "3"
services:
halo:
image: registry.fit2cloud.com/halo/halo:2.20
restart: on-failure:3
network_mode: "host"
volumes:
- ./halo2:/root/.halo2
command:
# 修改为自己已有的 MySQL 配置
- --spring.r2dbc.url=r2dbc:pool:mysql://localhost:3306/halo
- --spring.r2dbc.username=root
- --spring.r2dbc.password=
- --spring.sql.init.platform=mysql
# 外部访问地址,请根据实际需要修改
- --halo.external-url=http://localhost:8090/
# 端口号 默认8090
- --server.port=8090
# 额外启动脚本
environment:
- HALO_SECURITY_BASIC_AUTH_DISABLED=false
步骤一:安装依赖
使用以下命令安装所需的库:
pip install boto3 requests
步骤二:创建 Python 脚本
创建一个新的 Python 脚本(例如 backup_script_v2.py
),并将以下代码粘贴到文件中:
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os
import base64
import time
import requests
import json
import boto3
from botocore.exceptions import NoCredentialsError, PartialCredentialsError, ClientError
from datetime import datetime, timedelta
import logging
# ===================== 配置区域 ===================== #
# Halo 配置
HALO_USER = "admin" # Halo 用户名
HALO_PASSWORD = "your_password" # Halo 密码
HALO_WEBSITE = "https://yourdomain.com" # Halo 站点地址
HALO_BACKUP_PATH = "/path/to/backups" # 备份文件存储路径
# S3 兼容存储配置
S3_ENDPOINT = "http://your-s3-endpoint:port" # S3 服务地址
S3_ACCESS_KEY = "your_access_key" # Access Key
S3_SECRET_KEY = "your_secret_key" # Secret Key
S3_BUCKET = "your-bucket-name" # 存储桶名称
S3_REGION = "auto" # 区域(如无特殊保持 auto)
# 功能开关
ENABLE_MULTIPART_UPLOAD = True # 是否启用分段上传(大文件建议开启)
MULTIPART_THRESHOLD = 100 * 1024 * 1024 # 100MB 以上使用分段上传
CHUNK_SIZE = 50 * 1024 * 1024 # 每个分块50MB
# 备份有效期(天)
BACKUP_EXPIRY_DAYS = 3
# 日志配置
LOG_FILE = "/path/to/backup_log.txt" # 日志文件路径
# ================================================== #
class Color:
"""终端颜色输出"""
GREEN = '\033[92m'
YELLOW = '\033[93m'
RED = '\033[91m'
BLUE = '\033[94m'
BOLD = '\033[1m'
END = '\033[0m'
def setup_logging():
"""配置日志记录"""
logging.basicConfig(
filename=LOG_FILE,
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
console = logging.StreamHandler()
console.setLevel(logging.INFO)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
console.setFormatter(formatter)
logging.getLogger('').addHandler(console)
def print_header(text):
"""打印标题"""
print(f"\n{Color.BOLD}{Color.BLUE}=== {text} ==={Color.END}")
logging.info(f"=== {text} ===")
def print_success(text):
"""打印成功信息"""
print(f"{Color.GREEN}✓ {text}{Color.END}")
logging.info(f"SUCCESS: {text}")
def print_warning(text):
"""打印警告信息"""
print(f"{Color.YELLOW}⚠ {text}{Color.END}")
logging.warning(f"WARNING: {text}")
def print_error(text):
"""打印错误信息"""
print(f"{Color.RED}✗ {text}{Color.END}")
logging.error(f"ERROR: {text}")
def print_progress(current, total):
"""打印进度条"""
bar_length = 30
progress = current / total
block = int(round(bar_length * progress))
progress_percent = round(progress * 100, 2)
text = f"\r[{'█' * block}{' ' * (bar_length - block)}] {progress_percent}% ({current/1024/1024:.1f}/{total/1024/1024:.1f}MB)"
print(text, end='', flush=True)
def create_halo_backup():
"""创建 Halo 备份并返回备份文件名"""
print_header("触发 Halo 备份")
# 计算过期时间
expires_at = (datetime.utcnow() + timedelta(days=BACKUP_EXPIRY_DAYS)).strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z'
# API 配置
backup_api = f"{HALO_WEBSITE}/apis/migration.halo.run/v1alpha1/backups"
check_api = f"{HALO_WEBSITE}/apis/migration.halo.run/v1alpha1/backups?sort=metadata.creationTimestamp%2Cdesc"
auth_header = "Basic " + base64.b64encode(f"{HALO_USER}:{HALO_PASSWORD}".encode()).decode()
# 请求负载
payload = json.dumps({
"apiVersion": "migration.halo.run/v1alpha1",
"kind": "Backup",
"metadata": {"generateName": "auto-backup-"},
"spec": {"expiresAt": expires_at}
})
headers = {
'Content-Type': 'application/json',
'Authorization': auth_header,
}
# 发送备份请求
try:
response = requests.post(backup_api, headers=headers, data=payload, timeout=30)
response.raise_for_status()
print_success("备份任务已创建")
except requests.RequestException as e:
print_error(f"备份请求失败: {str(e)}")
return None
# 等待备份完成
print("⏳ 等待备份完成...", end="", flush=True)
backup_name = None
for _ in range(30): # 最多等待 5 分钟(30*10秒)
try:
check_response = requests.get(check_api, headers=headers, timeout=10)
check_data = check_response.json()
if check_data.get("items"):
latest_backup = check_data["items"][0]
if latest_backup["status"]["phase"] == "SUCCEEDED":
backup_name = latest_backup["metadata"]["name"]
print("\n", end="")
print_success("备份成功完成!")
break
elif latest_backup["status"]["phase"] == "FAILED":
print("\n", end="")
print_error("备份失败!")
return None
print(".", end="", flush=True)
time.sleep(10)
except Exception as e:
print("\n", end="")
print_error(f"检查备份状态时出错: {str(e)}")
return None
if not backup_name:
print("\n", end="")
print_error("备份超时!")
return None
# 查找备份文件
backup_files = [f for f in os.listdir(HALO_BACKUP_PATH)
if f.endswith('.zip') and backup_name in f]
if not backup_files:
print_error(f"未找到匹配的备份文件(查找名称包含: {backup_name})")
return None
backup_files.sort(reverse=True)
return os.path.join(HALO_BACKUP_PATH, backup_files[0])
def upload_to_s3(file_path):
"""上传文件到 S3 兼容存储"""
print_header(f"上传 {os.path.basename(file_path)} 到 S3")
try:
# 初始化 S3 客户端
s3 = boto3.client(
's3',
endpoint_url=S3_ENDPOINT,
aws_access_key_id=S3_ACCESS_KEY,
aws_secret_access_key=S3_SECRET_KEY,
region_name=S3_REGION,
config=boto3.session.Config(
s3={'addressing_style': 'path'}, # 兼容更多 S3 服务
signature_version='s3v4'
)
)
# 检查存储桶是否存在
try:
s3.head_bucket(Bucket=S3_BUCKET)
print_success(f"存储桶 {S3_BUCKET} 访问正常")
except ClientError as e:
error_code = e.response['Error']['Code']
if error_code == '404':
print_error(f"存储桶 {S3_BUCKET} 不存在!")
elif error_code == '403':
print_error(f"无权限访问存储桶 {S3_BUCKET}!")
else:
print_error(f"存储桶检查失败: {str(e)}")
return False
file_size = os.path.getsize(file_path)
human_size = f"{file_size/1024/1024:.2f}MB"
# 根据配置选择上传方式
if ENABLE_MULTIPART_UPLOAD and file_size > MULTIPART_THRESHOLD:
print_warning(f"大文件检测 ({human_size}),启用分段上传...")
print_warning(f"分块大小: {CHUNK_SIZE/1024/1024}MB")
transfer_config = boto3.s3.transfer.TransferConfig(
multipart_threshold=MULTIPART_THRESHOLD,
multipart_chunksize=CHUNK_SIZE,
max_concurrency=5
)
# 使用更可靠的上传方式
s3.upload_file(
file_path,
S3_BUCKET,
os.path.basename(file_path),
Config=transfer_config
)
else:
print_success(f"文件大小: {human_size},使用简单上传")
# 手动实现进度显示
with open(file_path, 'rb') as f:
uploaded = 0
while True:
chunk = f.read(1024 * 1024) # 每次读取1MB
if not chunk:
break
s3.put_object(
Bucket=S3_BUCKET,
Key=os.path.basename(file_path),
Body=chunk
)
uploaded += len(chunk)
print_progress(uploaded, file_size)
print("\n", end="") # 结束进度条
print_success(f"上传成功!文件位置: s3://{S3_BUCKET}/{os.path.basename(file_path)}")
return True
except Exception as e:
print("\n", end="")
print_error(f"上传失败: {str(e)}")
print_warning("建议检查:")
print_warning("1. S3 服务是否正常运行")
print_warning("2. 访问密钥是否有写入权限")
print_warning("3. 网络连接是否正常")
return False
def main():
setup_logging()
print(f"\n{Color.BOLD}{Color.BLUE}=== Halo 博客自动备份工具 v2.0 ===")
print(f"📅 备份时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print(f"📂 本地路径: {HALO_BACKUP_PATH}")
print(f"☁️ 云存储: {S3_ENDPOINT}/{S3_BUCKET}")
print(f"⚙️ 分段上传: {'启用' if ENABLE_MULTIPART_UPLOAD else '禁用'}")
print("="*40 + f"{Color.END}")
logging.info("=== 备份任务开始 ===")
# 步骤1:创建备份
backup_file = create_halo_backup()
if not backup_file:
print_error("备份流程中止")
logging.error("备份流程中止")
return
# 步骤2:上传到 S3
if not upload_to_s3(backup_file):
print_error("上传流程中止")
logging.error("上传流程中止")
return
print_success("\n🎉 所有操作已完成!备份已安全存储")
logging.info("所有操作已完成!备份已安全存储")
if __name__ == "__main__":
main()
步骤三:配置脚本
在脚本的配置区域,您需要修改以下参数:
# Halo 配置
HALO_USER = "admin" # Halo 用户名
HALO_PASSWORD = "your_password" # Halo 密码
HALO_WEBSITE = "https://yourdomain.com" # Halo 站点地址
HALO_BACKUP_PATH = "/path/to/backups" # 备份文件存储路径
# S3 兼容存储配置
S3_ENDPOINT = "http://your-s3-endpoint:port" # S3 服务地址
S3_ACCESS_KEY = "your_access_key" # Access Key
S3_SECRET_KEY = "your_secret_key" # Secret Key
S3_BUCKET = "your-bucket-name" # 存储桶名称
# 日志配置
LOG_FILE = "/path/to/backup_log.txt" # 日志文件路径
步骤四:设置功能开关
# 功能开关
ENABLE_MULTIPART_UPLOAD = True # 是否启用分段上传(大文件建议开启)
MULTIPART_THRESHOLD = 100 * 1024 * 1024 # 100MB 以上使用分段上传
CHUNK_SIZE = 50 * 1024 * 1024 # 每个分块50MB
# 备份有效期(天)
BACKUP_EXPIRY_DAYS = 3
步骤五:运行脚本
在终端中运行以下命令:
python3 /path/to/your/backup_script_v2.py
步骤六:查看日志
所有输出信息将显示在终端中,并同时记录到指定的日志文件中。确保检查 S3 存储桶,以确认文件是否上传成功。
步骤七:设置自动化
使用宝塔的计划任务
设置自动备份:
- 点击右上角的"添加任务"按钮
- 在"任务类型"中选择"Shell脚本"
- 在"任务内容"中输入以下命令:
python3 /path/to/your/backup_script_v2.py
- 设置执行周期(建议每天执行一次)
- 点击"提交"保存
常见问题解答
Q: 上传大文件时失败怎么办?
A: 尝试以下步骤:
- 确保
ENABLE_MULTIPART_UPLOAD = True
- 减小
CHUNK_SIZE
值(如改为20MB) - 检查网络连接稳定性
Q: 如何查看详细的错误信息?
A: 检查日志文件 /path/to/backup_log.txt
,其中记录了所有操作细节和错误信息。
Q: 备份文件没有出现在S3存储桶中?
A: 请检查:
- S3存储桶名称是否正确
- 访问密钥是否有写入权限
- 网络是否能正常访问S3端点
Q: 脚本执行时间过长怎么办?
A: 可以调整以下参数:
- 减小
MULTIPART_THRESHOLD
值 - 增加
max_concurrency
值(但不要超过5)
如需进一步帮助,请提供日志文件内容以便诊断问题。
- 感谢你赐予我前进的力量
赞赏者名单
因为你们的支持让我意识到写文章的价值🙏
本文是原创文章,完整转载请注明来自 枫の屋
评论
匿名评论
隐私政策
你无需删除空行,直接评论以获取最佳展示效果