broccoliandClaude Fable 5
edfb216561
公开数据看板、四位 Key 录入面板、管理后台与三级权限、动态字段配置、 PostgreSQL/SQLite 双支持、数据库备份，以及本版新增的成品图上传与预览。 图片相关： - 录入表单支持相册选图与手机端直接拍照，每条记录最多 6 张 - 浏览器内压缩到最长边 1600 并剥离 EXIF，入库统一为 JPG/PNG - 二进制直接入库，现有备份自动覆盖图片，部署无需新增卷 - 详情页缩略图宫格与全屏 lightbox，公开看板同样可见 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
94 lines
3.6 KiB
Python
94 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import shutil
|
|
import sqlite3
|
|
import subprocess
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .config import settings
|
|
from .models import BackupRecord
|
|
|
|
|
|
def file_sha256(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def postgres_cli_url(url: str) -> str:
|
|
"""PostgreSQL CLI 不识别 SQLAlchemy 的 +psycopg 驱动标记。"""
|
|
return url.replace("postgresql+psycopg://", "postgresql://", 1)
|
|
|
|
|
|
def create_backup(db: Session, user_id: int | None = None) -> BackupRecord:
|
|
stamp = datetime.now().astimezone().strftime("%Y%m%d-%H%M%S")
|
|
url = settings.database_url
|
|
if url.startswith("sqlite"):
|
|
filename = f"database-{stamp}.sqlite3"
|
|
target = settings.backup_dir / filename
|
|
source_path = Path(url.split("///", 1)[1])
|
|
if not source_path.is_absolute():
|
|
source_path = Path.cwd() / source_path
|
|
source = sqlite3.connect(str(source_path))
|
|
destination = sqlite3.connect(str(target))
|
|
try:
|
|
source.backup(destination)
|
|
finally:
|
|
destination.close()
|
|
source.close()
|
|
database_type = "sqlite"
|
|
elif url.startswith(("postgresql", "postgres")):
|
|
filename = f"database-{stamp}.dump"
|
|
target = settings.backup_dir / filename
|
|
result = subprocess.run(["pg_dump", "--format=custom", "--file", str(target), postgres_cli_url(url)], capture_output=True, text=True)
|
|
if result.returncode:
|
|
raise RuntimeError(f"pg_dump 失败：{result.stderr.strip()}")
|
|
database_type = "postgresql"
|
|
else:
|
|
raise RuntimeError("当前数据库类型暂不支持自动备份")
|
|
record = BackupRecord(filename=filename, database_type=database_type, size_bytes=target.stat().st_size, sha256=file_sha256(target), created_by=user_id)
|
|
db.add(record)
|
|
db.commit()
|
|
db.refresh(record)
|
|
cleanup_old_backups(db)
|
|
return record
|
|
|
|
|
|
def cleanup_old_backups(db: Session) -> None:
|
|
cutoff = datetime.now().astimezone() - timedelta(days=settings.backup_keep_days)
|
|
for record in db.query(BackupRecord).filter(BackupRecord.created_at < cutoff).all():
|
|
path = settings.backup_dir / record.filename
|
|
if path.exists():
|
|
path.unlink()
|
|
db.delete(record)
|
|
db.commit()
|
|
|
|
|
|
def restore_backup(path: Path) -> None:
|
|
if not path.resolve().is_relative_to(settings.backup_dir.resolve()):
|
|
raise ValueError("备份文件必须位于 backups 目录")
|
|
url = settings.database_url
|
|
if url.startswith("sqlite"):
|
|
source = sqlite3.connect(str(path))
|
|
raw_path = Path(url.split("///", 1)[1])
|
|
target_path = raw_path if raw_path.is_absolute() else Path.cwd() / raw_path
|
|
temporary = target_path.with_suffix(".restore.tmp")
|
|
destination = sqlite3.connect(str(temporary))
|
|
try:
|
|
source.backup(destination)
|
|
finally:
|
|
destination.close()
|
|
source.close()
|
|
shutil.move(temporary, target_path)
|
|
elif url.startswith(("postgresql", "postgres")):
|
|
result = subprocess.run(["pg_restore", "--clean", "--if-exists", "--no-owner", "--dbname", postgres_cli_url(url), str(path)], capture_output=True, text=True)
|
|
if result.returncode:
|
|
raise RuntimeError(f"pg_restore 失败：{result.stderr.strip()}")
|
|
else:
|
|
raise RuntimeError("当前数据库类型暂不支持恢复")
|