Files
cms-SiteForge/scripts/restore_backup.py
T
浅念 3c57c167f2 chore: import SiteForge Factory V5 source (baseline)
导入 SiteForge Factory V5 基线源码:
- FastAPI + SQLite + Docker Compose(Web + Worker)架构
- 四套商业主题 × 14 页面模板(1.1.0,含历史版本快照)
- 四种建站模式骨架 + 迁移审计 + 发布门禁
- 52 项 pytest 测试
- 补全计划主文档 docs/superpowers/plans/

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-10-22 10:42:19 +08:00

57 lines
2.7 KiB
Python

"""Offline restore of SiteForge backups. STOP API and worker containers first!
Usage: python scripts/restore_backup.py /path/to/backup.zip --confirmed-offline
The application must be stopped to prevent WAL writes during restore.
"""
import argparse
import os
import shutil
import sqlite3
import stat
import tempfile
import zipfile
from pathlib import Path
def restore(path, target, confirmed=False):
if not confirmed:raise RuntimeError('停止 API 与 worker 后,使用 --confirmed-offline 才能恢复')
target=Path(target).resolve();archive=Path(path).resolve()
if not archive.exists():raise FileNotFoundError(archive)
with zipfile.ZipFile(archive) as z:
infos=z.infolist()
if len(infos)>250000:raise ValueError('备份文件数量过大')
if sum(i.file_size for i in infos)>30_000_000_000:raise ValueError('解压数据总量过大')
for item in infos:
p=Path(item.filename)
if p.is_absolute() or any(x in ('..','.') for x in p.parts) or not p.parts:
raise ValueError('备份包含不安全路径')
if stat.S_ISLNK(item.external_attr>>16):raise ValueError('不允许符号链接')
if 'siteforge.sqlite3' not in z.namelist():raise ValueError('缺少数据库文件')
with tempfile.TemporaryDirectory(dir=target.parent) as tmp:
z.extractall(tmp)
restored=Path(tmp)/'siteforge.sqlite3'
with sqlite3.connect(restored) as db:
if db.execute('PRAGMA integrity_check').fetchone()[0]!='ok':
raise ValueError('备份数据库完整性校验失败')
target.mkdir(parents=True,exist_ok=True)
# Keep a copy of the old DB outside the restored data volume.
old_db=target/'siteforge.sqlite3'
if old_db.exists():
dest=target.parent/'siteforge-pre-restore.sqlite3'
shutil.copy2(old_db,dest)
for special in ('siteforge.sqlite3-wal','siteforge.sqlite3-shm'):
(target/special).unlink(missing_ok=True)
for src in Path(tmp).rglob('*'):
if src.is_file():
dest=target/src.relative_to(tmp)
dest.parent.mkdir(parents=True,exist_ok=True)
shutil.copy2(src,dest)
print('恢复完成。请检查文件权限后再启动 SiteForge。')
if __name__=='__main__':
parser=argparse.ArgumentParser()
parser.add_argument('archive')
parser.add_argument('--data',default=os.getenv('SITEFORGE_DATA_DIR','/data'))
parser.add_argument('--confirmed-offline',action='store_true')
args=parser.parse_args()
restore(args.archive,args.data,args.confirmed_offline)