Compare commits

..

1 Commits

Author SHA1 Message Date
7652718803 feat: 数据库备份机制 v2.4.0
- 新增备份管理 API 和页面
- 自动备份:每天 04:00 执行
- 手动备份:页面一键备份
- 备份清理规则:
  - 保留最近 30 天
  - 每月第一天永久保留
  - 手动备份最多保留 10 个
- 支持恢复和删除备份
2026-04-16 13:54:07 +08:00
3 changed files with 393 additions and 1 deletions

35
scripts/auto_backup.py Executable file
View File

@@ -0,0 +1,35 @@
#!/usr/bin/env python3
"""
Xian Favor 自动备份脚本
定时任务调用此脚本进行自动备份
"""
import sys
import os
# 添加项目路径
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from xian_favor.db import db
def main():
"""执行自动备份"""
print(f"[{datetime.now().isoformat()}] 开始自动备份...")
try:
backup_info = db.create_backup(manual=False)
print(f"备份成功: {backup_info['name']}")
print(f"大小: {backup_info['size']} bytes")
print(f"位置: {backup_info['path']}")
# 清理旧备份
db._cleanup_old_backups()
print("旧备份清理完成")
except Exception as e:
print(f"备份失败: {e}")
sys.exit(1)
if __name__ == '__main__':
from datetime import datetime
main()

View File

