hz4th_coder
df2f0860b5
- 新增备份管理页面（备份/恢复/下载/删除） - 新增 backup_service 服务层 - 文章编辑器增加工具栏图片上传按钮 - 优化图片上传交互提示
179 lines
6.2 KiB
Python
179 lines
6.2 KiB
Python
"""备份服务：一键备份网站数据和资源"""
|
|
import os
|
|
import shutil
|
|
import zipfile
|
|
from datetime import datetime
|
|
from flask import current_app
|
|
|
|
|
|
class BackupService:
|
|
"""备份服务类"""
|
|
|
|
# 备份目录（项目根目录下的 backups 文件夹）
|
|
BACKUP_DIR = 'backups'
|
|
|
|
@classmethod
|
|
def get_backup_dir(cls):
|
|
"""获取备份目录路径"""
|
|
# 相对于项目根目录
|
|
root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
|
|
backup_dir = os.path.join(root_dir, cls.BACKUP_DIR)
|
|
|
|
# 确保备份目录存在
|
|
if not os.path.exists(backup_dir):
|
|
os.makedirs(backup_dir)
|
|
|
|
return backup_dir
|
|
|
|
@classmethod
|
|
def create_backup(cls, include_uploads=True):
|
|
"""
|
|
创建完整备份
|
|
|
|
Args:
|
|
include_uploads: 是否包含上传的图片附件
|
|
|
|
Returns:
|
|
tuple: (backup_file_path, error_message)
|
|
"""
|
|
try:
|
|
# 获取项目根目录
|
|
root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
|
|
|
|
# 备份文件名：techblog_backup_YYYYMMDD_HHMMSS.zip
|
|
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
|
backup_filename = f'techblog_backup_{timestamp}.zip'
|
|
backup_filepath = os.path.join(cls.get_backup_dir(), backup_filename)
|
|
|
|
# 数据库文件路径
|
|
db_path = os.path.join(root_dir, 'data', 'techblog.db')
|
|
|
|
# 创建数据库备份文件（带时间戳）
|
|
db_backup_filename = f'techblog_backup_{timestamp}.db'
|
|
db_backup_path = os.path.join(root_dir, 'data', db_backup_filename)
|
|
|
|
# 复制数据库文件
|
|
if os.path.exists(db_path):
|
|
shutil.copy2(db_path, db_backup_path)
|
|
|
|
# 上传目录路径
|
|
uploads_path = os.path.join(root_dir, 'uploads')
|
|
|
|
# 创建 ZIP 备份文件
|
|
with zipfile.ZipFile(backup_filepath, 'w', zipfile.ZIP_DEFLATED) as zipf:
|
|
# 1. 备份数据库（备份文件）
|
|
if os.path.exists(db_backup_path):
|
|
zipf.write(db_backup_path, f'data/{db_backup_filename}')
|
|
# 删除临时备份文件
|
|
os.remove(db_backup_path)
|
|
|
|
# 2. 备份上传的图片和附件（保持目录结构）
|
|
if include_uploads and os.path.exists(uploads_path):
|
|
for root, dirs, files in os.walk(uploads_path):
|
|
for file in files:
|
|
file_path = os.path.join(root, file)
|
|
# 计算相对路径，保持目录结构
|
|
rel_path = os.path.relpath(file_path, uploads_path)
|
|
arcname = os.path.join('uploads', rel_path)
|
|
zipf.write(file_path, arcname)
|
|
|
|
# 3. 写入备份信息
|
|
info_content = f"""tech-blog 备份信息
|
|
========================
|
|
备份时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
|
|
数据库备份: {db_backup_filename}
|
|
上传文件: {'已包含' if include_uploads else '未包含'}
|
|
|
|
恢复说明:
|
|
1. 解压备份文件
|
|
2. 将 data/{db_backup_filename} 复制为 data/techblog.db
|
|
3. 将 uploads/ 目录复制到项目根目录
|
|
"""
|
|
zipf.writestr('backup_info.txt', info_content)
|
|
|
|
# 获取备份文件大小
|
|
file_size = os.path.getsize(backup_filepath)
|
|
|
|
return {
|
|
'success': True,
|
|
'filepath': backup_filepath,
|
|
'filename': backup_filename,
|
|
'size': file_size,
|
|
'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
|
|
'db_backup_name': db_backup_filename
|
|
}, None
|
|
|
|
except Exception as e:
|
|
# 清理临时文件
|
|
if 'db_backup_path' in locals() and os.path.exists(db_backup_path):
|
|
os.remove(db_backup_path)
|
|
return None, str(e)
|
|
|
|
@classmethod
|
|
def list_backups(cls):
|
|
"""
|
|
列出所有备份文件
|
|
|
|
Returns:
|
|
list: 备份文件列表，每个元素包含文件名、大小、时间
|
|
"""
|
|
backup_dir = cls.get_backup_dir()
|
|
backups = []
|
|
|
|
if not os.path.exists(backup_dir):
|
|
return backups
|
|
|
|
for filename in os.listdir(backup_dir):
|
|
if filename.endswith('.zip') and filename.startswith('techblog_backup_'):
|
|
filepath = os.path.join(backup_dir, filename)
|
|
stat = os.stat(filepath)
|
|
|
|
backups.append({
|
|
'filename': filename,
|
|
'filepath': filepath,
|
|
'size': stat.st_size,
|
|
'created_at': datetime.fromtimestamp(stat.st_mtime)
|
|
})
|
|
|
|
# 按时间倒序排列
|
|
backups.sort(key=lambda x: x['created_at'], reverse=True)
|
|
|
|
return backups
|
|
|
|
@classmethod
|
|
def delete_backup(cls, filename):
|
|
"""
|
|
删除备份文件
|
|
|
|
Args:
|
|
filename: 备份文件名
|
|
|
|
Returns:
|
|
tuple: (success, error_message)
|
|
"""
|
|
try:
|
|
backup_dir = cls.get_backup_dir()
|
|
filepath = os.path.join(backup_dir, filename)
|
|
|
|
if not os.path.exists(filepath):
|
|
return False, '备份文件不存在'
|
|
|
|
os.remove(filepath)
|
|
return True, None
|
|
|
|
except Exception as e:
|
|
return False, str(e)
|
|
|
|
@classmethod
|
|
def get_backup_path(cls, filename):
|
|
"""获取备份文件的完整路径"""
|
|
return os.path.join(cls.get_backup_dir(), filename)
|
|
|
|
@classmethod
|
|
def format_filesize(cls, size_bytes):
|
|
"""格式化文件大小"""
|
|
for unit in ['B', 'KB', 'MB', 'GB']:
|
|
if size_bytes < 1024.0:
|
|
return f"{size_bytes:.2f} {unit}"
|
|
size_bytes /= 1024.0
|
|
return f"{size_bytes:.2f} TB" |