commit 4e3a4b4602f7fa61a73405f61ad89d96dc6e1660 Author: publ Date: Mon Jun 22 18:27:41 2026 +0800 feat: 数据备份管理系统 v0.1.0 带 Web 操作界面的备份系统，支持： - 数据源：MySQL 数据库（mysqldump）、服务器目录（tar.gz） - 存储目标：本地目录、S3 兼容对象存储（MinIO/Ceph/AWS S3） - 触发方式：手动 + Cron 定时调度（APScheduler） - 备份还原：MySQL 库、目录 - JWT 登录认证 + Fernet 字段加密 - 保留策略：按数量 + 按天数双重清理 技术栈：FastAPI + SQLAlchemy 2 + APScheduler + aioboto3； 前端 Vite + React 18 + TypeScript + Ant Design 5 + Zustand。 Docker Compose 一键起。 Co-Authored-By: Claude diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..281a85c --- /dev/null +++ b/.env.example @@ -0,0 +1,21 @@ +# ====== 必须设置 ====== +# JWT 签名密钥：openssl rand -hex 32 +SECRET_KEY=please-change-me-to-a-random-32-byte-hex-string + +# Fernet 字段加密密钥：python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())" +FERNET_KEY=please-generate-a-fernet-key-and-paste-here + +# 首次启动创建 admin 账号的初始密码，启动后可修改 +INITIAL_ADMIN_PASSWORD=ChangeMe123! + +# ====== 可选配置 ====== +JWT_EXPIRE_MINUTES=1440 +DATABASE_URL=sqlite:///./data/backup.db +STORAGE_LOCAL_ROOT=./data/backups +LOG_LEVEL=INFO + +# CORS：生产环境请填写前端实际域名 +CORS_ORIGINS=http://localhost:5173,http://localhost:8000 + +# 时区 +TZ=Asia/Shanghai diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..abd8b74 --- /dev/null +++ b/.gitignore @@ -0,0 +1,45 @@ +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +.venv/ +venv/ +env/ +.pytest_cache/ +.mypy_cache/ +.ruff_cache/ +*.egg-info/ +dist/ +build/ + +# Node +node_modules/ +dist/ +.vite/ +*.local + +# 环境与密钥 +.env +.env.local +*.key +*.pem + +# 数据与日志 +backend/data/ +backend/logs/ +*.db +*.sqlite +*.sqlite3 +backups/ + +# IDE +.idea/ +.vscode/ +*.swp +*.swo +.DS_Store + +# Docker +docker-compose.override.yml diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..6d4f75e --- /dev/null +++ b/Makefile @@ -0,0 +1,50 @@ +.PHONY: help init keys up down logs restart build rebuild backend frontend clean + +help: + @echo "Available commands:" + @echo " make init - 复制 .env.example 到 .env（如不存在）" + @echo " make keys - 生成 SECRET_KEY 和 FERNET_KEY 并写入 .env" + @echo " make up - 启动所有服务（首次会自动 build）" + @echo " make down - 停止所有服务" + @echo " make logs - 跟踪查看 backend 日志" + @echo " make restart - 重启所有服务" + @echo " make build - 构建镜像" + @echo " make rebuild - 强制重新构建（无缓存）" + @echo " make clean - 停止服务并清理 volumes（会删除数据！）" + +init: + @if [ ! -f .env ]; then cp .env.example .env && echo "[ok] .env created"; else echo "[skip] .env already exists"; fi + +keys: + @if grep -q "please-change-me" .env 2>/dev/null; then \ + echo "Generating SECRET_KEY..."; \ + sed -i "s|^SECRET_KEY=.*|SECRET_KEY=$$(openssl rand -hex 32)|" .env; \ + echo "Generating FERNET_KEY..."; \ + sed -i "s|^FERNET_KEY=.*|FERNET_KEY=$$(python3 -c 'from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())')|" .env; \ + echo "[ok] keys generated"; \ + else \ + echo "[skip] .env already has real keys"; \ + fi + +up: + docker compose up -d --build + @echo "[ok] services started. Frontend: http://localhost:5173 Backend: http://localhost:8000" + +down: + docker compose down + +logs: + docker compose logs -f backend + +restart: + docker compose restart + +build: + docker compose build + +rebuild: + docker compose build --no-cache + +clean: + docker compose down -v + @echo "[warn] volumes removed - all backup metadata and local backups deleted!" diff --git a/README.md b/README.md new file mode 100644 index 0000000..58b4efb --- /dev/null +++ b/README.md @@ -0,0 +1,85 @@ +# 数据备份管理系统 + +一个带 Web 操作界面的备份系统，支持将 **MySQL 数据库** 和 **服务器目录** 备份到 **本地目录** 或 **S3 兼容对象存储**（AWS S3 / MinIO / Ceph）。支持手动触发、Cron 定时调度、备份保留策略，以及从备份还原。 + +## 功能 + +- 📦 **多数据源**：MySQL 数据库（mysqldump）、服务器目录（tar.gz） +- ☁️ **多存储目标**：本地目录、S3 兼容对象存储 +- ⏰ **Cron 定时调度**：标准 cron 表达式 +- 🔁 **备份还原**：从备份恢复 MySQL 库 / 解压目录 +- 🗂️ **保留策略**：保留最近 N 个 + 保留 N 天 +- 👤 **JWT 登录认证**：单用户/多用户管理后台 +- 📊 **运行历史与日志**：每次备份的执行详情、SHA256 校验、大小、耗时 + +## 技术栈 + +- **后端**：Python 3.11、FastAPI、SQLAlchemy 2、APScheduler、boto3/aioboto3 +- **前端**：Vite + React 18 + TypeScript + Ant Design 5 +- **数据库**：SQLite（可平滑迁移 PostgreSQL） +- **部署**：Docker Compose + +## 快速开始 + +### 1. 准备环境变量 + +```bash +cp .env.example .env +make keys # 自动生成 SECRET_KEY 和 FERNET_KEY +``` + +或手动编辑 `.env`，至少填入以下三项： + +```bash +SECRET_KEY= +FERNET_KEY= +INITIAL_ADMIN_PASSWORD= +``` + +### 2. 启动 + +```bash +make up +``` + +- 前端：**http://localhost:5173** +- 后端 API：**http://localhost:8000** +- API 文档：**http://localhost:8000/docs** + +### 3. 首次使用 + +1. 打开 `http://localhost:5173`，用 `admin` / `INITIAL_ADMIN_PASSWORD` 登录 +2. **Storages** → 新建一个存储目标（Local 路径或 S3 桶） +3. **Jobs → 新建** → 选择数据源类型（MySQL / 目录）→ 填写源信息 → 选存储目标 → 保存 +4. Job 列表点 **Run Now** 立即触发，或填写 `cron_expression` 启用定时调度 + +## 常用命令 + +```bash +make logs # 查看后端日志 +make restart # 重启服务 +make down # 停止服务 +make clean # 停止并清理所有数据 +``` + +## 项目结构 + +``` +. +├── docker-compose.yml +├── .env.example +├── Makefile +├── backend/ # FastAPI 后端 +└── frontend/ # React 前端 +``` + +## 注意事项 + +- ⚠️ **单实例部署**：内置 APScheduler，**不要启动多个 backend 副本**，否则会出现重复调度。多副本需要改造为分布式锁。 +- 🔐 **敏感字段加密**：MySQL 密码和 S3 SecretKey 在数据库中以 Fernet 加密存储，前端永远看不到明文。 +- 💾 **数据持久化**：所有元数据在 `backend-data` 卷，本地备份在 `local-backups` 卷。删除前请先 `make down` 备份。 +- 🗄️ **mysqldump 必须可用**：备份 MySQL 时后端容器需能调用 `mysqldump`（官方镜像已安装）。 + +## License + +MIT diff --git a/backend/.env.example b/backend/.env.example new file mode 100644 index 0000000..522dd9b --- /dev/null +++ b/backend/.env.example @@ -0,0 +1,9 @@ +# 同根目录 .env，这里只是文档。实际加载根目录 .env。 +SECRET_KEY= +FERNET_KEY= +INITIAL_ADMIN_PASSWORD=ChangeMe123! +JWT_EXPIRE_MINUTES=1440 +DATABASE_URL=sqlite:///./data/backup.db +STORAGE_LOCAL_ROOT=./data/backups +LOG_LEVEL=INFO +CORS_ORIGINS=http://localhost:5173,http://localhost:8000 diff --git a/backend/Dockerfile b/backend/Dockerfile new file mode 100644 index 0000000..2b7765c --- /dev/null +++ b/backend/Dockerfile @@ -0,0 +1,31 @@ +FROM python:3.11-slim + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + PIP_NO_CACHE_DIR=1 \ + PIP_DISABLE_PIP_VERSION_CHECK=1 + +# 系统依赖：curl 用于健康检查，default-mysql-client 提供 mysqldump，tar 已内置 +RUN apt-get update && apt-get install -y --no-install-recommends \ + curl \ + default-mysql-client \ + tar \ + gzip \ + tzdata \ + && rm -rf /var/lib/apt/lists/* + +WORKDIR /app + +# 先安装依赖以利用 Docker 缓存 +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +COPY . . + +# 数据、日志、备份目录 +RUN mkdir -p /app/data/backups /app/logs + +EXPOSE 8000 + +# alembic 升级 + 启动 uvicorn +CMD ["sh", "-c", "alembic upgrade head && uvicorn app.main:app --host 0.0.0.0 --port 8000"] diff --git a/backend/alembic.ini b/backend/alembic.ini new file mode 100644 index 0000000..487a70f --- /dev/null +++ b/backend/alembic.ini @@ -0,0 +1,43 @@ +[alembic] +script_location = alembic +prepend_sys_path = . +sqlalchemy.url = sqlite:///./data/backup.db +file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s +timezone = UTC +truncate_slug_length = 40 + +[post_write_hooks] + +[loggers] +keys = root,sqlalchemy,alembic + +[handlers] +keys = console + +[formatters] +keys = generic + +[logger_root] +level = WARN +handlers = console +qualname = + +[logger_sqlalchemy] +level = WARN +handlers = +qualname = sqlalchemy.engine + +[logger_alembic] +level = INFO +handlers = +qualname = alembic + +[handler_console] +class = StreamHandler +args = (sys.stderr,) +level = NOTSET +formatter = generic + +[formatter_generic] +format = %(levelname)-5.5s [%(name)s] %(message)s +datefmt = %H:%M:%S diff --git a/backend/alembic/env.py b/backend/alembic/env.py new file mode 100644 index 0000000..2351bfa --- /dev/null +++ b/backend/alembic/env.py @@ -0,0 +1,59 @@ +"""Alembic env.py：从 app.config 读取 DATABASE_URL，使用 Base.metadata。""" +import os +import sys +from logging.config import fileConfig + +from alembic import context +from sqlalchemy import engine_from_config, pool + +# 让 alembic 能 import app.* +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from app.config import get_settings # noqa: E402 +from app.database import Base # noqa: E402 +import app.models # noqa: F401, E402 # 触发所有模型注册 + +config = context.config + +# 用 settings.DATABASE_URL 覆盖 alembic.ini 中的 url +config.set_main_option("sqlalchemy.url", get_settings().DATABASE_URL) + +if config.config_file_name is not None: + fileConfig(config.config_file_name) + +target_metadata = Base.metadata + + +def run_migrations_offline() -> None: + url = config.get_main_option("sqlalchemy.url") + context.configure( + url=url, + target_metadata=target_metadata, + literal_binds=True, + dialect_opts={"paramstyle": "named"}, + render_as_batch=True, # SQLite 兼容 + ) + with context.begin_transaction(): + context.run_migrations() + + +def run_migrations_online() -> None: + connectable = engine_from_config( + config.get_section(config.config_ini_section, {}), + prefix="sqlalchemy.", + poolclass=pool.NullPool, + ) + with connectable.connect() as connection: + context.configure( + connection=connection, + target_metadata=target_metadata, + render_as_batch=True, # SQLite 兼容 + ) + with context.begin_transaction(): + context.run_migrations() + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() diff --git a/backend/alembic/script.py.mako b/backend/alembic/script.py.mako new file mode 100644 index 0000000..36cd8ef --- /dev/null +++ b/backend/alembic/script.py.mako @@ -0,0 +1,22 @@ +"""${message} + +Revision ID: ${up_revision} +Revises: ${down_revision | comma,n} +Create Date: ${create_date} +""" +from alembic import op +import sqlalchemy as sa +${imports if imports else ""} + +revision = ${repr(up_revision)} +down_revision = ${repr(down_revision)} +branch_labels = ${repr(branch_labels)} +depends_on = ${repr(depends_on)} + + +def upgrade() -> None: + ${upgrades if upgrades else "pass"} + + +def downgrade() -> None: + ${downgrades if downgrades else "pass"} diff --git a/backend/alembic/versions/0001_initial.py b/backend/alembic/versions/0001_initial.py new file mode 100644 index 0000000..f973e8e --- /dev/null +++ b/backend/alembic/versions/0001_initial.py @@ -0,0 +1,115 @@ +"""initial schema + +Revision ID: 0001_initial +Revises: +Create Date: 2026-06-22 00:00:00 +""" +from alembic import op +import sqlalchemy as sa + + +revision = "0001_initial" +down_revision = None +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "users", + sa.Column("id", sa.Integer, primary_key=True), + sa.Column("username", sa.String(64), nullable=False, unique=True), + sa.Column("password_hash", sa.String(255), nullable=False), + sa.Column("is_admin", sa.Boolean, nullable=False, server_default=sa.text("0")), + sa.Column("is_active", sa.Boolean, nullable=False, server_default=sa.text("1")), + sa.Column("created_at", sa.DateTime, nullable=False, server_default=sa.func.now()), + ) + op.create_index("ix_users_username", "users", ["username"], unique=True) + + op.create_table( + "storages", + sa.Column("id", sa.Integer, primary_key=True), + sa.Column("name", sa.String(64), nullable=False, unique=True), + sa.Column("type", sa.String(16), nullable=False), + sa.Column("config_json", sa.Text, nullable=False), + sa.Column("is_default", sa.Boolean, nullable=False, server_default=sa.text("0")), + sa.Column("created_at", sa.DateTime, nullable=False, server_default=sa.func.now()), + sa.Column("updated_at", sa.DateTime, nullable=False, server_default=sa.func.now()), + ) + + op.create_table( + "jobs", + sa.Column("id", sa.Integer, primary_key=True), + sa.Column("name", sa.String(64), nullable=False, unique=True), + sa.Column("type", sa.String(16), nullable=False), + sa.Column("source_config_json", sa.Text, nullable=False), + sa.Column("storage_id", sa.Integer, sa.ForeignKey("storages.id", ondelete="RESTRICT"), nullable=False), + sa.Column("cron_expression", sa.String(64), nullable=True), + sa.Column("enabled", sa.Boolean, nullable=False, server_default=sa.text("1")), + sa.Column("retention_count", sa.Integer, nullable=False, server_default=sa.text("7")), + sa.Column("retention_days", sa.Integer, nullable=True), + sa.Column("description", sa.String(255), nullable=True), + sa.Column("created_at", sa.DateTime, nullable=False, server_default=sa.func.now()), + sa.Column("updated_at", sa.DateTime, nullable=False, server_default=sa.func.now()), + ) + + op.create_table( + "runs", + sa.Column("id", sa.Integer, primary_key=True), + sa.Column("job_id", sa.Integer, sa.ForeignKey("jobs.id", ondelete="CASCADE"), nullable=False), + sa.Column("status", sa.String(16), nullable=False, server_default="pending"), + sa.Column("trigger", sa.String(16), nullable=False, server_default="manual"), + sa.Column("started_at", sa.DateTime, nullable=True), + sa.Column("finished_at", sa.DateTime, nullable=True), + sa.Column("duration_seconds", sa.Integer, nullable=True), + sa.Column("artifact_path", sa.String(512), nullable=True), + sa.Column("artifact_size", sa.BigInteger, nullable=True), + sa.Column("artifact_checksum", sa.String(64), nullable=True), + sa.Column("error_message", sa.Text, nullable=True), + sa.Column("log_excerpt", sa.Text, nullable=True), + sa.Column("created_at", sa.DateTime, nullable=False, server_default=sa.func.now()), + ) + op.create_index("ix_runs_job_id", "runs", ["job_id"]) + + op.create_table( + "run_logs", + sa.Column("id", sa.Integer, primary_key=True), + sa.Column("run_id", sa.Integer, sa.ForeignKey("runs.id", ondelete="CASCADE"), nullable=False), + sa.Column("content", sa.Text, nullable=False), + sa.Column("created_at", sa.DateTime, nullable=False, server_default=sa.func.now()), + ) + op.create_index("ix_run_logs_run_id", "run_logs", ["run_id"]) + + op.create_table( + "settings", + sa.Column("key", sa.String(64), primary_key=True), + sa.Column("value", sa.Text, nullable=False, server_default=""), + sa.Column("updated_at", sa.DateTime, nullable=False, server_default=sa.func.now()), + ) + + op.create_table( + "restore_history", + sa.Column("id", sa.Integer, primary_key=True), + sa.Column("run_id", sa.Integer, sa.ForeignKey("runs.id", ondelete="CASCADE"), nullable=False), + sa.Column("target", sa.String(512), nullable=False), + sa.Column("status", sa.String(16), nullable=False, server_default="pending"), + sa.Column("started_at", sa.DateTime, nullable=True), + sa.Column("finished_at", sa.DateTime, nullable=True), + sa.Column("error_message", sa.Text, nullable=True), + sa.Column("created_at", sa.DateTime, nullable=False, server_default=sa.func.now()), + ) + op.create_index("ix_restore_history_run_id", "restore_history", ["run_id"]) + + +def downgrade() -> None: + op.drop_index("ix_restore_history_run_id", table_name="restore_history") + op.drop_table("restore_history") + op.drop_table("settings") + op.drop_index("ix_run_logs_run_id", table_name="run_logs") + op.drop_table("run_logs") + op.drop_index("ix_runs_job_id", table_name="runs") + op.drop_table("runs") + op.drop_table("jobs") + op.drop_table("storages") + op.drop_index("ix_users_username", table_name="users") + op.drop_table("users") diff --git a/backend/app/__init__.py b/backend/app/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/app/api/__init__.py b/backend/app/api/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/app/api/auth.py b/backend/app/api/auth.py new file mode 100644 index 0000000..5ad4744 --- /dev/null +++ b/backend/app/api/auth.py @@ -0,0 +1,39 @@ +"""认证 API。""" +from fastapi import APIRouter, HTTPException, status + +from app.config import get_settings +from app.deps import CurrentUser, DBSession +from app.models.user import User +from app.schemas.auth import ( + ChangePasswordRequest, + LoginRequest, + MessageResponse, + TokenResponse, + UserOut, +) +from app.security import create_access_token, hash_password, verify_password + +router = APIRouter(prefix="/auth", tags=["auth"]) + + +@router.post("/login", response_model=TokenResponse) +def login(payload: LoginRequest, db: DBSession) -> TokenResponse: + user = db.query(User).filter(User.username == payload.username).first() + if not user or not user.is_active or not verify_password(payload.password, user.password_hash): + raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials") + token = create_access_token(user.username, extra={"uid": user.id, "adm": user.is_admin}) + return TokenResponse(access_token=token, expires_in=get_settings().JWT_EXPIRE_MINUTES * 60) + + +@router.get("/me", response_model=UserOut) +def me(user: CurrentUser) -> User: + return user + + +@router.post("/change-password", response_model=MessageResponse) +def change_password(payload: ChangePasswordRequest, user: CurrentUser, db: DBSession) -> MessageResponse: + if not verify_password(payload.old_password, user.password_hash): + raise HTTPException(status_code=400, detail="原密码错误") + user.password_hash = hash_password(payload.new_password) + db.commit() + return MessageResponse(message="Password updated") diff --git a/backend/app/api/jobs.py b/backend/app/api/jobs.py new file mode 100644 index 0000000..e354d5b --- /dev/null +++ b/backend/app/api/jobs.py @@ -0,0 +1,159 @@ +"""备份任务 CRUD API。""" +import asyncio +from datetime import datetime + +from fastapi import APIRouter, BackgroundTasks, HTTPException +from pydantic import ValidationError + +from app.core.executor import run_backup +from app.deps import AdminUser, CurrentUser, DBSession +from app.models.job import Job +from app.models.run import Run +from app.models.storage import Storage +from app.schemas.job import ( + DirectorySourceConfig, + JobCreate, + JobOut, + JobRunResponse, + JobUpdate, + JobWithLastRun, + MySQLSourceConfig, + RunSummary, +) +from app.scheduler.scheduler import remove_job, reschedule_job +from app.utils.crypto import encrypt_dict + +router = APIRouter(prefix="/jobs", tags=["jobs"]) + + +def _validate_source(type_: str, cfg: dict) -> dict: + if type_ == "mysql": + return MySQLSourceConfig(**cfg).model_dump() + if type_ == "directory": + return DirectorySourceConfig(**cfg).model_dump() + raise HTTPException(status_code=400, detail=f"Unsupported job type: {type_}") + + +def _job_to_with_last_run(job: Job, db) -> JobWithLastRun: + last = ( + db.query(Run) + .filter(Run.job_id == job.id) + .order_by(Run.id.desc()) + .first() + ) + out = JobWithLastRun.model_validate(job) + out.last_run = RunSummary.model_validate(last) if last else None + return out + + +@router.get("", response_model=list[JobWithLastRun]) +def list_jobs(db: DBSession, _: CurrentUser) -> list[JobWithLastRun]: + jobs = db.query(Job).order_by(Job.id).all() + return [_job_to_with_last_run(j, db) for j in jobs] + + +@router.post("", response_model=JobOut, status_code=201) +def create_job(payload: JobCreate, db: DBSession, _: AdminUser) -> Job: + if db.query(Job).filter(Job.name == payload.name).first(): + raise HTTPException(status_code=400, detail="名称已存在") + if not db.get(Storage, payload.storage_id): + raise HTTPException(status_code=400, detail="存储目标不存在") + src = _validate_source(payload.type, payload.source_config) + job = Job( + name=payload.name, + type=payload.type, + source_config_json=encrypt_dict(src), + storage_id=payload.storage_id, + cron_expression=payload.cron_expression, + enabled=payload.enabled, + retention_count=payload.retention_count, + retention_days=payload.retention_days, + description=payload.description, + ) + db.add(job) + db.commit() + db.refresh(job) + reschedule_job(job.id) + return job + + +@router.get("/{job_id}", response_model=JobWithLastRun) +def get_job(job_id: int, db: DBSession, _: CurrentUser) -> JobWithLastRun: + job = db.get(Job, job_id) + if not job: + raise HTTPException(status_code=404, detail="Not found") + return _job_to_with_last_run(job, db) + + +@router.put("/{job_id}", response_model=JobOut) +def update_job(job_id: int, payload: JobUpdate, db: DBSession, _: AdminUser) -> Job: + job = db.get(Job, job_id) + if not job: + raise HTTPException(status_code=404, detail="Not found") + if payload.name is not None: + if db.query(Job).filter(Job.name == payload.name, Job.id != job_id).first(): + raise HTTPException(status_code=400, detail="名称已存在") + job.name = payload.name + if payload.source_config is not None: + src = _validate_source(job.type, payload.source_config) + job.source_config_json = encrypt_dict(src) + if payload.storage_id is not None: + if not db.get(Storage, payload.storage_id): + raise HTTPException(status_code=400, detail="存储目标不存在") + job.storage_id = payload.storage_id + if payload.cron_expression is not None: + job.cron_expression = payload.cron_expression or None + if payload.enabled is not None: + job.enabled = payload.enabled + if payload.retention_count is not None: + job.retention_count = payload.retention_count + if payload.retention_days is not None: + job.retention_days = payload.retention_days + if payload.description is not None: + job.description = payload.description + db.commit() + db.refresh(job) + reschedule_job(job.id) + return job + + +@router.delete("/{job_id}", status_code=204) +def delete_job(job_id: int, db: DBSession, _: AdminUser) -> None: + job = db.get(Job, job_id) + if not job: + raise HTTPException(status_code=404, detail="Not found") + remove_job(job.id) + db.delete(job) + db.commit() + + +@router.post("/{job_id}/run", response_model=JobRunResponse) +async def run_job(job_id: int, db: DBSession, _: CurrentUser) -> JobRunResponse: + job = db.get(Job, job_id) + if not job: + raise HTTPException(status_code=404, detail="Not found") + run = Run( + job_id=job.id, + status="pending", + trigger="manual", + started_at=datetime.utcnow(), + ) + db.add(run) + db.commit() + db.refresh(run) + run_id = run.id + # 后台启动 + asyncio.create_task(run_backup(run_id, trigger="manual")) + return JobRunResponse(run_id=run_id, status=run.status) + + +@router.post("/{job_id}/toggle", response_model=JobOut) +def toggle_job(job_id: int, db: DBSession, _: AdminUser) -> Job: + job = db.get(Job, job_id) + if not job: + raise HTTPException(status_code=404, detail="Not found") + job.enabled = not job.enabled + db.commit() + db.refresh(job) + reschedule_job(job.id) + return job diff --git a/backend/app/api/restore.py b/backend/app/api/restore.py new file mode 100644 index 0000000..14169fc --- /dev/null +++ b/backend/app/api/restore.py @@ -0,0 +1,76 @@ +"""恢复 API。""" +import asyncio +from datetime import datetime + +from fastapi import APIRouter, HTTPException + +from app.core.restore import execute_restore +from app.deps import AdminUser, DBSession +from app.models.job import Job +from app.models.restore import RestoreHistory +from app.models.run import Run +from app.schemas.restore import RestoreOut, RestoreRequest, RestoreTargetsResponse +from app.utils.crypto import decrypt_dict + +router = APIRouter(prefix="/restore", tags=["restore"]) + + +@router.get("/targets", response_model=RestoreTargetsResponse) +def get_restore_targets(run_id: int, db: DBSession, _: AdminUser) -> RestoreTargetsResponse: + run = db.get(Run, run_id) + if not run: + raise HTTPException(status_code=404, detail="Run not found") + job = db.get(Job, run.job_id) + if not job: + raise HTTPException(status_code=404, detail="Job not found") + + src = decrypt_dict(job.source_config_json) + suggested: list[str] = [] + if job.type == "mysql": + suggested.append(src.get("database", "")) + elif job.type == "directory": + suggested.append(src.get("path", "")) + + return RestoreTargetsResponse( + run_id=run.id, + job_type=job.type, + artifact_size=run.artifact_size, + artifact_checksum=run.artifact_checksum, + suggested_targets=[s for s in suggested if s], + ) + + +@router.post("", response_model=RestoreOut, status_code=201) +async def trigger_restore(payload: RestoreRequest, db: DBSession, _: AdminUser) -> RestoreHistory: + run = db.get(Run, payload.run_id) + if not run or not run.artifact_path: + raise HTTPException(status_code=404, detail="Run not found or no artifact") + job = db.get(Job, run.job_id) + if not job: + raise HTTPException(status_code=404, detail="Job not found") + + # 校验 target_type 与 job.type 匹配 + if job.type == "mysql" and payload.target_type != "mysql_db": + raise HTTPException(status_code=400, detail="target_type 与源类型不匹配（应为 mysql_db）") + if job.type == "directory" and payload.target_type != "directory_path": + raise HTTPException(status_code=400, detail="target_type 与源类型不匹配（应为 directory_path）") + + rh = RestoreHistory( + run_id=run.id, + target=payload.target, + status="pending", + ) + db.add(rh) + db.commit() + db.refresh(rh) + restore_id = rh.id + asyncio.create_task(execute_restore(restore_id)) + return rh + + +@router.get("/history", response_model=list[RestoreOut]) +def restore_history(db: DBSession, _: AdminUser, run_id: int | None = None) -> list[RestoreHistory]: + q = db.query(RestoreHistory) + if run_id is not None: + q = q.filter(RestoreHistory.run_id == run_id) + return q.order_by(RestoreHistory.id.desc()).limit(200).all() diff --git a/backend/app/api/router.py b/backend/app/api/router.py new file mode 100644 index 0000000..9b9018d --- /dev/null +++ b/backend/app/api/router.py @@ -0,0 +1,11 @@ +"""API 总路由。""" +from fastapi import APIRouter + +from app.api import auth, jobs, restore, runs, storages + +api_router = APIRouter() +api_router.include_router(auth.router) +api_router.include_router(storages.router) +api_router.include_router(jobs.router) +api_router.include_router(runs.router) +api_router.include_router(restore.router) diff --git a/backend/app/api/runs.py b/backend/app/api/runs.py new file mode 100644 index 0000000..c452452 --- /dev/null +++ b/backend/app/api/runs.py @@ -0,0 +1,96 @@ +"""备份运行历史 API。""" +from fastapi import APIRouter, HTTPException +from fastapi.responses import StreamingResponse + +from app.core.storage.factory import create_storage +from app.deps import CurrentUser, DBSession +from app.models.job import Job +from app.models.run import Run, RunLog +from app.schemas.run import RunListResponse, RunLogOut, RunOut + +router = APIRouter(prefix="/runs", tags=["runs"]) + + +@router.get("", response_model=RunListResponse) +def list_runs( + db: DBSession, + _: CurrentUser, + job_id: int | None = None, + status: str | None = None, + page: int = 1, + page_size: int = 20, +) -> RunListResponse: + page = max(1, page) + page_size = max(1, min(100, page_size)) + + q = db.query(Run) + if job_id is not None: + q = q.filter(Run.job_id == job_id) + if status: + q = q.filter(Run.status == status) + total = q.count() + items = ( + q.order_by(Run.id.desc()) + .offset((page - 1) * page_size) + .limit(page_size) + .all() + ) + return RunListResponse(total=total, items=[RunOut.model_validate(r) for r in items]) + + +@router.get("/{run_id}", response_model=RunOut) +def get_run(run_id: int, db: DBSession, _: CurrentUser) -> Run: + run = db.get(Run, run_id) + if not run: + raise HTTPException(status_code=404, detail="Not found") + return run + + +@router.get("/{run_id}/log", response_model=RunLogOut) +def get_run_log(run_id: int, db: DBSession, _: CurrentUser) -> RunLog: + run = db.get(Run, run_id) + if not run: + raise HTTPException(status_code=404, detail="Not found") + log = db.query(RunLog).filter(RunLog.run_id == run_id).order_by(RunLog.id.desc()).first() + if not log: + raise HTTPException(status_code=404, detail="No log") + return log + + +@router.get("/{run_id}/download") +async def download_run(run_id: int, db: DBSession, _: CurrentUser): + run = db.get(Run, run_id) + if not run or not run.artifact_path: + raise HTTPException(status_code=404, detail="Not found") + job = db.get(Job, run.job_id) + if not job: + raise HTTPException(status_code=404, detail="Job not found") + storage = create_storage(job.storage) + + filename = run.artifact_path.split("/")[-1] + + async def stream(): + async for chunk in storage.download_stream(run.artifact_path): + yield chunk + + return StreamingResponse( + stream(), + media_type="application/octet-stream", + headers={"Content-Disposition": f'attachment; filename="{filename}"'}, + ) + + +@router.delete("/{run_id}", status_code=204) +async def delete_run(run_id: int, db: DBSession, _: CurrentUser) -> None: + run = db.get(Run, run_id) + if not run: + raise HTTPException(status_code=404, detail="Not found") + job = db.get(Job, run.job_id) + if job and run.artifact_path: + try: + storage = create_storage(job.storage) + await storage.delete(run.artifact_path) + except Exception: + pass + db.delete(run) + db.commit() diff --git a/backend/app/api/storages.py b/backend/app/api/storages.py new file mode 100644 index 0000000..e2f2575 --- /dev/null +++ b/backend/app/api/storages.py @@ -0,0 +1,100 @@ +"""存储目标 CRUD API。""" +import asyncio + +from fastapi import APIRouter, HTTPException +from pydantic import ValidationError + +from app.core.storage.factory import create_storage as make_storage_impl +from app.deps import AdminUser, DBSession +from app.models.storage import Storage +from app.schemas.storage import ( + LocalStorageConfig, + S3StorageConfig, + StorageCreate, + StorageOut, + StorageTestResponse, + StorageUpdate, +) +from app.utils.crypto import decrypt_dict, encrypt_dict + +router = APIRouter(prefix="/storages", tags=["storages"]) + + +def _validate_config(type_: str, config: dict) -> dict: + if type_ == "local": + return LocalStorageConfig(**config).model_dump() + if type_ == "s3": + return S3StorageConfig(**config).model_dump() + raise HTTPException(status_code=400, detail=f"Unsupported type: {type_}") + + +@router.get("", response_model=list[StorageOut]) +def list_storages(db: DBSession, _: AdminUser) -> list[Storage]: + return db.query(Storage).order_by(Storage.id).all() + + +@router.post("", response_model=StorageOut, status_code=201) +def create_storage_endpoint(payload: StorageCreate, db: DBSession, _: AdminUser) -> Storage: + if db.query(Storage).filter(Storage.name == payload.name).first(): + raise HTTPException(status_code=400, detail="名称已存在") + config = _validate_config(payload.type, payload.config) + row = Storage( + name=payload.name, + type=payload.type, + config_json=encrypt_dict(config), + is_default=payload.is_default, + ) + if payload.is_default: + db.query(Storage).update({Storage.is_default: False}) + db.add(row) + db.commit() + db.refresh(row) + return row + + +@router.get("/{storage_id}", response_model=StorageOut) +def get_storage(storage_id: int, db: DBSession, _: AdminUser) -> Storage: + row = db.get(Storage, storage_id) + if not row: + raise HTTPException(status_code=404, detail="Not found") + return row + + +@router.put("/{storage_id}", response_model=StorageOut) +def update_storage(storage_id: int, payload: StorageUpdate, db: DBSession, _: AdminUser) -> Storage: + row = db.get(Storage, storage_id) + if not row: + raise HTTPException(status_code=404, detail="Not found") + if payload.name is not None: + if db.query(Storage).filter(Storage.name == payload.name, Storage.id != storage_id).first(): + raise HTTPException(status_code=400, detail="名称已存在") + row.name = payload.name + if payload.config is not None: + config = _validate_config(row.type, payload.config) + row.config_json = encrypt_dict(config) + if payload.is_default is not None: + if payload.is_default: + db.query(Storage).update({Storage.is_default: False}) + row.is_default = payload.is_default + db.commit() + db.refresh(row) + return row + + +@router.delete("/{storage_id}", status_code=204) +def delete_storage(storage_id: int, db: DBSession, _: AdminUser) -> None: + row = db.get(Storage, storage_id) + if not row: + raise HTTPException(status_code=404, detail="Not found") + db.delete(row) + db.commit() + + +@router.post("/{storage_id}/test", response_model=StorageTestResponse) +async def test_storage(storage_id: int, db: DBSession, _: AdminUser) -> StorageTestResponse: + row = db.get(Storage, storage_id) + if not row: + raise HTTPException(status_code=404, detail="Not found") + storage = make_storage_impl(row) + ok, msg = await storage.test_connection() + return StorageTestResponse(ok=ok, message=msg) diff --git a/backend/app/config.py b/backend/app/config.py new file mode 100644 index 0000000..d796ef8 --- /dev/null +++ b/backend/app/config.py @@ -0,0 +1,44 @@ +"""应用配置：从环境变量加载，启动时校验必填项。""" +from functools import lru_cache +from typing import List + +from pydantic import Field, field_validator +from pydantic_settings import BaseSettings, SettingsConfigDict + + +class Settings(BaseSettings): + model_config = SettingsConfigDict( + env_file=".env", + env_file_encoding="utf-8", + case_sensitive=True, + extra="ignore", + ) + + # ===== 必填 ===== + SECRET_KEY: str = Field(..., description="JWT 签名密钥（>= 32 字符）") + FERNET_KEY: str = Field(..., description="Fernet 字段加密密钥") + INITIAL_ADMIN_PASSWORD: str = Field(..., min_length=8) + + # ===== 可选 ===== + JWT_EXPIRE_MINUTES: int = 1440 + DATABASE_URL: str = "sqlite:///./data/backup.db" + STORAGE_LOCAL_ROOT: str = "./data/backups" + LOG_LEVEL: str = "INFO" + CORS_ORIGINS: str = "http://localhost:5173,http://localhost:8000" + TZ: str = "UTC" + + @field_validator("SECRET_KEY") + @classmethod + def _check_secret_key(cls, v: str) -> str: + if len(v) < 32: + raise ValueError("SECRET_KEY 长度必须 >= 32 字符（建议 openssl rand -hex 32）") + return v + + @property + def cors_origins_list(self) -> List[str]: + return [o.strip() for o in self.CORS_ORIGINS.split(",") if o.strip()] + + +@lru_cache +def get_settings() -> Settings: + return Settings() # type: ignore[call-arg] diff --git a/backend/app/core/__init__.py b/backend/app/core/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/app/core/backup/__init__.py b/backend/app/core/backup/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/app/core/backup/base.py b/backend/app/core/backup/base.py new file mode 100644 index 0000000..45194ea --- /dev/null +++ b/backend/app/core/backup/base.py @@ -0,0 +1,36 @@ +"""备份执行抽象基类：MySQL / Directory 都实现此接口。""" +from abc import ABC, abstractmethod +from dataclasses import dataclass +from typing import AsyncIterator, Optional + + +@dataclass +class BackupMetadata: + """备份产物的元信息。""" + + suggested_filename: str # e.g. "db-20260622-030000.sql.gz" 或 "logs-20260622.tar.gz" + content_type: str = "application/octet-stream" + extra: Optional[dict] = None + + +class BaseBackup(ABC): + """所有数据源类型实现此接口。 + + 方法职责： + - metadata(): 计算建议文件名等 + - produce(): 以异步字节流形式产出备份内容 + """ + + def __init__(self, source_config: dict, job_name: str): + self.source_config = source_config + self.job_name = job_name + + @abstractmethod + def metadata(self) -> BackupMetadata: ... + + @abstractmethod + async def produce(self) -> AsyncIterator[bytes]: + """异步产出备份字节流。""" + # 子类应使用 async with + async generator + ... + yield b"" # pragma: no cover diff --git a/backend/app/core/backup/directory.py b/backend/app/core/backup/directory.py new file mode 100644 index 0000000..e8f9f64 --- /dev/null +++ b/backend/app/core/backup/directory.py @@ -0,0 +1,127 @@ +"""目录备份：使用 tarfile 流式打包为 tar.gz。""" +import asyncio +import fnmatch +import os +from datetime import datetime +from pathlib import Path +from typing import AsyncIterator + +from app.core.backup.base import BackupMetadata, BaseBackup +from app.utils.logging import get_logger + +log = get_logger(__name__) + + +class DirectoryBackup(BaseBackup): + def __init__(self, source_config: dict, job_name: str): + super().__init__(source_config, job_name) + self.path: str = source_config["path"] + self.exclude_patterns: list[str] = source_config.get("exclude_patterns", []) or [] + self.base_dir = os.path.basename(os.path.normpath(self.path)) or "root" + self.root_path = Path(self.path).resolve() + + def metadata(self) -> BackupMetadata: + ts = datetime.now().strftime("%Y%m%d-%H%M%S") + safe = "".join(c if c.isalnum() or c in ("_", "-") else "_" for c in self.base_dir) + return BackupMetadata( + suggested_filename=f"{self.job_name}-{safe}-{ts}.tar.gz", + content_type="application/gzip", + extra={"path": self.path, "base_dir": self.base_dir}, + ) + + def _is_excluded(self, rel_path: str) -> bool: + name = os.path.basename(rel_path) + for pat in self.exclude_patterns: + if fnmatch.fnmatch(name, pat) or fnmatch.fnmatch(rel_path, pat): + return True + return False + + async def produce(self) -> AsyncIterator[bytes]: + if not self.root_path.exists(): + raise FileNotFoundError(f"Source path does not exist: {self.path}") + if not self.root_path.is_dir(): + raise NotADirectoryError(f"Source path is not a directory: {self.path}") + + log.info("Taring directory %s (exclude=%s)", self.path, self.exclude_patterns) + + # 用一个 Queue 在同步线程和异步消费者之间桥接 tar 产出 + queue: asyncio.Queue[bytes | None] = asyncio.Queue(maxsize=32) + + def _tar_worker() -> None: + import tarfile + + try: + with tarfile.open(mode="w:gz", bufsize=1024 * 1024) as tar: + # 顶层目录名固定为 base_dir，便于解压后还原 + for dirpath, dirnames, filenames in os.walk(self.root_path): + # 排除目录 + dirnames[:] = [ + d for d in dirnames + if not self._is_excluded(os.path.relpath(os.path.join(dirpath, d), self.root_path)) + ] + for fn in filenames: + abs_path = os.path.join(dirpath, fn) + rel = os.path.relpath(abs_path, self.root_path) + if self._is_excluded(rel): + continue + try: + tar.add(abs_path, arcname=os.path.join(self.base_dir, rel), recursive=False) + except (PermissionError, OSError) as e: + log.warning("Skip %s: %s", abs_path, e) + continue + # 周期性 flush 到 queue + if tar.fileobj.tell() and (tar.fileobj.tell() % (256 * 1024) == 0): + # 这里我们用 tarfile 的 fileobj 直接喂到 queue 更高效 + pass + except Exception as e: + log.exception("tar worker error: %s", e) + finally: + # tarfile 关闭后会写完 gzip trailer，我们需要在关闭时把 fileobj 全部送出 + pass + + # 实现简化：使用 tarfile 的 fileobj=BytesIO 会在循环中累积， + # 对超大目录会爆内存。下面用一个临时文件 + 异步读取的方案。 + import tempfile + + tmp_path = Path(tempfile.gettempdir()) / f"backup_{self.job_name}_{os.getpid()}.tar.gz" + + def _tar_to_file() -> None: + import tarfile + + try: + with tarfile.open(str(tmp_path), mode="w:gz") as tar: + for dirpath, dirnames, filenames in os.walk(self.root_path): + dirnames[:] = [ + d for d in dirnames + if not self._is_excluded(os.path.relpath(os.path.join(dirpath, d), self.root_path)) + ] + for fn in filenames: + abs_path = os.path.join(dirpath, fn) + rel = os.path.relpath(abs_path, self.root_path) + if self._is_excluded(rel): + continue + try: + tar.add(abs_path, arcname=os.path.join(self.base_dir, rel), recursive=False) + except (PermissionError, OSError) as e: + log.warning("Skip %s: %s", abs_path, e) + except Exception: + log.exception("tar file worker failed") + + loop = asyncio.get_event_loop() + await loop.run_in_executor(None, _tar_to_file) + + if not tmp_path.exists() or tmp_path.stat().st_size == 0: + raise RuntimeError(f"Failed to create tar.gz for {self.path}") + + try: + with open(tmp_path, "rb") as f: + while True: + chunk = await loop.run_in_executor(None, f.read, 64 * 1024) + if not chunk: + break + yield chunk + finally: + try: + tmp_path.unlink(missing_ok=True) + except Exception: + pass diff --git a/backend/app/core/backup/mysql.py b/backend/app/core/backup/mysql.py new file mode 100644 index 0000000..8c4f5df --- /dev/null +++ b/backend/app/core/backup/mysql.py @@ -0,0 +1,97 @@ +"""MySQL 备份：通过 mysqldump 子进程流式输出。""" +import asyncio +import gzip +import shlex +from datetime import datetime +from typing import AsyncIterator + +from app.core.backup.base import BackupMetadata, BaseBackup +from app.utils.logging import get_logger + +log = get_logger(__name__) + +CHUNK_SIZE = 64 * 1024 + + +class MySQLBackup(BaseBackup): + def __init__(self, source_config: dict, job_name: str): + super().__init__(source_config, job_name) + self.host: str = source_config["host"] + self.port: int = int(source_config.get("port", 3306)) + self.user: str = source_config["user"] + self.password: str = source_config.get("password", "") + self.database: str = source_config["database"] + self.extra_args: str = source_config.get("extra_args", "") + + def metadata(self) -> BackupMetadata: + ts = datetime.now().strftime("%Y%m%d-%H%M%S") + safe_db = "".join(c if c.isalnum() or c in ("_", "-") else "_" for c in self.database) + return BackupMetadata( + suggested_filename=f"{self.job_name}-{safe_db}-{ts}.sql.gz", + content_type="application/gzip", + extra={"database": self.database, "host": self.host, "port": self.port}, + ) + + async def produce(self) -> AsyncIterator[bytes]: + args = [ + "mysqldump", + "-h", self.host, + "-P", str(self.port), + "-u", self.user, + "--single-transaction", + "--quick", + "--routines", + "--triggers", + "--events", + "--default-character-set=utf8mb4", + ] + if self.extra_args: + args.extend(shlex.split(self.extra_args)) + args.append(self.database) + + log.info("Executing mysqldump for db=%s host=%s", self.database, self.host) + + # mysqldump 直接写 stdout，我们 pipe 给 gzip 再 yield + # 用 -p 紧贴密码避免出现在 ps 里（空密码时 mysqldump 会要求交互输入） + if self.password: + args_with_pwd = args.copy() + # 找到 -u 后插入 -pXXX + u_idx = args_with_pwd.index("-u") + args_with_pwd.insert(u_idx + 2, f"-p{self.password}") + else: + args_with_pwd = args + + proc = await asyncio.create_subprocess_exec( + *args_with_pwd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + + assert proc.stdout is not None + + # 在线程里跑 gzip.compress，因为它是阻塞的 + loop = asyncio.get_event_loop() + + async def gzip_stream() -> AsyncIterator[bytes]: + chunk_buf = bytearray() + async for chunk in proc.stdout: # type: ignore[union-attr] + # gzip 接受 bytes，返回 bytes + compressed = await loop.run_in_executor(None, gzip.compress, chunk) + if compressed: + yield compressed + + try: + async for piece in gzip_stream(): + yield piece + finally: + try: + await proc.wait() + except Exception: + proc.kill() + await proc.wait() + + if proc.returncode != 0: + stderr = (await proc.stderr.read()).decode(errors="replace") if proc.stderr else "" + raise RuntimeError( + f"mysqldump failed (exit={proc.returncode}): {stderr.strip()[:2000]}" + ) diff --git a/backend/app/core/executor.py b/backend/app/core/executor.py new file mode 100644 index 0000000..62d9cb5 --- /dev/null +++ b/backend/app/core/executor.py @@ -0,0 +1,151 @@ +"""备份执行编排：解密配置 → 流式产出 → 流式上传 → 应用保留策略 → 写日志。""" +import asyncio +import hashlib +import io +import logging +import os +import tempfile +import time +from datetime import datetime +from pathlib import Path +from typing import Optional + +from sqlalchemy.orm import Session + +from app.core.backup.base import BaseBackup +from app.core.backup.directory import DirectoryBackup +from app.core.backup.mysql import MySQLBackup +from app.core.retention import apply_retention +from app.core.storage.factory import create_storage +from app.database import SessionLocal +from app.models.job import Job +from app.models.run import Run, RunLog +from app.utils.crypto import decrypt_dict +from app.utils.logging import get_logger + +log = get_logger(__name__) + + +def _make_backup(job: Job) -> BaseBackup: + src = decrypt_dict(job.source_config_json) + if job.type == "mysql": + return MySQLBackup(src, job.name) + if job.type == "directory": + return DirectoryBackup(src, job.name) + raise ValueError(f"Unsupported job type: {job.type}") + + +async def run_backup(run_id: int, trigger: str = "manual") -> None: + """执行一次备份。所有副作用都在自己 new 的 Session 上完成。""" + started = datetime.utcnow() + db: Session = SessionLocal() + try: + run: Run | None = db.get(Run, run_id) + if not run: + log.error("run_id=%s not found", run_id) + return + + job: Job | None = db.get(Job, run.job_id) + if not job: + run.status = "failed" + run.error_message = f"Job id={run.job_id} disappeared" + run.finished_at = datetime.utcnow() + db.commit() + return + + log_buffer = io.StringIO() + tee = _TeeLogger(log, log_buffer) + + run.status = "running" + run.trigger = trigger + run.started_at = started + db.commit() + + tee.info("Backup started: run_id=%s job=%s trigger=%s", run_id, job.name, trigger) + + try: + backup = _make_backup(job) + meta = backup.metadata() + tee.info("Metadata: filename=%s content_type=%s", meta.suggested_filename, meta.content_type) + + storage = create_storage(job.storage) + ok, msg = await storage.test_connection() + tee.info("Storage test: ok=%s msg=%s", ok, msg) + if not ok: + raise RuntimeError(f"Storage not reachable: {msg}") + + # 关键路径：备份文件名以 {job_name}/{YYYY-MM-DD}/{filename} 组织 + date_prefix = datetime.utcnow().strftime("%Y-%m-%d") + object_key = f"{job.name}/{date_prefix}/{meta.suggested_filename}" + + tee.info("Streaming backup to storage key=%s ...", object_key) + + async def _consume_with_progress(): + async for chunk in backup.produce(): + yield chunk + + t0 = time.monotonic() + size, sha = await storage.upload_stream( + _consume_with_progress(), + object_key, + content_type=meta.content_type, + ) + elapsed = int(time.monotonic() - t0) + + run.artifact_path = object_key + run.artifact_size = size + run.artifact_checksum = sha + run.duration_seconds = elapsed + run.status = "success" + run.finished_at = datetime.utcnow() + tee.info( + "Backup success: size=%s bytes sha256=%s elapsed=%ss", + size, sha[:12], elapsed, + ) + + # 应用保留策略 + try: + removed = apply_retention(db, job) + if removed: + tee.info("Retention cleanup removed %s old runs", removed) + except Exception as e: + tee.warning("Retention cleanup failed: %s", e) + + except Exception as e: + log.exception("Backup run_id=%s failed", run_id) + tee.error("Backup failed: %s: %s", type(e).__name__, e) + run.status = "failed" + run.error_message = f"{type(e).__name__}: {e}"[:2000] + run.finished_at = datetime.utcnow() + finally: + log_text = log_buffer.getvalue() + run.log_excerpt = log_text[-4000:] if len(log_text) > 4000 else log_text + run_log = RunLog(run_id=run.id, content=log_text) + db.add(run_log) + db.commit() + log.info("Backup run_id=%s finished status=%s", run_id, run.status) + finally: + db.close() + + +class _TeeLogger: + """同时写标准 logging 和内存 StringIO（用于存数据库）。""" + + def __init__(self, real: logging.Logger, buf: io.StringIO): + self._real = real + self._buf = buf + + def _emit(self, level: int, msg: str) -> None: + ts = datetime.utcnow().strftime("%H:%M:%S") + line = f"[{ts}] {msg}" + self._real.log(level, line) + self._buf.write(line + "\n") + + def info(self, msg: str, *args) -> None: + self._emit(logging.INFO, msg % args if args else msg) + + def warning(self, msg: str, *args) -> None: + self._emit(logging.WARNING, msg % args if args else msg) + + def error(self, msg: str, *args) -> None: + self._emit(logging.ERROR, msg % args if args else msg) diff --git a/backend/app/core/restore.py b/backend/app/core/restore.py new file mode 100644 index 0000000..da9f0cf --- /dev/null +++ b/backend/app/core/restore.py @@ -0,0 +1,165 @@ +"""恢复执行器：把 backup artifact 还原到目标（MySQL 库 / 目录路径）。""" +import asyncio +import gzip +import io +import os +import shutil +import tarfile +import tempfile +from datetime import datetime +from pathlib import Path +from typing import Optional + +from app.core.storage.factory import create_storage +from app.database import SessionLocal +from app.models.job import Job +from app.models.restore import RestoreHistory +from app.models.run import Run, RunLog +from app.utils.logging import get_logger + +log = get_logger(__name__) + + +async def execute_restore(restore_id: int) -> None: + """执行一次恢复。""" + db = SessionLocal() + try: + rh: RestoreHistory | None = db.get(RestoreHistory, restore_id) + if not rh: + return + + rh.status = "running" + rh.started_at = datetime.utcnow() + db.commit() + + run: Run | None = db.get(Run, rh.run_id) + if not run or not run.artifact_path: + raise RuntimeError("源 run 或 artifact 不存在") + job: Job | None = db.get(Job, run.job_id) + if not job: + raise RuntimeError("源 job 不存在") + + storage = create_storage(job.storage) + log_text = io.StringIO() + + def logl(msg: str) -> None: + line = f"[{datetime.utcnow():%H:%M:%S}] {msg}" + log.info(line) + log_text.write(line + "\n") + + try: + logl(f"Restoring run_id={run.id} -> {rh.target} (type={job.type})") + + # 先下载到临时文件 + tmpdir = Path(tempfile.gettempdir()) / f"restore_{rh.id}_{os.getpid()}" + tmpdir.mkdir(parents=True, exist_ok=True) + local_tmp = tmpdir / Path(run.artifact_path).name + + logl(f"Downloading {run.artifact_path} -> {local_tmp}") + size = 0 + with open(local_tmp, "wb") as f: + async for chunk in storage.download_stream(run.artifact_path): + f.write(chunk) + size += len(chunk) + logl(f"Downloaded {size} bytes") + + if job.type == "mysql": + await _restore_mysql(local_tmp, rh.target, job, logl) + elif job.type == "directory": + _restore_directory(local_tmp, rh.target, logl) + else: + raise RuntimeError(f"Unsupported job type: {job.type}") + + rh.status = "success" + rh.finished_at = datetime.utcnow() + logl("Restore success") + except Exception as e: + log.exception("Restore failed") + logl(f"Restore failed: {type(e).__name__}: {e}") + rh.status = "failed" + rh.error_message = f"{type(e).__name__}: {e}"[:2000] + rh.finished_at = datetime.utcnow() + finally: + # 写日志 + content = log_text.getvalue() + db.add(RunLog(run_id=run.id, content=f"[restore #{rh.id}]\n{content}")) + db.commit() + try: + shutil.rmtree(tmpdir, ignore_errors=True) + except Exception: + pass + finally: + db.close() + + +async def _restore_mysql(archive_path: Path, target_db: str, job: Job, logl) -> None: + """把 sql.gz 通过 gunzip | mysql 还原到 target_db。""" + from app.utils.crypto import decrypt_dict + + src = decrypt_dict(job.source_config_json) + host = src["host"] + port = int(src.get("port", 3306)) + user = src["user"] + password = src.get("password", "") + + logl(f"Decompressing and piping to mysql db={target_db} host={host}:{port}") + # 解压 + decompressed = archive_path.with_suffix("") # .sql + if archive_path.name.endswith(".gz"): + with gzip.open(archive_path, "rb") as gz, open(decompressed, "wb") as out: + shutil.copyfileobj(gz, out) + sql_file = decompressed + else: + sql_file = archive_path + + # 调用 mysql client + args = ["mysql", "-h", host, "-P", str(port), "-u", user, target_db] + if password: + # 在 -u 后插入 -pPWD + idx = args.index("-u") + args.insert(idx + 2, f"-p{password}") + + proc = await asyncio.create_subprocess_exec( + *args, + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + assert proc.stdin is not None + with open(sql_file, "rb") as f: + while True: + chunk = f.read(64 * 1024) + if not chunk: + break + proc.stdin.write(chunk) + await proc.stdin.drain() + proc.stdin.close() + rc = await proc.wait() + if rc != 0: + stderr = (await proc.stderr.read()).decode(errors="replace") if proc.stderr else "" + raise RuntimeError(f"mysql restore failed (exit={rc}): {stderr[:2000]}") + try: + decompressed.unlink(missing_ok=True) + except Exception: + pass + + +def _restore_directory(archive_path: Path, target_dir: str, logl) -> None: + """解压 tar.gz 到 target_dir（必须为空或不存在）。""" + target = Path(target_dir).resolve() + if target.exists() and any(target.iterdir()): + raise RuntimeError(f"目标目录非空: {target}") + target.mkdir(parents=True, exist_ok=True) + + logl(f"Extracting tar.gz to {target}") + if archive_path.name.endswith(".gz"): + mode = "r:gz" + else: + mode = "r:" + with tarfile.open(str(archive_path), mode) as tar: + # 安全检查：拒绝路径逃逸 + for member in tar.getmembers(): + member_path = (target / member.name).resolve() + if not str(member_path).startswith(str(target)): + raise RuntimeError(f"Refusing to extract unsafe path: {member.name}") + tar.extractall(path=str(target)) diff --git a/backend/app/core/retention.py b/backend/app/core/retention.py new file mode 100644 index 0000000..d191d3d --- /dev/null +++ b/backend/app/core/retention.py @@ -0,0 +1,80 @@ +"""备份保留策略清理。""" +from datetime import datetime, timedelta +from typing import Optional + +from sqlalchemy.orm import Session + +from app.core.storage.factory import create_storage +from app.models.job import Job +from app.models.run import Run +from app.utils.logging import get_logger + +log = get_logger(__name__) + + +def apply_retention(db: Session, job: Job) -> int: + """根据 job 的 retention_count / retention_days 清理旧备份。 + + 返回删除的 run 数。 + """ + # 取该 job 所有成功 runs，按 finished_at 倒序 + runs = ( + db.query(Run) + .filter(Run.job_id == job.id, Run.status == "success") + .order_by(Run.finished_at.desc()) + .all() + ) + if not runs: + return 0 + + keep_count = job.retention_count or 7 + keep_days = job.retention_days + + to_keep_ids: set[int] = set() + for i, r in enumerate(runs): + keep_by_count = i < keep_count + keep_by_days = True + if keep_days and r.finished_at: + threshold = datetime.utcnow() - timedelta(days=keep_days) + if r.finished_at < threshold: + keep_by_days = False + if keep_by_count or keep_by_days: + to_keep_ids.add(r.id) + + to_delete = [r for r in runs if r.id not in to_keep_ids] + if not to_delete: + return 0 + + storage = create_storage(job.storage) + removed = 0 + for r in to_delete: + try: + if r.artifact_path: + # 用同步方法简单调用即可（local 是同步包装 async，但这里 aioboto3/aiofiles 都是异步； + # 实际上 base storage 都是 async 方法，所以这里需要在同步上下文跑 event loop） + pass + except Exception: + log.exception("Failed to delete artifact %s (db row still kept)", r.artifact_path) + db.delete(r) + removed += 1 + + db.commit() + log.info("Retention cleanup for job=%s: deleted %s runs", job.name, removed) + + # 真实删除 S3/local 文件（需要在 async 上下文） + import asyncio + try: + loop = asyncio.new_event_loop() + try: + for r in to_delete: + if r.artifact_path: + try: + loop.run_until_complete(storage.delete(r.artifact_path)) + except Exception: + log.exception("Artifact delete failed: %s", r.artifact_path) + finally: + loop.close() + except Exception: + log.exception("Artifact cleanup loop error") + + return removed diff --git a/backend/app/core/storage/__init__.py b/backend/app/core/storage/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/app/core/storage/base.py b/backend/app/core/storage/base.py new file mode 100644 index 0000000..add6d86 --- /dev/null +++ b/backend/app/core/storage/base.py @@ -0,0 +1,45 @@ +"""存储后端抽象接口。""" +from abc import ABC, abstractmethod +from dataclasses import dataclass +from typing import AsyncIterator, Optional + + +@dataclass +class StorageObject: + key: str + size: int + last_modified: Optional[str] = None # ISO 8601 + + +class BaseStorage(ABC): + """所有存储后端实现的统一接口。""" + + type: str = "base" + + def __init__(self, config: dict): + self.config = config + + @abstractmethod + async def test_connection(self) -> tuple[bool, str]: + """测试连通性。返回 (ok, message)。""" + + @abstractmethod + async def upload_stream( + self, + source: AsyncIterator[bytes], + key: str, + content_type: str = "application/octet-stream", + ) -> tuple[int, str]: + """流式上传。返回 (size_bytes, sha256_hex)。""" + + @abstractmethod + async def download_stream(self, key: str) -> AsyncIterator[bytes]: ... + + @abstractmethod + async def delete(self, key: str) -> None: ... + + @abstractmethod + async def exists(self, key: str) -> bool: ... + + @abstractmethod + async def list(self, prefix: str = "") -> list[StorageObject]: ... diff --git a/backend/app/core/storage/factory.py b/backend/app/core/storage/factory.py new file mode 100644 index 0000000..b4b0f03 --- /dev/null +++ b/backend/app/core/storage/factory.py @@ -0,0 +1,15 @@ +"""存储工厂：根据 storage 行创建对应实现。""" +from app.core.storage.base import BaseStorage +from app.core.storage.local import LocalStorage +from app.core.storage.s3 import S3Storage +from app.models.storage import Storage +from app.utils.crypto import decrypt_dict + + +def create_storage(row: Storage) -> BaseStorage: + config = decrypt_dict(row.config_json) + if row.type == "local": + return LocalStorage(config) + if row.type == "s3": + return S3Storage(config) + raise ValueError(f"Unsupported storage type: {row.type}") diff --git a/backend/app/core/storage/local.py b/backend/app/core/storage/local.py new file mode 100644 index 0000000..a48a9e5 --- /dev/null +++ b/backend/app/core/storage/local.py @@ -0,0 +1,98 @@ +"""本地文件系统存储。""" +import asyncio +import hashlib +import os +import shutil +from datetime import datetime +from pathlib import Path +from typing import AsyncIterator + +import aiofiles + +from app.core.storage.base import BaseStorage, StorageObject +from app.utils.logging import get_logger + +log = get_logger(__name__) + + +class LocalStorage(BaseStorage): + type = "local" + + def __init__(self, config: dict): + super().__init__(config) + self.root: str = os.path.abspath(config["path"]) + Path(self.root).mkdir(parents=True, exist_ok=True) + + def _resolve(self, key: str) -> Path: + # 防 path traversal：禁止 .. + if ".." in key.split("/"): + raise ValueError(f"Invalid key (path traversal): {key}") + return Path(self.root) / key + + async def test_connection(self) -> tuple[bool, str]: + try: + Path(self.root).mkdir(parents=True, exist_ok=True) + test_file = Path(self.root) / ".backup_system_test" + async with aiofiles.open(test_file, "w") as f: + await f.write("ok") + # aiofiles 不暴露 os 命名空间，用线程池跑同步删除 + await asyncio.to_thread(os.remove, test_file) + return True, f"OK: writable {self.root}" + except Exception as e: + return False, f"FAIL: {e}" + + async def upload_stream( + self, + source: AsyncIterator[bytes], + key: str, + content_type: str = "application/octet-stream", + ) -> tuple[int, str]: + path = self._resolve(key) + path.parent.mkdir(parents=True, exist_ok=True) + sha = hashlib.sha256() + size = 0 + async with aiofiles.open(path, "wb") as f: + async for chunk in source: + await f.write(chunk) + sha.update(chunk) + size += len(chunk) + return size, sha.hexdigest() + + async def download_stream(self, key: str) -> AsyncIterator[bytes]: + path = self._resolve(key) + if not path.exists(): + raise FileNotFoundError(f"Not found: {key}") + async with aiofiles.open(path, "rb") as f: + while True: + chunk = await f.read(64 * 1024) + if not chunk: + break + yield chunk + + async def delete(self, key: str) -> None: + path = self._resolve(key) + try: + path.unlink() + except FileNotFoundError: + pass + + async def exists(self, key: str) -> bool: + return self._resolve(key).exists() + + async def list(self, prefix: str = "") -> list[StorageObject]: + base = Path(self.root) / prefix if prefix else Path(self.root) + if not base.exists(): + return [] + out: list[StorageObject] = [] + for p in base.rglob("*"): + if p.is_file(): + rel = p.relative_to(Path(self.root)).as_posix() + st = p.stat() + out.append( + StorageObject( + key=rel, + size=st.st_size, + last_modified=datetime.fromtimestamp(st.st_mtime).isoformat(), + ) + ) + return out diff --git a/backend/app/core/storage/s3.py b/backend/app/core/storage/s3.py new file mode 100644 index 0000000..420611f --- /dev/null +++ b/backend/app/core/storage/s3.py @@ -0,0 +1,176 @@ +"""S3 兼容对象存储（支持 MinIO / Ceph / AWS S3）。""" +import asyncio +import hashlib +from contextlib import asynccontextmanager +from datetime import datetime +from typing import Any, AsyncIterator + +import aioboto3 + +from app.core.storage.base import BaseStorage, StorageObject +from app.utils.logging import get_logger + +log = get_logger(__name__) + + +class S3Storage(BaseStorage): + type = "s3" + + def __init__(self, config: dict): + super().__init__(config) + self.endpoint_url: str | None = config.get("endpoint_url") or None + self.region: str = config.get("region") or "us-east-1" + self.bucket: str = config["bucket"] + self.access_key: str = config["access_key"] + self.secret_key: str = config["secret_key"] + self.use_ssl: bool = bool(config.get("use_ssl", True)) + self.path_prefix: str = config.get("path_prefix", "").strip("/") + self.addressing_style: str = config.get("addressing_style", "auto") + + @asynccontextmanager + async def _client(self): + session = aioboto3.Session() + async with session.client( + "s3", + endpoint_url=self.endpoint_url, + region_name=self.region, + aws_access_key_id=self.access_key, + aws_secret_access_key=self.secret_key, + use_ssl=self.use_ssl, + config=__import__("botocore.config", fromlist=["Config"]).Config( + signature_version="s3v4", + s3={"addressing_style": self.addressing_style}, + ), + ) as client: + yield client + + def _full_key(self, key: str) -> str: + key = key.lstrip("/") + if self.path_prefix: + return f"{self.path_prefix}/{key}" + return key + + async def test_connection(self) -> tuple[bool, str]: + try: + async with self._client() as client: + await client.head_bucket(Bucket=self.bucket) + return True, f"OK: bucket={self.bucket}" + except Exception as e: + return False, f"FAIL: {type(e).__name__}: {e}" + + async def upload_stream( + self, + source: AsyncIterator[bytes], + key: str, + content_type: str = "application/octet-stream", + ) -> tuple[int, str]: + sha = hashlib.sha256() + size = 0 + full_key = self._full_key(key) + + async with self._client() as client: + # 用 UploadPart + 内存缓冲整个对象再 put_object（对中小备份足够） + # 大文件可改为 multipart；这里用 MultipartUploader 模式，兼顾大文件 + from io import BytesIO + + buffer = BytesIO() + async for chunk in source: + buffer.write(chunk) + sha.update(chunk) + size += len(chunk) + + # 5MB 阈值：超过用 multipart + PART_SIZE = 5 * 1024 * 1024 + if size <= PART_SIZE: + buffer.seek(0) + await client.put_object( + Bucket=self.bucket, + Key=full_key, + Body=buffer.getvalue(), + ContentType=content_type, + ) + else: + # multipart + mpu = await client.create_multipart_upload( + Bucket=self.bucket, Key=full_key, ContentType=content_type + ) + parts: list[dict[str, Any]] = [] + try: + buffer.seek(0) + part_number = 1 + while True: + data = buffer.read(PART_SIZE) + if not data: + break + resp = await client.upload_part( + Bucket=self.bucket, + Key=full_key, + PartNumber=part_number, + UploadId=mpu["UploadId"], + Body=data, + ) + parts.append({"PartNumber": part_number, "ETag": resp["ETag"]}) + part_number += 1 + await client.complete_multipart_upload( + Bucket=self.bucket, + Key=full_key, + UploadId=mpu["UploadId"], + MultipartUpload={"Parts": parts}, + ) + except Exception: + await client.abort_multipart_upload( + Bucket=self.bucket, Key=full_key, UploadId=mpu["UploadId"] + ) + raise + + return size, sha.hexdigest() + + async def download_stream(self, key: str) -> AsyncIterator[bytes]: + full_key = self._full_key(key) + async with self._client() as client: + obj = await client.get_object(Bucket=self.bucket, Key=full_key) + body = obj["Body"] + try: + while True: + chunk = await asyncio.get_event_loop().run_in_executor( + None, body.read, 64 * 1024 + ) + if not chunk: + break + yield chunk + finally: + body.close() + + async def delete(self, key: str) -> None: + full_key = self._full_key(key) + async with self._client() as client: + await client.delete_object(Bucket=self.bucket, Key=full_key) + + async def exists(self, key: str) -> bool: + full_key = self._full_key(key) + async with self._client() as client: + try: + await client.head_object(Bucket=self.bucket, Key=full_key) + return True + except Exception: + return False + + async def list(self, prefix: str = "") -> list[StorageObject]: + full_prefix = self._full_key(prefix) + out: list[StorageObject] = [] + async with self._client() as client: + paginator = client.get_paginator("list_objects_v2") + async for page in paginator.paginate(Bucket=self.bucket, Prefix=full_prefix): + for obj in page.get("Contents", []): + key = obj["Key"] + if self.path_prefix and key.startswith(self.path_prefix + "/"): + key = key[len(self.path_prefix) + 1 :] + last_modified = obj.get("LastModified") + out.append( + StorageObject( + key=key, + size=int(obj.get("Size", 0)), + last_modified=last_modified.isoformat() if last_modified else None, + ) + ) + return out diff --git a/backend/app/database.py b/backend/app/database.py new file mode 100644 index 0000000..fc5c28d --- /dev/null +++ b/backend/app/database.py @@ -0,0 +1,59 @@ +"""SQLAlchemy 引擎与 Session 工厂。""" +from contextlib import contextmanager +from typing import Iterator + +from sqlalchemy import create_engine +from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker + +from app.config import get_settings + +settings = get_settings() + +# SQLite 需要 check_same_thread=False（虽然我们用 FastAPI async/threadpool，但仍建议） +connect_args = {} +if settings.DATABASE_URL.startswith("sqlite"): + connect_args["check_same_thread"] = False + +engine = create_engine( + settings.DATABASE_URL, + connect_args=connect_args, + pool_pre_ping=True, + future=True, +) + +SessionLocal = sessionmaker( + bind=engine, + autoflush=False, + autocommit=False, + expire_on_commit=False, + class_=Session, +) + + +class Base(DeclarativeBase): + """所有 ORM 模型的基类。""" + + pass + + +def get_db() -> Iterator[Session]: + """FastAPI 依赖：每次请求一个 Session。""" + db = SessionLocal() + try: + yield db + finally: + db.close() + + +@contextmanager +def session_scope() -> Iterator[Session]: + """普通上下文管理器版本，供非请求路径使用（如初始化、调度器）。""" + db = SessionLocal() + try: + yield db + db.commit() + except Exception: + db.rollback() + raise + finally: + db.close() diff --git a/backend/app/deps.py b/backend/app/deps.py new file mode 100644 index 0000000..644a029 --- /dev/null +++ b/backend/app/deps.py @@ -0,0 +1,52 @@ +"""FastAPI 依赖注入：当前用户、DB Session 等。""" +from typing import Annotated + +from fastapi import Depends, HTTPException, status +from fastapi.security import OAuth2PasswordBearer +from jose import JWTError +from sqlalchemy.orm import Session + +from app.database import get_db +from app.models.user import User +from app.security import decode_token + +oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login", auto_error=False) + +DBSession = Annotated[Session, Depends(get_db)] +TokenDep = Annotated[str | None, Depends(oauth2_scheme)] + + +def get_current_user( + token: TokenDep, + db: DBSession, +) -> User: + if not token: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Not authenticated", + headers={"WWW-Authenticate": "Bearer"}, + ) + try: + payload = decode_token(token) + username: str | None = payload.get("sub") + if not username: + raise HTTPException(status_code=401, detail="Invalid token") + except JWTError: + raise HTTPException(status_code=401, detail="Invalid or expired token") + + user = db.query(User).filter(User.username == username).first() + if not user or not user.is_active: + raise HTTPException(status_code=401, detail="User not found or inactive") + return user + + +CurrentUser = Annotated[User, Depends(get_current_user)] + + +def require_admin(user: CurrentUser) -> User: + if not user.is_admin: + raise HTTPException(status_code=403, detail="Admin privilege required") + return user + + +AdminUser = Annotated[User, Depends(require_admin)] diff --git a/backend/app/main.py b/backend/app/main.py new file mode 100644 index 0000000..ea1f8ae --- /dev/null +++ b/backend/app/main.py @@ -0,0 +1,99 @@ +"""FastAPI 应用入口：lifespan 管理 scheduler、首次启动创建 admin。""" +import logging +from contextlib import asynccontextmanager + +from fastapi import FastAPI +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import JSONResponse +from sqlalchemy.exc import OperationalError + +from app.api.router import api_router +from app.config import get_settings +from app.database import Base, SessionLocal, engine +from app.models.user import User +from app.scheduler.scheduler import scheduler_lifespan +from app.security import hash_password +from app.utils.logging import setup_logging + +setup_logging() +log = logging.getLogger(__name__) +settings = get_settings() + + +def init_database() -> None: + """建表（Alembic 是首选；这里兜底确保首次启动也能跑）。""" + try: + # 实际生产用 alembic upgrade head；这里仅当表不存在时创建 + Base.metadata.create_all(bind=engine) + log.info("Database tables ensured") + except OperationalError as e: + log.error("Database init failed: %s", e) + raise + + +def init_admin() -> None: + """首次启动创建 admin 账号（仅当 users 表为空时）。""" + with SessionLocal() as db: + try: + existing = db.query(User).count() + if existing > 0: + log.info("Existing users found (%d), skip admin init", existing) + return + admin = User( + username="admin", + password_hash=hash_password(settings.INITIAL_ADMIN_PASSWORD), + is_admin=True, + is_active=True, + ) + db.add(admin) + db.commit() + log.warning( + "Initial admin account created. username=admin " + "Please change the password after first login." + ) + except Exception: + db.rollback() + raise + + +@asynccontextmanager +async def lifespan(app: FastAPI): + init_database() + init_admin() + async with scheduler_lifespan(): + log.info("Application ready") + yield + log.info("Application shutting down") + + +app = FastAPI( + title="Backup System", + description="数据备份管理系统 - MySQL / 目录 备份到本地 / S3", + version="0.1.0", + lifespan=lifespan, +) + +app.add_middleware( + CORSMiddleware, + allow_origins=settings.cors_origins_list, + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + + +@app.get("/api/health") +def health() -> dict: + return {"status": "ok"} + + +app.include_router(api_router, prefix="/api") + + +@app.exception_handler(Exception) +async def global_exception_handler(request, exc): + log.exception("Unhandled error on %s %s", request.method, request.url) + return JSONResponse( + status_code=500, + content={"detail": f"Internal server error: {type(exc).__name__}"}, + ) diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py new file mode 100644 index 0000000..39cf693 --- /dev/null +++ b/backend/app/models/__init__.py @@ -0,0 +1,7 @@ +"""ORM 模型集中导入，确保 Base.metadata 能识别所有表。""" +from app.models.user import User # noqa: F401 +from app.models.storage import Storage # noqa: F401 +from app.models.job import Job # noqa: F401 +from app.models.run import Run, RunLog # noqa: F401 +from app.models.setting import Setting # noqa: F401 +from app.models.restore import RestoreHistory # noqa: F401 diff --git a/backend/app/models/job.py b/backend/app/models/job.py new file mode 100644 index 0000000..e9091f3 --- /dev/null +++ b/backend/app/models/job.py @@ -0,0 +1,39 @@ +from datetime import datetime + +from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, Text, func +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from app.database import Base + + +class Job(Base): + """备份任务定义。 + + source_config_json 是 Fernet 加密的 JSON： + - mysql: {"host", "port", "user", "password", "database"} + - directory: {"path", "exclude_patterns": [...]} + """ + + __tablename__ = "jobs" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + name: Mapped[str] = mapped_column(String(64), unique=True, nullable=False) + type: Mapped[str] = mapped_column(String(16), nullable=False) # 'mysql' / 'directory' + source_config_json: Mapped[str] = mapped_column(Text, nullable=False) + storage_id: Mapped[int] = mapped_column( + Integer, ForeignKey("storages.id", ondelete="RESTRICT"), nullable=False + ) + cron_expression: Mapped[str | None] = mapped_column(String(64), nullable=True) + enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True) + retention_count: Mapped[int] = mapped_column(Integer, nullable=False, default=7) + retention_days: Mapped[int | None] = mapped_column(Integer, nullable=True) + description: Mapped[str | None] = mapped_column(String(255), nullable=True) + + storage: Mapped["Storage"] = relationship(lazy="joined") # noqa: F821 + + created_at: Mapped[datetime] = mapped_column( + DateTime, server_default=func.now(), nullable=False + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime, server_default=func.now(), onupdate=func.now(), nullable=False + ) diff --git a/backend/app/models/restore.py b/backend/app/models/restore.py new file mode 100644 index 0000000..3ebeef2 --- /dev/null +++ b/backend/app/models/restore.py @@ -0,0 +1,25 @@ +from datetime import datetime + +from sqlalchemy import DateTime, ForeignKey, Integer, String, Text, func +from sqlalchemy.orm import Mapped, mapped_column + +from app.database import Base + + +class RestoreHistory(Base): + """恢复执行历史。""" + + __tablename__ = "restore_history" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + run_id: Mapped[int] = mapped_column( + Integer, ForeignKey("runs.id", ondelete="CASCADE"), nullable=False, index=True + ) + target: Mapped[str] = mapped_column(String(512), nullable=False) + status: Mapped[str] = mapped_column(String(16), nullable=False, default="pending") + started_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + finished_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + error_message: Mapped[str | None] = mapped_column(Text, nullable=True) + created_at: Mapped[datetime] = mapped_column( + DateTime, server_default=func.now(), nullable=False + ) diff --git a/backend/app/models/run.py b/backend/app/models/run.py new file mode 100644 index 0000000..2b946f5 --- /dev/null +++ b/backend/app/models/run.py @@ -0,0 +1,61 @@ +from datetime import datetime + +from sqlalchemy import ( + BigInteger, + DateTime, + ForeignKey, + Integer, + String, + Text, + func, +) +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from app.database import Base + + +class Run(Base): + """单次备份执行记录。""" + + __tablename__ = "runs" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + job_id: Mapped[int] = mapped_column( + Integer, ForeignKey("jobs.id", ondelete="CASCADE"), nullable=False, index=True + ) + status: Mapped[str] = mapped_column(String(16), nullable=False, default="pending") + trigger: Mapped[str] = mapped_column(String(16), nullable=False, default="manual") + started_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + finished_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + duration_seconds: Mapped[int | None] = mapped_column(Integer, nullable=True) + artifact_path: Mapped[str | None] = mapped_column(String(512), nullable=True) + artifact_size: Mapped[int | None] = mapped_column(BigInteger, nullable=True) + artifact_checksum: Mapped[str | None] = mapped_column(String(64), nullable=True) + error_message: Mapped[str | None] = mapped_column(Text, nullable=True) + log_excerpt: Mapped[str | None] = mapped_column(Text, nullable=True) + + job: Mapped["Job"] = relationship(lazy="joined") # noqa: F821 + logs: Mapped[list["RunLog"]] = relationship( # noqa: F821 + back_populates="run", cascade="all, delete-orphan", lazy="select" + ) + + created_at: Mapped[datetime] = mapped_column( + DateTime, server_default=func.now(), nullable=False + ) + + +class RunLog(Base): + """完整日志（可独立表以避免 runs 表臃肿）。""" + + __tablename__ = "run_logs" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + run_id: Mapped[int] = mapped_column( + Integer, ForeignKey("runs.id", ondelete="CASCADE"), nullable=False, index=True + ) + content: Mapped[str] = mapped_column(Text, nullable=False) + created_at: Mapped[datetime] = mapped_column( + DateTime, server_default=func.now(), nullable=False + ) + + run: Mapped["Run"] = relationship(back_populates="logs") diff --git a/backend/app/models/setting.py b/backend/app/models/setting.py new file mode 100644 index 0000000..45bf721 --- /dev/null +++ b/backend/app/models/setting.py @@ -0,0 +1,18 @@ +from datetime import datetime + +from sqlalchemy import DateTime, String, Text, func +from sqlalchemy.orm import Mapped, mapped_column + +from app.database import Base + + +class Setting(Base): + """系统级 KV 设置。""" + + __tablename__ = "settings" + + key: Mapped[str] = mapped_column(String(64), primary_key=True) + value: Mapped[str] = mapped_column(Text, nullable=False, default="") + updated_at: Mapped[datetime] = mapped_column( + DateTime, server_default=func.now(), onupdate=func.now(), nullable=False + ) diff --git a/backend/app/models/storage.py b/backend/app/models/storage.py new file mode 100644 index 0000000..4b27120 --- /dev/null +++ b/backend/app/models/storage.py @@ -0,0 +1,31 @@ +from datetime import datetime + +from sqlalchemy import Boolean, DateTime, Integer, String, Text, func +from sqlalchemy.orm import Mapped, mapped_column + +from app.database import Base + + +class Storage(Base): + """备份目标存储（local 路径 或 S3 桶）。 + + config_json 是 Fernet 加密的 JSON 字符串，结构因 type 而异： + - local: {"path": "/data/backups"} + - s3: {"endpoint_url": "...", "region": "...", "bucket": "...", + "access_key": "...", "secret_key": "...", "use_ssl": true, + "path_prefix": "optional/prefix"} + """ + + __tablename__ = "storages" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + name: Mapped[str] = mapped_column(String(64), unique=True, nullable=False) + type: Mapped[str] = mapped_column(String(16), nullable=False) # 'local' / 's3' + config_json: Mapped[str] = mapped_column(Text, nullable=False) + is_default: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + created_at: Mapped[datetime] = mapped_column( + DateTime, server_default=func.now(), nullable=False + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime, server_default=func.now(), onupdate=func.now(), nullable=False + ) diff --git a/backend/app/models/user.py b/backend/app/models/user.py new file mode 100644 index 0000000..5c1f469 --- /dev/null +++ b/backend/app/models/user.py @@ -0,0 +1,22 @@ +from datetime import datetime + +from sqlalchemy import Boolean, DateTime, Integer, String, func +from sqlalchemy.orm import Mapped, mapped_column + +from app.database import Base + + +class User(Base): + __tablename__ = "users" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + username: Mapped[str] = mapped_column(String(64), unique=True, nullable=False, index=True) + password_hash: Mapped[str] = mapped_column(String(255), nullable=False) + is_admin: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True) + created_at: Mapped[datetime] = mapped_column( + DateTime, server_default=func.now(), nullable=False + ) + + def __repr__(self) -> str: + return f"" diff --git a/backend/app/scheduler/__init__.py b/backend/app/scheduler/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/app/scheduler/scheduler.py b/backend/app/scheduler/scheduler.py new file mode 100644 index 0000000..679e5e2 --- /dev/null +++ b/backend/app/scheduler/scheduler.py @@ -0,0 +1,131 @@ +"""APScheduler 调度器：FastAPI lifespan 中启动/关闭。""" +import asyncio +import logging +from contextlib import asynccontextmanager +from datetime import datetime +from typing import Optional + +from apscheduler.schedulers.asyncio import AsyncIOScheduler +from apscheduler.triggers.cron import CronTrigger + +from app.config import get_settings +from app.database import SessionLocal +from app.models.job import Job +from app.models.run import Run +from app.utils.logging import get_logger + +log = get_logger(__name__) +settings = get_settings() + +_scheduler: Optional[AsyncIOScheduler] = None + + +async def _execute_scheduled_job(job_id: int) -> None: + """调度器回调：创建 run 记录并启动执行。""" + db = SessionLocal() + try: + job = db.get(Job, job_id) + if not job or not job.enabled: + return + run = Run( + job_id=job.id, + status="pending", + trigger="scheduled", + started_at=datetime.utcnow(), + ) + db.add(run) + db.commit() + db.refresh(run) + run_id = run.id + finally: + db.close() + + from app.core.executor import run_backup + + asyncio.create_task(run_backup(run_id, trigger="scheduled")) + + +def _register_job(job: Job) -> Optional[str]: + """注册一个 job 的 cron trigger，返回 APScheduler job id。""" + assert _scheduler is not None + if not job.enabled or not job.cron_expression: + return None + try: + trigger = CronTrigger.from_crontab(job.cron_expression, timezone=settings.TZ) + except Exception as e: + log.error("Invalid cron for job %s: %s", job.name, e) + return None + sched_job = _scheduler.add_job( + _execute_scheduled_job, + trigger=trigger, + args=[job.id], + id=f"job-{job.id}", + replace_existing=True, + misfire_grace_time=600, + coalesce=True, + ) + log.info("Registered scheduled job: id=%s name=%s cron=%s", job.id, job.name, job.cron_expression) + return sched_job.id + + +def _reload_all_jobs() -> None: + """从 DB 重新加载所有 enabled job。""" + assert _scheduler is not None + # 移除现有 + for sj in _scheduler.get_jobs(): + sj.remove() + db = SessionLocal() + try: + jobs = db.query(Job).filter(Job.enabled == True, Job.cron_expression.isnot(None)).all() # noqa: E712 + for j in jobs: + _register_job(j) + finally: + db.close() + + +@asynccontextmanager +async def scheduler_lifespan(): + global _scheduler + _scheduler = AsyncIOScheduler(timezone=settings.TZ) + _scheduler.start() + log.info("Scheduler started") + try: + _reload_all_jobs() + yield + finally: + _scheduler.shutdown(wait=False) + log.info("Scheduler stopped") + _scheduler = None + + +def reschedule_job(job_id: int) -> None: + """job 更新后调用此函数重新注册。""" + if _scheduler is None: + return + sched_id = f"job-{job_id}" + # 移除旧的 + try: + existing = _scheduler.get_job(sched_id) + if existing: + existing.remove() + except Exception: + pass + # 重新加载 + db = SessionLocal() + try: + job = db.get(Job, job_id) + if job: + _register_job(job) + finally: + db.close() + + +def remove_job(job_id: int) -> None: + if _scheduler is None: + return + try: + existing = _scheduler.get_job(f"job-{job_id}") + if existing: + existing.remove() + except Exception: + pass diff --git a/backend/app/schemas/__init__.py b/backend/app/schemas/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/app/schemas/auth.py b/backend/app/schemas/auth.py new file mode 100644 index 0000000..b5fe059 --- /dev/null +++ b/backend/app/schemas/auth.py @@ -0,0 +1,34 @@ +from datetime import datetime +from typing import Optional + +from pydantic import BaseModel, ConfigDict, Field + + +class LoginRequest(BaseModel): + username: str = Field(..., min_length=1, max_length=64) + password: str = Field(..., min_length=1) + + +class TokenResponse(BaseModel): + access_token: str + token_type: str = "bearer" + expires_in: int + + +class UserOut(BaseModel): + model_config = ConfigDict(from_attributes=True) + id: int + username: str + is_admin: bool + is_active: bool + created_at: datetime + + +class ChangePasswordRequest(BaseModel): + old_password: str = Field(..., min_length=1) + new_password: str = Field(..., min_length=8) + + +class MessageResponse(BaseModel): + message: str + detail: Optional[str] = None diff --git a/backend/app/schemas/job.py b/backend/app/schemas/job.py new file mode 100644 index 0000000..fc2f357 --- /dev/null +++ b/backend/app/schemas/job.py @@ -0,0 +1,96 @@ +from datetime import datetime +from typing import Literal, Optional + +from pydantic import BaseModel, ConfigDict, Field, field_validator + +from app.schemas.storage import StorageOut + + +class MySQLSourceConfig(BaseModel): + host: str = Field(..., min_length=1) + port: int = Field(3306, ge=1, le=65535) + user: str = Field(..., min_length=1) + password: str = Field("", description="可空（trust auth）") + database: str = Field(..., min_length=1, description="要备份的库名") + extra_args: str = Field("", description="额外 mysqldump 参数，如 '--skip-lock-tables'") + + +class DirectorySourceConfig(BaseModel): + path: str = Field(..., min_length=1, description="要备份的目录绝对路径") + exclude_patterns: list[str] = Field( + default_factory=lambda: [".DS_Store", "__pycache__", "node_modules", ".git"] + ) + + +class JobBase(BaseModel): + name: str = Field(..., min_length=1, max_length=64) + type: Literal["mysql", "directory"] + source_config: dict + storage_id: int + cron_expression: Optional[str] = Field(None, description="标准 5 段 cron 表达式，空表示手动") + enabled: bool = True + retention_count: int = Field(7, ge=1, le=365) + retention_days: Optional[int] = Field(None, ge=1, le=3650) + description: Optional[str] = Field(None, max_length=255) + + @field_validator("cron_expression") + @classmethod + def _check_cron(cls, v: Optional[str]) -> Optional[str]: + if v is None or v.strip() == "": + return None + v = v.strip() + parts = v.split() + if len(parts) != 5: + raise ValueError("cron 必须是 5 段：分 时 日 月 周") + return v + + +class JobCreate(JobBase): + pass + + +class JobUpdate(BaseModel): + name: Optional[str] = Field(None, min_length=1, max_length=64) + source_config: Optional[dict] = None + storage_id: Optional[int] = None + cron_expression: Optional[str] = None + enabled: Optional[bool] = None + retention_count: Optional[int] = Field(None, ge=1, le=365) + retention_days: Optional[int] = Field(None, ge=1, le=3650) + description: Optional[str] = None + + +class JobOut(BaseModel): + model_config = ConfigDict(from_attributes=True) + id: int + name: str + type: str + cron_expression: Optional[str] + enabled: bool + retention_count: int + retention_days: Optional[int] + description: Optional[str] + storage: StorageOut + created_at: datetime + updated_at: datetime + + +class RunSummary(BaseModel): + model_config = ConfigDict(from_attributes=True) + id: int + status: str + trigger: str + started_at: Optional[datetime] + finished_at: Optional[datetime] + duration_seconds: Optional[int] + artifact_size: Optional[int] + error_message: Optional[str] + + +class JobWithLastRun(JobOut): + last_run: Optional[RunSummary] = None + + +class JobRunResponse(BaseModel): + run_id: int + status: str diff --git a/backend/app/schemas/restore.py b/backend/app/schemas/restore.py new file mode 100644 index 0000000..32ecb04 --- /dev/null +++ b/backend/app/schemas/restore.py @@ -0,0 +1,33 @@ +from datetime import datetime +from typing import Literal, Optional + +from pydantic import BaseModel, ConfigDict, Field + + +class RestoreTargetsResponse(BaseModel): + """列出某个 run 可恢复的目标元信息。""" + + run_id: int + job_type: str # 'mysql' / 'directory' + artifact_size: Optional[int] + artifact_checksum: Optional[str] + suggested_targets: list[str] + + +class RestoreRequest(BaseModel): + run_id: int + target_type: Literal["mysql_db", "directory_path"] + target: str = Field(..., min_length=1, description="MySQL 库名 或 目录路径") + options: dict = Field(default_factory=dict, description="额外选项，如 {'drop_existing': true}") + + +class RestoreOut(BaseModel): + model_config = ConfigDict(from_attributes=True) + id: int + run_id: int + target: str + status: str + started_at: Optional[datetime] + finished_at: Optional[datetime] + error_message: Optional[str] + created_at: datetime diff --git a/backend/app/schemas/run.py b/backend/app/schemas/run.py new file mode 100644 index 0000000..3916a47 --- /dev/null +++ b/backend/app/schemas/run.py @@ -0,0 +1,32 @@ +from datetime import datetime +from typing import Optional + +from pydantic import BaseModel, ConfigDict + +from app.schemas.job import RunSummary + + +class RunOut(RunSummary): + model_config = ConfigDict(from_attributes=True) + job_id: int + artifact_path: Optional[str] + artifact_checksum: Optional[str] + log_excerpt: Optional[str] + created_at: datetime + + +class RunLogOut(BaseModel): + model_config = ConfigDict(from_attributes=True) + id: int + content: str + created_at: datetime + + +class RunListResponse(BaseModel): + total: int + items: list[RunOut] + + +class RunTriggerResponse(BaseModel): + run_id: int + status: str diff --git a/backend/app/schemas/setting.py b/backend/app/schemas/setting.py new file mode 100644 index 0000000..3929ce1 --- /dev/null +++ b/backend/app/schemas/setting.py @@ -0,0 +1,13 @@ +from typing import Optional + +from pydantic import BaseModel + + +class SettingOut(BaseModel): + key: str + value: str + + +class SettingUpdate(BaseModel): + key: str + value: Optional[str] = None diff --git a/backend/app/schemas/storage.py b/backend/app/schemas/storage.py new file mode 100644 index 0000000..0ab92b7 --- /dev/null +++ b/backend/app/schemas/storage.py @@ -0,0 +1,50 @@ +from datetime import datetime +from typing import Literal, Optional + +from pydantic import BaseModel, ConfigDict, Field + + +class LocalStorageConfig(BaseModel): + path: str = Field(..., min_length=1, description="本地存储根目录绝对路径") + + +class S3StorageConfig(BaseModel): + endpoint_url: Optional[str] = Field(None, description="MinIO/Ceph 自定义 endpoint，留空使用 AWS") + region: str = Field("us-east-1") + bucket: str = Field(..., min_length=1) + access_key: str = Field(..., min_length=1) + secret_key: str = Field(..., min_length=1) + use_ssl: bool = True + path_prefix: str = Field("", description="对象 key 前缀，可空") + addressing_style: Literal["auto", "path", "virtual"] = "auto" + + +class StorageBase(BaseModel): + name: str = Field(..., min_length=1, max_length=64) + type: Literal["local", "s3"] + is_default: bool = False + + +class StorageCreate(StorageBase): + config: dict + + +class StorageUpdate(BaseModel): + name: Optional[str] = Field(None, min_length=1, max_length=64) + is_default: Optional[bool] = None + config: Optional[dict] = None + + +class StorageOut(BaseModel): + model_config = ConfigDict(from_attributes=True) + id: int + name: str + type: str + is_default: bool + created_at: datetime + updated_at: datetime + + +class StorageTestResponse(BaseModel): + ok: bool + message: str diff --git a/backend/app/security.py b/backend/app/security.py new file mode 100644 index 0000000..5ec0808 --- /dev/null +++ b/backend/app/security.py @@ -0,0 +1,35 @@ +"""JWT 与密码哈希工具。""" +from datetime import datetime, timedelta, timezone +from typing import Any, Optional + +from jose import JWTError, jwt +from passlib.context import CryptContext + +from app.config import get_settings + +settings = get_settings() + +pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") + +ALGORITHM = "HS256" + + +def hash_password(password: str) -> str: + return pwd_context.hash(password) + + +def verify_password(plain: str, hashed: str) -> bool: + return pwd_context.verify(plain, hashed) + + +def create_access_token(subject: str, extra: Optional[dict[str, Any]] = None) -> str: + expire = datetime.now(timezone.utc) + timedelta(minutes=settings.JWT_EXPIRE_MINUTES) + payload: dict[str, Any] = {"sub": subject, "exp": expire} + if extra: + payload.update(extra) + return jwt.encode(payload, settings.SECRET_KEY, algorithm=ALGORITHM) + + +def decode_token(token: str) -> dict[str, Any]: + """解码 token，失败抛 JWTError。""" + return jwt.decode(token, settings.SECRET_KEY, algorithms=[ALGORITHM]) diff --git a/backend/app/utils/__init__.py b/backend/app/utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/app/utils/crypto.py b/backend/app/utils/crypto.py new file mode 100644 index 0000000..4c41cdb --- /dev/null +++ b/backend/app/utils/crypto.py @@ -0,0 +1,46 @@ +"""Fernet 对称加密工具：用于加密存储敏感字段（MySQL 密码、S3 SecretKey）。""" +from cryptography.fernet import Fernet, InvalidToken + +from app.config import get_settings + +_fernet: Fernet | None = None + + +def _get_fernet() -> Fernet: + global _fernet + if _fernet is None: + _fernet = Fernet(get_settings().FERNET_KEY.encode()) + return _fernet + + +def encrypt_str(plain: str) -> str: + """加密字符串，返回 base64-urlsafe 字符串（可安全存数据库 Text 字段）。""" + if plain is None: + return "" + return _get_fernet().encrypt(plain.encode()).decode() + + +def decrypt_str(token: str) -> str: + """解密字符串。空字符串返回空字符串。""" + if not token: + return "" + try: + return _get_fernet().decrypt(token.encode()).decode() + except InvalidToken as e: + raise ValueError(f"Failed to decrypt token (FERNET_KEY 可能已被更改): {e}") from e + + +def encrypt_dict(data: dict) -> str: + """加密字典：序列化为 JSON 再加密。""" + import json + + return encrypt_str(json.dumps(data, ensure_ascii=False)) + + +def decrypt_dict(token: str) -> dict: + """解密字典。""" + import json + + if not token: + return {} + return json.loads(decrypt_str(token)) diff --git a/backend/app/utils/logging.py b/backend/app/utils/logging.py new file mode 100644 index 0000000..69aa3c5 --- /dev/null +++ b/backend/app/utils/logging.py @@ -0,0 +1,50 @@ +"""日志配置：控制台 + 文件 + 单 run StringIO Tee。""" +import logging +import os +import sys +from datetime import datetime +from logging.handlers import RotatingFileHandler +from pathlib import Path + +from app.config import get_settings + +_settings = get_settings() + +_LOG_DIR = Path("logs") +_LOG_DIR.mkdir(parents=True, exist_ok=True) + + +def setup_logging() -> None: + """应用启动时调用一次。""" + root = logging.getLogger() + root.setLevel(_settings.LOG_LEVEL.upper()) + + # 避免重复添加 handler（uvicorn 重载时） + if root.handlers: + return + + fmt = logging.Formatter( + "%(asctime)s | %(levelname)-7s | %(name)s | %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + + # 控制台 + console = logging.StreamHandler(sys.stdout) + console.setFormatter(fmt) + root.addHandler(console) + + # 文件 + log_file = _LOG_DIR / f"app-{datetime.now():%Y-%m-%d}.log" + file_handler = RotatingFileHandler( + log_file, maxBytes=20 * 1024 * 1024, backupCount=10, encoding="utf-8" + ) + file_handler.setFormatter(fmt) + root.addHandler(file_handler) + + # 调整 uvicorn 噪音 + logging.getLogger("uvicorn.access").setLevel(logging.WARNING) + logging.getLogger("apscheduler.scheduler").setLevel(logging.INFO) + + +def get_logger(name: str) -> logging.Logger: + return logging.getLogger(name) diff --git a/backend/requirements.txt b/backend/requirements.txt new file mode 100644 index 0000000..dac3df7 --- /dev/null +++ b/backend/requirements.txt @@ -0,0 +1,15 @@ +fastapi==0.115.0 +uvicorn[standard]==0.30.6 +pydantic==2.9.2 +pydantic-settings==2.5.2 +sqlalchemy==2.0.35 +alembic==1.13.3 +python-jose[cryptography]==3.3.0 +passlib[bcrypt]==1.7.4 +bcrypt==4.0.1 +python-multipart==0.0.12 +apscheduler==3.10.4 +aiofiles==24.1.0 +aioboto3==13.1.1 +cryptography==43.0.1 +httpx==0.27.2 diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..0cdda81 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,47 @@ +version: "3.9" + +services: + backend: + build: + context: ./backend + dockerfile: Dockerfile + image: backup-system/backend:latest + container_name: backup-backend + env_file: + - .env + environment: + # 容器内默认存储根目录 + STORAGE_LOCAL_ROOT: /app/data/backups + DATABASE_URL: sqlite:////app/data/backup.db + volumes: + - backend-data:/app/data + - backend-logs:/app/logs + # 默认 local storage 根；若自定义 local storage 路径，可再挂载 + - local-backups:/app/data/backups + ports: + - "8765:8000" + restart: unless-stopped + healthcheck: + test: ["CMD", "curl", "-f", "http://localhost:8000/api/health"] + interval: 30s + timeout: 5s + retries: 5 + start_period: 20s + + frontend: + build: + context: ./frontend + dockerfile: Dockerfile + image: backup-system/frontend:latest + container_name: backup-frontend + ports: + - "5173:80" + depends_on: + backend: + condition: service_healthy + restart: unless-stopped + +volumes: + backend-data: + backend-logs: + local-backups: diff --git a/frontend/.env.example b/frontend/.env.example new file mode 100644 index 0000000..14ea4ad --- /dev/null +++ b/frontend/.env.example @@ -0,0 +1 @@ +VITE_API_BASE_URL=/api diff --git a/frontend/Dockerfile b/frontend/Dockerfile new file mode 100644 index 0000000..9726355 --- /dev/null +++ b/frontend/Dockerfile @@ -0,0 +1,18 @@ +FROM node:20-alpine AS builder + +WORKDIR /app + +COPY package.json ./ +RUN npm install --no-audit --no-fund + +COPY . . +RUN npm run build + +FROM nginx:1.27-alpine + +COPY nginx.conf /etc/nginx/conf.d/default.conf +COPY --from=builder /app/dist /usr/share/nginx/html + +EXPOSE 80 + +CMD ["nginx", "-g", "daemon off;"] diff --git a/frontend/index.html b/frontend/index.html new file mode 100644 index 0000000..3069ee4 --- /dev/null +++ b/frontend/index.html @@ -0,0 +1,13 @@ + + + + + + + 数据备份管理系统 + + +
+ + + diff --git a/frontend/nginx.conf b/frontend/nginx.conf new file mode 100644 index 0000000..d70243f --- /dev/null +++ b/frontend/nginx.conf @@ -0,0 +1,43 @@ +server { + listen 80; + server_name _; + + root /usr/share/nginx/html; + index index.html; + + # gzip + gzip on; + gzip_types text/plain text/css application/json application/javascript text/xml application/xml application/xml+rss text/javascript; + gzip_min_length 1024; + + # 前端 SPA：所有非 /api 路径回退到 index.html + location / { + try_files $uri $uri/ /index.html; + } + + # 反代到后端 + location /api/ { + proxy_pass http://backend:8000; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + + # 流式响应支持（大文件下载 / 日志） + proxy_buffering off; + proxy_request_buffering off; + proxy_read_timeout 3600s; + proxy_send_timeout 3600s; + } + + # 健康检查透传 + location /api/health { + proxy_pass http://backend:8000/api/health; + } + + # 缓存静态资源 + location ~* \.(js|css|png|jpg|jpeg|gif|ico|svg|woff|woff2|ttf)$ { + expires 1y; + add_header Cache-Control "public, immutable"; + } +} diff --git a/frontend/package.json b/frontend/package.json new file mode 100644 index 0000000..66719ce --- /dev/null +++ b/frontend/package.json @@ -0,0 +1,29 @@ +{ + "name": "backup-system-frontend", + "private": true, + "version": "0.1.0", + "type": "module", + "scripts": { + "dev": "vite", + "build": "tsc -b && vite build", + "preview": "vite preview", + "lint": "eslint . --ext ts,tsx --report-unused-disable-directives --max-warnings 0" + }, + "dependencies": { + "antd": "^5.21.0", + "@ant-design/icons": "^5.5.1", + "axios": "^1.7.7", + "dayjs": "^1.11.13", + "react": "^18.3.1", + "react-dom": "^18.3.1", + "react-router-dom": "^6.26.2", + "zustand": "^4.5.5" + }, + "devDependencies": { + "@types/react": "^18.3.5", + "@types/react-dom": "^18.3.0", + "@vitejs/plugin-react": "^4.3.1", + "typescript": "^5.5.4", + "vite": "^5.4.6" + } +} diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx new file mode 100644 index 0000000..de78d23 --- /dev/null +++ b/frontend/src/App.tsx @@ -0,0 +1,10 @@ +import { BrowserRouter } from 'react-router-dom' +import { AppRouter } from './router' + +export default function App() { + return ( + + + + ) +} diff --git a/frontend/src/api/auth.ts b/frontend/src/api/auth.ts new file mode 100644 index 0000000..f1a81fc --- /dev/null +++ b/frontend/src/api/auth.ts @@ -0,0 +1,17 @@ +import { api } from './client' +import type { MessageResponse, TokenResponse, User } from '../types' + +export interface LoginPayload { + username: string + password: string +} + +export const authApi = { + login: (payload: LoginPayload) => + api.post('/auth/login', payload).then((r) => r.data), + me: () => api.get('/auth/me').then((r) => r.data), + changePassword: (old_password: string, new_password: string) => + api + .post('/auth/change-password', { old_password, new_password }) + .then((r) => r.data), +} diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts new file mode 100644 index 0000000..3ef88a7 --- /dev/null +++ b/frontend/src/api/client.ts @@ -0,0 +1,45 @@ +import axios, { AxiosError, InternalAxiosRequestConfig } from 'axios' +import { message } from 'antd' +import { useAuthStore } from '../stores/authStore' + +const baseURL = import.meta.env.VITE_API_BASE_URL || '/api' + +export const api = axios.create({ + baseURL, + timeout: 60_000, +}) + +api.interceptors.request.use((config: InternalAxiosRequestConfig) => { + const token = useAuthStore.getState().token + if (token) { + config.headers.Authorization = `Bearer ${token}` + } + return config +}) + +api.interceptors.response.use( + (resp) => resp, + (error: AxiosError<{ detail?: string | { msg: string }[] }>) => { + const status = error.response?.status + const detail = error.response?.data?.detail + + let msg: string + if (Array.isArray(detail)) { + msg = detail.map((d) => d.msg).join('; ') + } else if (typeof detail === 'string') { + msg = detail + } else { + msg = error.message + } + + if (status === 401) { + useAuthStore.getState().logout() + if (window.location.pathname !== '/login') { + window.location.href = '/login' + } + } else if (status && status >= 400) { + message.error(msg) + } + return Promise.reject(error) + }, +) diff --git a/frontend/src/api/jobs.ts b/frontend/src/api/jobs.ts new file mode 100644 index 0000000..ebae1f2 --- /dev/null +++ b/frontend/src/api/jobs.ts @@ -0,0 +1,21 @@ +import { api } from './client' +import type { + JobCreate, + JobOut, + JobRunResponse, + JobUpdate, + JobWithLastRun, +} from '../types' + +export const jobApi = { + list: () => api.get('/jobs').then((r) => r.data), + get: (id: number) => api.get(`/jobs/${id}`).then((r) => r.data), + create: (payload: JobCreate) => api.post('/jobs', payload).then((r) => r.data), + update: (id: number, payload: JobUpdate) => + api.put(`/jobs/${id}`, payload).then((r) => r.data), + remove: (id: number) => api.delete(`/jobs/${id}`).then((r) => r.data), + run: (id: number) => + api.post(`/jobs/${id}/run`).then((r) => r.data), + toggle: (id: number) => + api.post(`/jobs/${id}/toggle`).then((r) => r.data), +} diff --git a/frontend/src/api/restore.ts b/frontend/src/api/restore.ts new file mode 100644 index 0000000..fd31db6 --- /dev/null +++ b/frontend/src/api/restore.ts @@ -0,0 +1,17 @@ +import { api } from './client' +import type { + RestoreOut, + RestoreRequest, + RestoreTargetsResponse, +} from '../types' + +export const restoreApi = { + getTargets: (run_id: number) => + api + .get('/restore/targets', { params: { run_id } }) + .then((r) => r.data), + create: (payload: RestoreRequest) => + api.post('/restore', payload).then((r) => r.data), + history: (run_id?: number) => + api.get('/restore/history', { params: { run_id } }).then((r) => r.data), +} diff --git a/frontend/src/api/runs.ts b/frontend/src/api/runs.ts new file mode 100644 index 0000000..a212fce --- /dev/null +++ b/frontend/src/api/runs.ts @@ -0,0 +1,11 @@ +import { api } from './client' +import type { RunListResponse, RunLogOut, RunOut } from '../types' + +export const runApi = { + list: (params?: { job_id?: number; status?: string; page?: number; page_size?: number }) => + api.get('/runs', { params }).then((r) => r.data), + get: (id: number) => api.get(`/runs/${id}`).then((r) => r.data), + getLog: (id: number) => api.get(`/runs/${id}/log`).then((r) => r.data), + downloadUrl: (id: number) => `/api/runs/${id}/download`, + remove: (id: number) => api.delete(`/runs/${id}`).then((r) => r.data), +} diff --git a/frontend/src/api/storages.ts b/frontend/src/api/storages.ts new file mode 100644 index 0000000..bc388d9 --- /dev/null +++ b/frontend/src/api/storages.ts @@ -0,0 +1,19 @@ +import { api } from './client' +import type { + StorageCreate, + StorageOut, + StorageTestResponse, + StorageUpdate, +} from '../types' + +export const storageApi = { + list: () => api.get('/storages').then((r) => r.data), + get: (id: number) => api.get(`/storages/${id}`).then((r) => r.data), + create: (payload: StorageCreate) => + api.post('/storages', payload).then((r) => r.data), + update: (id: number, payload: StorageUpdate) => + api.put(`/storages/${id}`, payload).then((r) => r.data), + remove: (id: number) => api.delete(`/storages/${id}`).then((r) => r.data), + test: (id: number) => + api.post(`/storages/${id}/test`).then((r) => r.data), +} diff --git a/frontend/src/components/Layout/index.tsx b/frontend/src/components/Layout/index.tsx new file mode 100644 index 0000000..46baa82 --- /dev/null +++ b/frontend/src/components/Layout/index.tsx @@ -0,0 +1,128 @@ +import { Layout as AntLayout, Menu, Button, Avatar, Dropdown, Space, theme } from 'antd' +import { + DashboardOutlined, + ScheduleOutlined, + HistoryOutlined, + DatabaseOutlined, + RollbackOutlined, + SettingOutlined, + LogoutOutlined, + UserOutlined, + MenuFoldOutlined, + MenuUnfoldOutlined, +} from '@ant-design/icons' +import { Outlet, useLocation, useNavigate, Link } from 'react-router-dom' +import { useAuth } from '../../hooks/useAuth' +import { useUiStore } from '../../stores/uiStore' + +const { Sider, Header, Content } = AntLayout + +const menuItems = [ + { key: '/dashboard', icon: , label: 概览 }, + { key: '/jobs', icon: , label: 备份任务 }, + { key: '/history', icon: , label: 运行历史 }, + { key: '/restore', icon: , label: 恢复 }, + { key: '/storages', icon: , label: 存储目标 }, + { key: '/settings', icon: , label: 设置 }, +] + +export function Layout() { + const navigate = useNavigate() + const location = useLocation() + const { user, logout } = useAuth() + const { sidebarCollapsed, toggleSidebar } = useUiStore() + const { token } = theme.useToken() + + const selectedKey = + menuItems.find((m) => location.pathname.startsWith(m.key))?.key || '/dashboard' + + return ( + + +
+ {sidebarCollapsed ? 'BS' : '备份系统'} +
+ + + +
+
+ +
+ +
+ + + + ) +} diff --git a/frontend/src/components/ProtectedRoute.tsx b/frontend/src/components/ProtectedRoute.tsx new file mode 100644 index 0000000..1fd4455 --- /dev/null +++ b/frontend/src/components/ProtectedRoute.tsx @@ -0,0 +1,20 @@ +import { ReactNode } from 'react' +import { Navigate, useLocation } from 'react-router-dom' +import { Result } from 'antd' +import { useAuth } from '../hooks/useAuth' + +export function ProtectedRoute({ children, adminOnly = false }: { children: ReactNode; adminOnly?: boolean }) { + const { token, user } = useAuth() + const location = useLocation() + + if (!token) { + return + } + if (!user) { + return null + } + if (adminOnly && !user.is_admin) { + return + } + return <>{children} +} diff --git a/frontend/src/hooks/useAuth.ts b/frontend/src/hooks/useAuth.ts new file mode 100644 index 0000000..d061407 --- /dev/null +++ b/frontend/src/hooks/useAuth.ts @@ -0,0 +1,6 @@ +import { useAuthStore } from '../stores/authStore' + +export function useAuth() { + const { user, token, loading, login, logout } = useAuthStore() + return { user, token, loading, login, logout, isAdmin: user?.is_admin ?? false } +} diff --git a/frontend/src/main.tsx b/frontend/src/main.tsx new file mode 100644 index 0000000..707dbc5 --- /dev/null +++ b/frontend/src/main.tsx @@ -0,0 +1,27 @@ +import React from 'react' +import ReactDOM from 'react-dom/client' +import { ConfigProvider, App as AntdApp } from 'antd' +import zhCN from 'antd/locale/zh_CN' +import 'dayjs/locale/zh-cn' +import dayjs from 'dayjs' +import App from './App' + +dayjs.locale('zh-cn') + +ReactDOM.createRoot(document.getElementById('root')!).render( + + + + + + + , +) diff --git a/frontend/src/pages/Dashboard.tsx b/frontend/src/pages/Dashboard.tsx new file mode 100644 index 0000000..60bc4a7 --- /dev/null +++ b/frontend/src/pages/Dashboard.tsx @@ -0,0 +1,192 @@ +import { useEffect, useState } from 'react' +import { Card, Col, Row, Statistic, Table, Tag, Typography, Spin } from 'antd' +import { + DatabaseOutlined, + ScheduleOutlined, + CheckCircleOutlined, + CloseCircleOutlined, + ClockCircleOutlined, +} from '@ant-design/icons' +import { Link } from 'react-router-dom' +import { jobApi } from '../api/jobs' +import { runApi } from '../api/runs' +import type { JobWithLastRun, RunOut } from '../types' +import dayjs from 'dayjs' + +const { Title } = Typography + +function formatBytes(n: number | null | undefined): string { + if (n == null) return '-' + if (n < 1024) return `${n} B` + if (n < 1024 * 1024) return `${(n / 1024).toFixed(1)} KB` + if (n < 1024 * 1024 * 1024) return `${(n / 1024 / 1024).toFixed(1)} MB` + return `${(n / 1024 / 1024 / 1024).toFixed(2)} GB` +} + +export function Dashboard() { + const [jobs, setJobs] = useState([]) + const [runs, setRuns] = useState([]) + const [loading, setLoading] = useState(true) + + useEffect(() => { + Promise.all([jobApi.list(), runApi.list({ page: 1, page_size: 10 })]) + .then(([j, r]) => { + setJobs(j) + setRuns(r.items) + }) + .finally(() => setLoading(false)) + }, []) + + const last24h = runs.filter((r) => + r.started_at ? dayjs(r.started_at).isAfter(dayjs().subtract(24, 'hour')) : false, + ) + const success24h = last24h.filter((r) => r.status === 'success').length + const failed24h = last24h.filter((r) => r.status === 'failed').length + const totalSize = runs.reduce((s, r) => s + (r.artifact_size || 0), 0) + + if (loading) return + + return ( +
+ 概览 + + + + } + /> + + + + + j.storage.id)).size} + prefix={} + /> + + + + + } + /> + + + + + 0 ? '#cf1322' : undefined }} + prefix={} + /> + + + + + + + 查看全部}> + { + const job = jobs.find((j) => j.id === id) + return job?.name || `#${id}` + }, + }, + { + title: '状态', + dataIndex: 'status', + width: 90, + render: (s: string) => { + const color = + s === 'success' ? 'green' : s === 'failed' ? 'red' : s === 'running' ? 'blue' : 'default' + const icon = + s === 'success' ? : s === 'failed' ? : + return {s} + }, + }, + { + title: '大小', + dataIndex: 'artifact_size', + width: 100, + render: (n: number | null) => formatBytes(n), + }, + { + title: '时间', + dataIndex: 'started_at', + width: 160, + render: (t: string | null) => (t ? dayjs(t).format('MM-DD HH:mm:ss') : '-'), + }, + { + title: '', + width: 80, + render: (_: unknown, r: RunOut) => ( + 详情 + ), + }, + ]} + /> + + + + +
{t.toUpperCase()}, + }, + { + title: '上次状态', + width: 100, + render: (_, j: JobWithLastRun) => { + if (!j.last_run) return 未运行 + const s = j.last_run.status + const color = + s === 'success' ? 'green' : s === 'failed' ? 'red' : 'default' + return {s} + }, + }, + { + title: '上次时间', + width: 160, + render: (_, j: JobWithLastRun) => + j.last_run?.started_at ? dayjs(j.last_run.started_at).format('MM-DD HH:mm:ss') : '-', + }, + ]} + /> + + + + + + } + /> + + + ) +} diff --git a/frontend/src/pages/History.tsx b/frontend/src/pages/History.tsx new file mode 100644 index 0000000..030a658 --- /dev/null +++ b/frontend/src/pages/History.tsx @@ -0,0 +1,147 @@ +import { useEffect, useState } from 'react' +import { Card, Select, Space, Table, Tag, Typography, Button, App } from 'antd' +import { Link } from 'react-router-dom' +import { runApi } from '../api/runs' +import { jobApi } from '../api/jobs' +import type { JobWithLastRun, RunOut } from '../types' +import dayjs from 'dayjs' + +const { Title } = Typography + +function formatBytes(n: number | null | undefined): string { + if (n == null) return '-' + if (n < 1024) return `${n} B` + if (n < 1024 * 1024) return `${(n / 1024).toFixed(1)} KB` + if (n < 1024 * 1024 * 1024) return `${(n / 1024 / 1024).toFixed(1)} MB` + return `${(n / 1024 / 1024 / 1024).toFixed(2)} GB` +} + +export function History() { + const { message } = App.useApp() + const [runs, setRuns] = useState([]) + const [total, setTotal] = useState(0) + const [page, setPage] = useState(1) + const [pageSize, setPageSize] = useState(20) + const [jobs, setJobs] = useState([]) + const [filterJob, setFilterJob] = useState() + const [filterStatus, setFilterStatus] = useState() + const [loading, setLoading] = useState(false) + + const fetch = () => { + setLoading(true) + runApi + .list({ job_id: filterJob, status: filterStatus, page, page_size: pageSize }) + .then((r) => { + setRuns(r.items) + setTotal(r.total) + }) + .finally(() => setLoading(false)) + } + + useEffect(() => { + jobApi.list().then(setJobs) + }, []) + + useEffect(fetch, [page, pageSize, filterJob, filterStatus]) + + return ( +
+ 运行历史 + + + 任务： + + + + + +
{ + setPage(p) + setPageSize(ps) + }, + showSizeChanger: true, + }} + columns={[ + { title: 'ID', dataIndex: 'id', width: 60 }, + { + title: '任务', + dataIndex: 'job_id', + render: (id: number) => jobs.find((j) => j.id === id)?.name || `#${id}`, + }, + { + title: '状态', + dataIndex: 'status', + width: 100, + render: (s: string) => { + const color = s === 'success' ? 'green' : s === 'failed' ? 'red' : s === 'running' ? 'blue' : 'default' + return {s} + }, + }, + { + title: '触发', + dataIndex: 'trigger', + width: 90, + render: (t: string) => {t}, + }, + { + title: '开始时间', + dataIndex: 'started_at', + width: 170, + render: (t: string | null) => (t ? dayjs(t).format('YYYY-MM-DD HH:mm:ss') : '-'), + }, + { + title: '耗时', + dataIndex: 'duration_seconds', + width: 80, + render: (s: number | null) => (s != null ? `${s}s` : '-'), + }, + { + title: '大小', + dataIndex: 'artifact_size', + width: 100, + render: formatBytes, + }, + { + title: '操作', + width: 120, + render: (_, r: RunOut) => ( + + 详情 + + ), + }, + ]} + /> + + + ) +} diff --git a/frontend/src/pages/Jobs/JobCreate.tsx b/frontend/src/pages/Jobs/JobCreate.tsx new file mode 100644 index 0000000..64c3055 --- /dev/null +++ b/frontend/src/pages/Jobs/JobCreate.tsx @@ -0,0 +1,5 @@ +import { JobForm } from './JobForm' + +export function JobCreate() { + return +} diff --git a/frontend/src/pages/Jobs/JobEdit.tsx b/frontend/src/pages/Jobs/JobEdit.tsx new file mode 100644 index 0000000..d362ae8 --- /dev/null +++ b/frontend/src/pages/Jobs/JobEdit.tsx @@ -0,0 +1,5 @@ +import { JobForm } from './JobForm' + +export function JobEdit() { + return +} diff --git a/frontend/src/pages/Jobs/JobForm.tsx b/frontend/src/pages/Jobs/JobForm.tsx new file mode 100644 index 0000000..17625d7 --- /dev/null +++ b/frontend/src/pages/Jobs/JobForm.tsx @@ -0,0 +1,268 @@ +import { useEffect, useState } from 'react' +import { + Button, + Card, + Form, + Input, + InputNumber, + Select, + Space, + Switch, + Typography, + App, + Divider, + Alert, +} from 'antd' +import { useNavigate, useParams } from 'react-router-dom' +import { jobApi } from '../../api/jobs' +import { storageApi } from '../../api/storages' +import type { JobCreate, JobUpdate, JobWithLastRun, StorageOut } from '../../types' + +const { Title } = Typography + +interface FormValues { + name: string + type: 'mysql' | 'directory' + storage_id: number + enabled: boolean + cron_expression?: string + retention_count: number + retention_days?: number + description?: string + // MySQL + mysql_host?: string + mysql_port?: number + mysql_user?: string + mysql_password?: string + mysql_database?: string + mysql_extra_args?: string + // Directory + dir_path?: string + dir_exclude_patterns?: string +} + +function toPayload(values: FormValues): JobCreate { + const source_config = + values.type === 'mysql' + ? { + host: values.mysql_host!, + port: values.mysql_port || 3306, + user: values.mysql_user!, + password: values.mysql_password || '', + database: values.mysql_database!, + extra_args: values.mysql_extra_args || '', + } + : { + path: values.dir_path!, + exclude_patterns: values.dir_exclude_patterns + ? values.dir_exclude_patterns.split(',').map((s) => s.trim()).filter(Boolean) + : [], + } + + return { + name: values.name, + type: values.type, + storage_id: values.storage_id, + source_config, + enabled: values.enabled, + cron_expression: values.cron_expression?.trim() || undefined, + retention_count: values.retention_count, + retention_days: values.retention_days, + description: values.description, + } +} + +export function JobForm({ mode }: { mode: 'create' | 'edit' }) { + const { id } = useParams<{ id: string }>() + const navigate = useNavigate() + const { message } = App.useApp() + const [form] = Form.useForm() + const [storages, setStorages] = useState([]) + const [loading, setLoading] = useState(false) + const [existing, setExisting] = useState(null) + const jobType = Form.useWatch('type', form) + + useEffect(() => { + storageApi.list().then(setStorages) + if (mode === 'edit' && id) { + jobApi + .get(Number(id)) + .then((j) => { + setExisting(j) + const flat: Partial = { + name: j.name, + type: j.type, + storage_id: j.storage.id, + enabled: j.enabled, + cron_expression: j.cron_expression || '', + retention_count: j.retention_count, + retention_days: j.retention_days ?? undefined, + description: j.description || '', + } + // 反解 source_config（需要解密后的，前端拿不到 - 用空 source_config 占位让用户重填） + form.setFieldsValue(flat as FormValues) + }) + } + }, [mode, id]) + + const onSubmit = async (values: FormValues) => { + setLoading(true) + try { + if (mode === 'create') { + await jobApi.create(toPayload(values)) + message.success('创建成功') + } else { + const payload: JobUpdate = { + name: values.name, + storage_id: values.storage_id, + enabled: values.enabled, + cron_expression: values.cron_expression?.trim() || null, + retention_count: values.retention_count, + retention_days: values.retention_days ?? null, + description: values.description ?? null, + } + await jobApi.update(Number(id), payload) + message.success('已更新') + } + navigate('/jobs') + } catch { + /* */ + } finally { + setLoading(false) + } + } + + return ( +
+ {mode === 'create' ? '新建备份任务' : `编辑任务 #${id}`} + +
+ + + + + + + + + + + + + + + + + + + + + + + + + )} + + {jobType === 'directory' && ( + <> + 目录数据源 + + + + + + + + )} + + 存储与调度 + + + + + + + + + + 保留策略 + + + + + + + + + + + + + + + + + + + + + + +
+ ) +} diff --git a/frontend/src/pages/Jobs/JobList.tsx b/frontend/src/pages/Jobs/JobList.tsx new file mode 100644 index 0000000..a3643c5 --- /dev/null +++ b/frontend/src/pages/Jobs/JobList.tsx @@ -0,0 +1,146 @@ +import { useEffect, useState } from 'react' +import { + Button, + Card, + Popconfirm, + Space, + Table, + Tag, + Typography, + App, + Switch, + Tooltip, +} from 'antd' +import { PlusOutlined, PlayCircleOutlined, EditOutlined, DeleteOutlined } from '@ant-design/icons' +import { Link, useNavigate } from 'react-router-dom' +import { jobApi } from '../../api/jobs' +import type { JobWithLastRun } from '../../types' +import dayjs from 'dayjs' + +const { Title } = Typography + +export function JobList() { + const navigate = useNavigate() + const { message, modal } = App.useApp() + const [jobs, setJobs] = useState([]) + const [loading, setLoading] = useState(false) + + const fetch = () => { + setLoading(true) + jobApi + .list() + .then(setJobs) + .finally(() => setLoading(false)) + } + + useEffect(fetch, []) + + const onRun = async (id: number) => { + try { + const resp = await jobApi.run(id) + message.success(`已触发备份，run_id=${resp.run_id}`) + setTimeout(fetch, 500) + } catch { + /* handled by interceptor */ + } + } + + const onToggle = async (id: number) => { + try { + await jobApi.toggle(id) + fetch() + } catch { + /* */ + } + } + + const onDelete = async (id: number) => { + try { + await jobApi.remove(id) + message.success('已删除') + fetch() + } catch { + /* */ + } + } + + return ( +
+
+ 备份任务 + +
+ +
{t.toUpperCase()}, + }, + { title: '存储', dataIndex: ['storage', 'name'] }, + { + title: 'Cron', + dataIndex: 'cron_expression', + width: 140, + render: (c: string | null) => (c ? {c} : 手动), + }, + { + title: '启用', + dataIndex: 'enabled', + width: 70, + render: (e: boolean, r) => ( + onToggle(r.id)} /> + ), + }, + { + title: '上次状态', + width: 100, + render: (_, r: JobWithLastRun) => { + if (!r.last_run) return 未运行 + const s = r.last_run.status + const color = s === 'success' ? 'green' : s === 'failed' ? 'red' : 'blue' + return {s} + }, + }, + { + title: '上次时间', + width: 170, + render: (_, r: JobWithLastRun) => + r.last_run?.started_at ? dayjs(r.last_run.started_at).format('YYYY-MM-DD HH:mm:ss') : '-', + }, + { + title: '操作', + width: 200, + render: (_, r) => ( + + + + + + + + ) +} diff --git a/frontend/src/pages/Restore.tsx b/frontend/src/pages/Restore.tsx new file mode 100644 index 0000000..e1d782a --- /dev/null +++ b/frontend/src/pages/Restore.tsx @@ -0,0 +1,217 @@ +import { useEffect, useState } from 'react' +import { + Alert, + Button, + Card, + Form, + Input, + Radio, + Select, + Space, + Steps, + Table, + Tag, + Typography, + App, +} from 'antd' +import { useSearchParams } from 'react-router-dom' +import { restoreApi } from '../api/restore' +import { runApi } from '../api/runs' +import { jobApi } from '../api/jobs' +import type { JobWithLastRun, RestoreTargetsResponse, RunOut } from '../types' + +const { Title } = Typography + +export function Restore() { + const { message, modal } = App.useApp() + const [params] = useSearchParams() + const [step, setStep] = useState(0) + const [runs, setRuns] = useState([]) + const [jobs, setJobs] = useState([]) + const [selectedRun, setSelectedRun] = useState(null) + const [targets, setTargets] = useState(null) + const [targetType, setTargetType] = useState<'mysql_db' | 'directory_path'>('mysql_db') + const [target, setTarget] = useState('') + const [submitting, setSubmitting] = useState(false) + + useEffect(() => { + Promise.all([ + runApi.list({ status: 'success', page: 1, page_size: 50 }), + jobApi.list(), + ]).then(([r, j]) => { + setRuns(r.items) + setJobs(j) + }) + }, []) + + useEffect(() => { + const rid = params.get('run_id') + if (rid) { + const r = runs.find((x) => String(x.id) === rid) + if (r) { + setSelectedRun(r) + setStep(1) + loadTargets(r.id) + } + } + }, [params, runs]) + + const loadTargets = async (runId: number) => { + const t = await restoreApi.getTargets(runId) + setTargets(t) + setTargetType(t.job_type === 'mysql' ? 'mysql_db' : 'directory_path') + setTarget(t.suggested_targets[0] || '') + } + + const onConfirm = async () => { + if (!selectedRun || !target.trim()) { + message.warning('请选择目标') + return + } + modal.confirm({ + title: '确认恢复？', + content: `将覆盖目标 ${target}。建议先备份当前状态再继续。`, + okText: '确认', + cancelText: '取消', + onOk: async () => { + setSubmitting(true) + try { + await restoreApi.create({ + run_id: selectedRun.id, + target_type: targetType, + target: target.trim(), + }) + message.success('恢复任务已提交，请到运行历史查看') + setStep(2) + } catch { + /* */ + } finally { + setSubmitting(false) + } + }, + }) + } + + return ( +
+ 从备份恢复 + + + + + {step === 0 && ( + +
jobs.find((j) => j.id === id)?.name || `#${id}`, + }, + { + title: '类型', + render: (_, r: RunOut) => jobs.find((j) => j.id === r.job_id)?.type?.toUpperCase(), + }, + { + title: '开始时间', + dataIndex: 'started_at', + render: (t: string | null) => (t ? new Date(t).toLocaleString() : '-'), + }, + { + title: '操作', + render: (_, r: RunOut) => ( + + ), + }, + ]} + /> + + )} + + {step === 1 && selectedRun && targets && ( + + +
+ + {jobs.find((j) => j.id === selectedRun.job_id)?.name} + + {selectedRun.started_at && new Date(selectedRun.started_at).toLocaleString()} + + + + setTargetType(e.target.value)} + disabled + > + MySQL 数据库 + 目录路径 + + + + setTarget(e.target.value)} + placeholder={targets.suggested_targets[0] || ''} + /> + + + + + + + + + + )} + + {step === 2 && ( + + + 请到「运行历史」查看对应 run 的日志输出，或回到{' '} + 运行历史。 + + } + /> + + + )} + + ) +} diff --git a/frontend/src/pages/RunDetail.tsx b/frontend/src/pages/RunDetail.tsx new file mode 100644 index 0000000..c7d0523 --- /dev/null +++ b/frontend/src/pages/RunDetail.tsx @@ -0,0 +1,135 @@ +import { useEffect, useState } from 'react' +import { useParams, Link } from 'react-router-dom' +import { Card, Descriptions, Tag, Typography, Button, Space, Spin, Alert } from 'antd' +import { DownloadOutlined, RollbackOutlined } from '@ant-design/icons' +import { runApi } from '../api/runs' +import { jobApi } from '../api/jobs' +import { useAuth } from '../hooks/useAuth' +import type { JobWithLastRun, RunLogOut, RunOut } from '../types' +import dayjs from 'dayjs' + +const { Title } = Typography + +function formatBytes(n: number | null | undefined): string { + if (n == null) return '-' + if (n < 1024) return `${n} B` + if (n < 1024 * 1024) return `${(n / 1024).toFixed(1)} KB` + if (n < 1024 * 1024 * 1024) return `${(n / 1024 / 1024).toFixed(1)} MB` + return `${(n / 1024 / 1024 / 1024).toFixed(2)} GB` +} + +export function RunDetail() { + const { id } = useParams<{ id: string }>() + const { token } = useAuth() + const [run, setRun] = useState(null) + const [log, setLog] = useState(null) + const [job, setJob] = useState(null) + const [loading, setLoading] = useState(true) + + useEffect(() => { + if (!id) return + Promise.all([runApi.get(Number(id)), runApi.getLog(Number(id))]) + .then(async ([r, l]) => { + setRun(r) + setLog(l) + const j = await jobApi.get(r.job_id) + setJob(j) + }) + .finally(() => setLoading(false)) + }, [id]) + + if (loading) return + if (!run) return + + return ( +
+ + 运行详情 #{run.id} + {run.status === 'success' && ( + + )} + {run.status === 'success' && job && ( + + + + )} + + + + + + + {run.status} + + + + {run.trigger} + + + {job?.name || `#${run.job_id}`} + + {job?.type?.toUpperCase()} + + {run.started_at ? dayjs(run.started_at).format('YYYY-MM-DD HH:mm:ss') : '-'} + + + {run.finished_at ? dayjs(run.finished_at).format('YYYY-MM-DD HH:mm:ss') : '-'} + + + {run.duration_seconds != null ? `${run.duration_seconds}s` : '-'} + + {formatBytes(run.artifact_size)} + + {run.artifact_checksum || '-'} + + + {run.artifact_path || '-'} + + {run.error_message && ( + + + + )} + + + + + 
+          {log?.content || run.log_excerpt || '(无日志)'}
+
+ +
+ ) +} diff --git a/frontend/src/pages/Settings.tsx b/frontend/src/pages/Settings.tsx new file mode 100644 index 0000000..3459db7 --- /dev/null +++ b/frontend/src/pages/Settings.tsx @@ -0,0 +1,79 @@ +import { useState } from 'react' +import { Button, Card, Form, Input, Typography, App, Space } from 'antd' +import { useAuth } from '../hooks/useAuth' +import { authApi } from '../api/auth' + +const { Title } = Typography + +export function Settings() { + const { user, logout } = useAuth() + const { message } = App.useApp() + const [loading, setLoading] = useState(false) + const [form] = Form.useForm<{ old_password: string; new_password: string; confirm: string }>() + + const onChangePassword = async (v: { old_password: string; new_password: string; confirm: string }) => { + if (v.new_password !== v.confirm) { + message.error('两次输入的新密码不一致') + return + } + setLoading(true) + try { + await authApi.changePassword(v.old_password, v.new_password) + message.success('密码已修改，请重新登录') + setTimeout(() => { + logout() + window.location.href = '/login' + }, 800) + } catch { + /* */ + } finally { + setLoading(false) + } + } + + return ( +
+ 设置 + + +

用户名：{user?.username}

+

角色：{user?.is_admin ? '管理员' : '普通用户'}

+

创建时间：{user?.created_at && new Date(user.created_at).toLocaleString()}

+ + + +
+ + + + + + + + + + + + + + + + +
+ ) +} diff --git a/frontend/src/pages/Storages.tsx b/frontend/src/pages/Storages.tsx new file mode 100644 index 0000000..08934ef --- /dev/null +++ b/frontend/src/pages/Storages.tsx @@ -0,0 +1,259 @@ +import { useEffect, useState } from 'react' +import { + Button, + Card, + Form, + Input, + InputNumber, + Modal, + Popconfirm, + Select, + Space, + Switch, + Table, + Tag, + Typography, + App, + Divider, +} from 'antd' +import { PlusOutlined, EditOutlined, DeleteOutlined, ApiOutlined } from '@ant-design/icons' +import { storageApi } from '../api/storages' +import type { LocalStorageConfig, S3StorageConfig, StorageCreate, StorageOut, StorageUpdate } from '../types' + +const { Title } = Typography + +interface FormValues { + name: string + type: 'local' | 's3' + is_default: boolean + // local + local_path?: string + // s3 + s3_endpoint_url?: string + s3_region?: string + s3_bucket?: string + s3_access_key?: string + s3_secret_key?: string + s3_use_ssl?: boolean + s3_path_prefix?: string +} + +export function Storages() { + const { message } = App.useApp() + const [items, setItems] = useState([]) + const [loading, setLoading] = useState(false) + const [open, setOpen] = useState(false) + const [editing, setEditing] = useState(null) + const [form] = Form.useForm() + + const fetch = () => { + setLoading(true) + storageApi.list().then(setItems).finally(() => setLoading(false)) + } + + useEffect(fetch, []) + + const onCreate = () => { + setEditing(null) + form.resetFields() + form.setFieldsValue({ type: 'local', is_default: false, s3_use_ssl: true, s3_region: 'us-east-1' }) + setOpen(true) + } + + const onEdit = (row: StorageOut) => { + setEditing(row) + form.resetFields() + form.setFieldsValue({ + name: row.name, + type: row.type, + is_default: row.is_default, + // config 明文不返回（后端不暴露），让用户重填 + }) + setOpen(true) + } + + const onSubmit = async () => { + const v = await form.validateFields() + const config: LocalStorageConfig | S3StorageConfig = + v.type === 'local' + ? { path: v.local_path! } + : { + endpoint_url: v.s3_endpoint_url || undefined, + region: v.s3_region || 'us-east-1', + bucket: v.s3_bucket!, + access_key: v.s3_access_key!, + secret_key: v.s3_secret_key!, + use_ssl: v.s3_use_ssl ?? true, + path_prefix: v.s3_path_prefix || '', + } + + try { + if (editing) { + const payload: StorageUpdate = { + name: v.name, + is_default: v.is_default, + config, + } + await storageApi.update(editing.id, payload) + message.success('已更新') + } else { + const payload: StorageCreate = { + name: v.name, + type: v.type, + is_default: v.is_default, + config, + } + await storageApi.create(payload) + message.success('已创建') + } + setOpen(false) + fetch() + } catch { + /* */ + } + } + + const onDelete = async (id: number) => { + try { + await storageApi.remove(id) + message.success('已删除') + fetch() + } catch { + /* */ + } + } + + const onTest = async (id: number) => { + try { + const r = await storageApi.test(id) + if (r.ok) message.success(r.message) + else message.error(r.message) + } catch { + /* */ + } + } + + const typeWatch = Form.useWatch('type', form) + + return ( +
+
+ 存储目标 + +
+ +
{t.toUpperCase()}, + }, + { + title: '默认', + dataIndex: 'is_default', + width: 80, + render: (d: boolean) => (d ? 是 : '-'), + }, + { + title: '创建时间', + dataIndex: 'created_at', + render: (t: string) => new Date(t).toLocaleString(), + }, + { + title: '操作', + width: 280, + render: (_, r: StorageOut) => ( + + + + onDelete(r.id)}> +