@@ -511,6 +511,44 @@ def send_email():
return jsonify({'success': False, 'error': f'发送失败: {str(e)}'}), 500
# ============ 备份管理 API ============
@app.route('/api/backups', methods=['GET'])
def list_backups():
"""列出所有备份"""
backups = db.list_backups()
return jsonify({'success': True, 'data': backups})
@app.route('/api/backups', methods=['POST'])
def create_backup():
"""创建手动备份"""
try:
backup_info = db.create_backup(manual=True)
return jsonify({'success': True, 'data': backup_info}), 201
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/backups/<backup_name>', methods=['POST'])
def restore_backup(backup_name):
"""恢复备份"""
try:
if db.restore_backup(backup_name):
return jsonify({'success': True, 'message': f'已恢复备份 {backup_name}'})
return jsonify({'success': False, 'error': '备份不存在'}), 404
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/backups/<backup_name>', methods=['DELETE'])
def delete_backup(backup_name):
"""删除备份"""
if db.delete_backup(backup_name):
return jsonify({'success': True})
return jsonify({'success': False, 'error': '备份不存在'}), 404
# ============ Web 页面 ============
@app.route('/')
@@ -579,6 +617,7 @@ INDEX_TEMPLATE = '''
<hr class="border-secondary">
<a href="#" onclick="showTagManager(); return false;"><i class="bi bi-tags"></i> 标签管理</a>
<a href="#" onclick="showEmailManager(); return false;"><i class="bi bi-envelope"></i> 邮箱管理</a>
<a href="#" onclick="showBackupManager(); return false;"><i class="bi bi-archive"></i> 备份管理</a>
</nav>
</div>
@@ -954,6 +993,41 @@ INDEX_TEMPLATE = '''
</div>
</div>
<!-- 备份管理模态框 -->
<div class="modal fade" id="backupModal" tabindex="-1">
<div class="modal-dialog modal-lg">
<div class="modal-content">
<div class="modal-header bg-info text-white">
<h5 class="modal-title"><i class="bi bi-archive"></i> 备份管理</h5>
<button type="button" class="btn-close btn-close-white" data-bs-dismiss="modal"></button>
</div>
<div class="modal-body">
<div class="mb-3">
<button class="btn btn-primary" onclick="createManualBackup()">
<i class="bi bi-plus-circle"></i> 立即备份
</button>
<span class="text-muted ms-3">自动备份:每天 04:00</span>
</div>
<div class="alert alert-info">
<strong>备份规则:</strong>
<ul class="mb-0">
<li>自动备份每天 04:00 执行</li>
<li>保留最近 30 天的备份</li>
<li>每月第一天的备份永久保留</li>
<li>手动备份最多保留 10 个</li>
</ul>
</div>
<div id="backupList">
<!-- 动态填充 -->
</div>
</div>
<div class="modal-footer">
<button type="button" class="btn btn-secondary" data-bs-dismiss="modal">关闭</button>
</div>
</div>
</div>
</div>
<!-- AI自动添加模态框 -->
<div class="modal fade" id="aiAddModal" tabindex="-1">
<div class="modal-dialog modal-lg">
@@ -2030,6 +2104,106 @@ async function showEmailManager() {
new bootstrap.Modal(document.getElementById('emailManagerModal')).show();
}
// ============ 备份管理 ============
async function showBackupManager() {
await loadBackupList();
new bootstrap.Modal(document.getElementById('backupModal')).show();
}
async function loadBackupList() {
const res = await fetch(`${API_BASE}/backups`);
const data = await res.json();
if (!data.success) return;
const container = document.getElementById('backupList');
if (!data.data.length) {
container.innerHTML = '<div class="text-center text-muted py-3">暂无备份</div>';
return;
}
container.innerHTML = data.data.map(backup => `
<div class="d-flex justify-content-between align-items-center p-2 border-bottom rounded mb-1 ${backup.manual ? 'bg-light' : ''}">
<div>
<strong>${backup.name}</strong>
${backup.manual ? '<span class="badge bg-primary ms-1">手动</span>' : ''}
${backup.is_first_of_month ? '<span class="badge bg-success ms-1">月首</span>' : ''}
<br>
<small class="text-muted">
${formatBackupDate(backup.created_at)} |
${formatBackupSize(backup.size)}
</small>
</div>
<div class="btn-group btn-group-sm">
<button class="btn btn-outline-success" onclick="restoreBackup('${backup.name}')" title="恢复">
<i class="bi bi-arrow-counterclockwise"></i> 恢复
</button>
<button class="btn btn-outline-danger" onclick="deleteBackupConfirm('${backup.name}')" title="删除">
<i class="bi bi-trash"></i>
</button>
</div>
</div>
`).join('');
}
async function createManualBackup() {
const res = await fetch(`${API_BASE}/backups`, { method: 'POST' });
const data = await res.json();
if (data.success) {
alert('备份创建成功!');
loadBackupList();
loadStats();
} else {
alert('备份失败: ' + data.error);
}
}
async function restoreBackup(name) {
if (!confirm(`确认恢复备份 "${name}"\n当前数据将被覆盖!`)) return;
const res = await fetch(`${API_BASE}/backups/${name}`, { method: 'POST' });
const data = await res.json();
if (data.success) {
alert('备份已恢复!页面将刷新...');
location.reload();
} else {
alert('恢复失败: ' + data.error);
}
}
async function deleteBackupConfirm(name) {
if (!confirm(`确认删除备份 "${name}"`)) return;
const res = await fetch(`${API_BASE}/backups/${name}`, { method: 'DELETE' });
const data = await res.json();
if (data.success) {
loadBackupList();
} else {
alert('删除失败: ' + data.error);
}
}
function formatBackupDate(dateStr) {
const date = new Date(dateStr);
return date.toLocaleString('zh-CN', {
year: 'numeric',
month: '2-digit',
day: '2-digit',
hour: '2-digit',
minute: '2-digit'
});
}
function formatBackupSize(bytes) {
if (bytes < 1024) return bytes + ' B';
if (bytes < 1024 * 1024) return (bytes / 1024).toFixed(1) + ' KB';
return (bytes / 1024 / 1024).toFixed(2) + ' MB';
}
async function loadEmailManagerList() {
const res = await fetch(`${API_BASE}/emails`);
const data = await res.json();

View File

@@ -2,12 +2,17 @@
import sqlite3
import json
from datetime import datetime
import shutil
import os
from datetime import datetime, timedelta
from typing import Optional, List, Dict, Any
from contextlib import contextmanager
from .config import DATABASE_URL, TODO_STATUS, PRIORITY_LEVELS
# 备份目录
BACKUP_DIR = os.path.join(os.path.dirname(DATABASE_URL), 'backups')
class Database:
"""SQLite数据库管理"""
@@ -501,6 +506,184 @@ class Database:
reminders['total'] = len(reminders['overdue']) + len(reminders['due_today']) + len(reminders['due_soon'])
return reminders
# ============ 备份操作 ============
def create_backup(self, manual: bool = False) -> Dict[str, Any]:
"""创建数据库备份"""
import os
# 确保备份目录存在
os.makedirs(BACKUP_DIR, exist_ok=True)
now = datetime.now()
backup_name = now.strftime('%Y-%m-%d_%H%M%S')
if manual:
backup_name += '_manual'
backup_path = os.path.join(BACKUP_DIR, f'{backup_name}.db')
# 复制数据库文件
shutil.copy2(self.db_path, backup_path)
# 获取备份信息
backup_info = {
'name': backup_name,
'path': backup_path,
'size': os.path.getsize(backup_path),
'created_at': now.isoformat(),
'manual': manual,
'is_first_of_month': now.day == 1
}
# 保存备份元数据
self._save_backup_meta(backup_info)
# 清理旧备份
self._cleanup_old_backups()
return backup_info
def list_backups(self) -> List[Dict[str, Any]]:
"""列出所有备份"""
import os
if not os.path.exists(BACKUP_DIR):
return []
# 读取备份元数据
meta_path = os.path.join(BACKUP_DIR, 'backup_meta.json')
if os.path.exists(meta_path):
with open(meta_path, 'r') as f:
backups = json.load(f)
else:
# 从文件重建元数据
backups = []
for f in os.listdir(BACKUP_DIR):
if f.endswith('.db'):
path = os.path.join(BACKUP_DIR, f)
backups.append({
'name': f.replace('.db', ''),
'path': path,
'size': os.path.getsize(path),
'created_at': datetime.fromtimestamp(os.path.getmtime(path)).isoformat(),
'manual': '_manual' in f,
'is_first_of_month': self._is_first_of_month_filename(f)
})
# 按时间倒序排列
backups.sort(key=lambda x: x['created_at'], reverse=True)
return backups
def restore_backup(self, backup_name: str) -> bool:
"""恢复备份"""
import os
backup_path = os.path.join(BACKUP_DIR, f'{backup_name}.db')
if not os.path.exists(backup_path):
return False
# 先备份当前数据库(以防万一)
current_backup = self.db_path + '.before_restore'
shutil.copy2(self.db_path, current_backup)
# 恢复备份
shutil.copy2(backup_path, self.db_path)
return True
def delete_backup(self, backup_name: str) -> bool:
"""删除备份"""
import os
backup_path = os.path.join(BACKUP_DIR, f'{backup_name}.db')
if not os.path.exists(backup_path):
return False
os.remove(backup_path)
# 更新元数据
self._remove_backup_meta(backup_name)
return True
def _save_backup_meta(self, backup_info: Dict[str, Any]):
"""保存备份元数据"""
import os
meta_path = os.path.join(BACKUP_DIR, 'backup_meta.json')
# 读取现有元数据
backups = []
if os.path.exists(meta_path):
with open(meta_path, 'r') as f:
backups = json.load(f)
# 添加新备份
backups.append(backup_info)
# 保存
with open(meta_path, 'w') as f:
json.dump(backups, f, indent=2)
def _remove_backup_meta(self, backup_name: str):
"""从元数据中删除备份"""
import os
meta_path = os.path.join(BACKUP_DIR, 'backup_meta.json')
if not os.path.exists(meta_path):
return
with open(meta_path, 'r') as f:
backups = json.load(f)
backups = [b for b in backups if b['name'] != backup_name]
with open(meta_path, 'w') as f:
json.dump(backups, f, indent=2)
def _is_first_of_month_filename(self, filename: str) -> bool:
"""判断是否是每月第一天的备份"""
# 格式2026-04-01_040000.db 或 2026-05-01_...
try:
date_part = filename.split('_')[0]
day = int(date_part.split('-')[2])
return day == 1
except:
return False
def _cleanup_old_backups(self):
"""清理旧备份保留30天 + 每月第一天"""
import os
backups = self.list_backups()
now = datetime.now()
keep_paths = []
for backup in backups:
backup_date = datetime.fromisoformat(backup['created_at'])
days_old = (now - backup_date).days
# 保留条件:
# 1. 手动备份永久保留最多10个
# 2. 30天内的备份
# 3. 每月第一天的备份
if backup['manual']:
# 手动备份保留但最多10个
manual_backups = [b for b in backups if b['manual']]
if manual_backups.index(backup) < 10:
keep_paths.append(backup['path'])
else:
self.delete_backup(backup['name'])
elif days_old <= 30:
keep_paths.append(backup['path'])
elif backup['is_first_of_month']:
# 每月第一天永久保留
keep_paths.append(backup['path'])
else:
# 删除
self.delete_backup(backup['name'])
# 全局数据库实例