dekun
245 lines
7.6 KiB
Python
245 lines
7.6 KiB
Python
"""系统设置：管理员账号、数据备份与恢复。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import time
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from fastapi.responses import FileResponse
|
|
from pydantic import BaseModel, Field
|
|
|
|
from apps.api.auth import auth_disabled, require_auth, verify_credentials
|
|
from packages.config import get_settings, reload_settings
|
|
from packages.config.env_file import update_env_file
|
|
from packages.db.backup import (
|
|
DISPLAY_LIMIT,
|
|
backup_db,
|
|
ensure_backup_dir,
|
|
latest_backup_mtime_ms,
|
|
list_backups,
|
|
prune_backups,
|
|
resolve_backup_file,
|
|
restore_db,
|
|
)
|
|
|
|
router = APIRouter(
|
|
prefix="/settings",
|
|
tags=["settings"],
|
|
dependencies=[Depends(require_auth)],
|
|
)
|
|
|
|
_USERNAME_RE = re.compile(r"^[a-zA-Z0-9_\-]{2,32}$")
|
|
|
|
|
|
class UpdateAccountBody(BaseModel):
|
|
current_password: str = Field(min_length=1, max_length=256)
|
|
new_username: str | None = Field(default=None, max_length=32)
|
|
new_password: str | None = Field(default=None, max_length=256)
|
|
|
|
|
|
class BackupActionBody(BaseModel):
|
|
current_password: str = Field(min_length=1, max_length=256)
|
|
|
|
|
|
class RestoreBody(BaseModel):
|
|
current_password: str = Field(min_length=1, max_length=256)
|
|
backup_name: str = Field(min_length=1, max_length=128)
|
|
|
|
|
|
class AutoBackupBody(BaseModel):
|
|
current_password: str = Field(min_length=1, max_length=256)
|
|
enabled: bool
|
|
interval_hours: float = Field(default=24, ge=1, le=168)
|
|
|
|
|
|
def _verify_password_or_401(current_password: str) -> None:
|
|
s = get_settings()
|
|
if auth_disabled():
|
|
raise HTTPException(status_code=400, detail="auth disabled")
|
|
if not verify_credentials(s.admin_username, current_password):
|
|
raise HTTPException(status_code=401, detail="current password incorrect")
|
|
|
|
|
|
def _backup_meta(s) -> dict:
|
|
backup_dir = s.backup_dir_path
|
|
latest_ms = latest_backup_mtime_ms(backup_dir)
|
|
next_due_ms = None
|
|
if s.backup_auto_enabled and s.backup_interval_hours > 0:
|
|
if latest_ms is None:
|
|
next_due_ms = int(time.time() * 1000)
|
|
else:
|
|
next_due_ms = latest_ms + int(s.backup_interval_hours * 3600 * 1000)
|
|
return {
|
|
"auto_enabled": s.backup_auto_enabled,
|
|
"interval_hours": s.backup_interval_hours,
|
|
"keep_count": s.backup_keep_count,
|
|
"display_limit": DISPLAY_LIMIT,
|
|
"latest_backup_ms": latest_ms,
|
|
"next_due_ms": next_due_ms,
|
|
}
|
|
|
|
|
|
@router.get("/backup")
|
|
def get_backup_info() -> dict:
|
|
s = get_settings()
|
|
backup_dir = s.backup_dir_path
|
|
writable = False
|
|
try:
|
|
ensure_backup_dir(backup_dir)
|
|
probe = backup_dir / ".write_probe"
|
|
probe.write_text("", encoding="utf-8")
|
|
probe.unlink(missing_ok=True)
|
|
writable = True
|
|
items = list_backups(backup_dir, limit=DISPLAY_LIMIT)
|
|
except OSError as e:
|
|
items = list_backups(backup_dir, limit=DISPLAY_LIMIT) if backup_dir.is_dir() else []
|
|
if not items:
|
|
raise HTTPException(status_code=500, detail=f"backup dir error: {e}") from e
|
|
return {
|
|
"backup_dir": str(backup_dir),
|
|
"db_path": str(s.db_path),
|
|
"writable": writable,
|
|
"backups": items,
|
|
**_backup_meta(s),
|
|
}
|
|
|
|
|
|
@router.put("/backup/auto")
|
|
def update_auto_backup(body: AutoBackupBody) -> dict:
|
|
_verify_password_or_401(body.current_password)
|
|
s = get_settings()
|
|
updates = {
|
|
"BACKUP_AUTO_ENABLED": "1" if body.enabled else "0",
|
|
"BACKUP_INTERVAL_HOURS": str(
|
|
int(body.interval_hours)
|
|
if float(body.interval_hours).is_integer()
|
|
else body.interval_hours
|
|
),
|
|
}
|
|
try:
|
|
update_env_file(s.env_file_path, updates)
|
|
except OSError as e:
|
|
raise HTTPException(status_code=500, detail=f"failed to write {s.env_file_path}: {e}") from e
|
|
reload_settings()
|
|
info = get_backup_info()
|
|
return {
|
|
"ok": True,
|
|
"message": "自动备份设置已保存",
|
|
**info,
|
|
}
|
|
|
|
|
|
@router.post("/backup")
|
|
def create_backup(body: BackupActionBody) -> dict:
|
|
_verify_password_or_401(body.current_password)
|
|
s = get_settings()
|
|
try:
|
|
dest = backup_db(s.db_path, s.backup_dir_path, tz=s.tz)
|
|
prune_backups(s.backup_dir_path, s.backup_keep_count)
|
|
except FileNotFoundError as e:
|
|
raise HTTPException(status_code=404, detail=str(e)) from e
|
|
except OSError as e:
|
|
raise HTTPException(status_code=500, detail=f"backup failed: {e}") from e
|
|
st = dest.stat()
|
|
return {
|
|
"ok": True,
|
|
"message": "备份完成",
|
|
"name": dest.name,
|
|
"path": str(dest),
|
|
"size_bytes": st.st_size,
|
|
}
|
|
|
|
|
|
@router.get("/backup/download/{backup_name}")
|
|
def download_backup(backup_name: str) -> FileResponse:
|
|
s = get_settings()
|
|
try:
|
|
path = resolve_backup_file(s.backup_dir_path, backup_name.strip())
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e)) from e
|
|
except FileNotFoundError as e:
|
|
raise HTTPException(status_code=404, detail=str(e)) from e
|
|
return FileResponse(
|
|
path=str(path),
|
|
filename=path.name,
|
|
media_type="application/octet-stream",
|
|
)
|
|
|
|
|
|
@router.post("/restore")
|
|
def restore_backup(body: RestoreBody) -> dict:
|
|
_verify_password_or_401(body.current_password)
|
|
s = get_settings()
|
|
try:
|
|
backup_path = resolve_backup_file(s.backup_dir_path, body.backup_name.strip())
|
|
restore_db(backup_path, s.db_path)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e)) from e
|
|
except FileNotFoundError as e:
|
|
raise HTTPException(status_code=404, detail=str(e)) from e
|
|
except OSError as e:
|
|
raise HTTPException(status_code=500, detail=f"restore failed: {e}") from e
|
|
return {
|
|
"ok": True,
|
|
"message": f"已从 {body.backup_name} 恢复数据库",
|
|
"restored_from": body.backup_name,
|
|
}
|
|
|
|
|
|
@router.get("/account")
|
|
def get_account() -> dict:
|
|
s = get_settings()
|
|
return {
|
|
"username": s.admin_username,
|
|
"auth_required": not auth_disabled(),
|
|
"env_file": str(s.env_file_path),
|
|
}
|
|
|
|
|
|
@router.put("/account")
|
|
def update_account(body: UpdateAccountBody) -> dict:
|
|
s = get_settings()
|
|
if auth_disabled():
|
|
raise HTTPException(status_code=400, detail="auth disabled; edit .env manually")
|
|
|
|
if not verify_credentials(s.admin_username, body.current_password):
|
|
raise HTTPException(status_code=401, detail="current password incorrect")
|
|
|
|
if body.new_password is not None and len(body.new_password) < 6:
|
|
raise HTTPException(status_code=400, detail="password must be at least 6 characters")
|
|
|
|
new_user = (body.new_username or s.admin_username).strip()
|
|
new_pass = body.new_password or s.admin_password
|
|
|
|
if body.new_username is not None and not _USERNAME_RE.fullmatch(new_user):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="username: 2-32 chars, letters/digits/_/- only",
|
|
)
|
|
|
|
if new_user == s.admin_username and new_pass == s.admin_password:
|
|
return {"ok": True, "changed": False, "message": "no changes", "username": new_user}
|
|
|
|
updates = {
|
|
"ADMIN_USERNAME": new_user,
|
|
"ADMIN_PASSWORD": new_pass,
|
|
}
|
|
path = s.env_file_path
|
|
try:
|
|
update_env_file(path, updates)
|
|
except OSError as e:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"failed to write {path}: {e}",
|
|
) from e
|
|
|
|
reload_settings()
|
|
return {
|
|
"ok": True,
|
|
"changed": True,
|
|
"username": new_user,
|
|
"relogin_required": True,
|
|
"message": "账号已更新，请重新登录",
|
|
}
|