250 lines
9.5 KiB
Python
250 lines
9.5 KiB
Python
"""YakPanel - Backup plans API"""
|
|
import os
|
|
import tarfile
|
|
from datetime import datetime
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select
|
|
from pydantic import BaseModel
|
|
from croniter import croniter
|
|
|
|
from app.core.database import get_db
|
|
from app.core.config import get_runtime_config
|
|
from app.core.notification import send_email
|
|
from app.api.auth import get_current_user
|
|
from app.models.user import User
|
|
from app.models.site import Site
|
|
from app.models.database import Database
|
|
from app.models.backup_plan import BackupPlan
|
|
from app.services.database_service import backup_mysql_database, backup_postgresql_database, backup_mongodb_database
|
|
|
|
router = APIRouter(prefix="/backup", tags=["backup"])
|
|
|
|
|
|
class CreateBackupPlanRequest(BaseModel):
|
|
name: str
|
|
plan_type: str # site | database
|
|
target_id: int
|
|
schedule: str # cron, e.g. "0 2 * * *"
|
|
enabled: bool = True
|
|
s3_bucket: str = ""
|
|
s3_endpoint: str = ""
|
|
s3_key_prefix: str = ""
|
|
|
|
|
|
@router.get("/plans")
|
|
async def backup_plans_list(
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""List all backup plans"""
|
|
result = await db.execute(select(BackupPlan).order_by(BackupPlan.id))
|
|
rows = result.scalars().all()
|
|
return [
|
|
{
|
|
"id": r.id,
|
|
"name": r.name,
|
|
"plan_type": r.plan_type,
|
|
"target_id": r.target_id,
|
|
"schedule": r.schedule,
|
|
"enabled": r.enabled,
|
|
"s3_bucket": getattr(r, "s3_bucket", None) or "",
|
|
"s3_endpoint": getattr(r, "s3_endpoint", None) or "",
|
|
"s3_key_prefix": getattr(r, "s3_key_prefix", None) or "",
|
|
}
|
|
for r in rows
|
|
]
|
|
|
|
|
|
@router.post("/plans")
|
|
async def backup_plan_create(
|
|
body: CreateBackupPlanRequest,
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Create a backup plan"""
|
|
if body.plan_type not in ("site", "database"):
|
|
raise HTTPException(status_code=400, detail="plan_type must be site or database")
|
|
if not body.schedule or len(body.schedule) < 9:
|
|
raise HTTPException(status_code=400, detail="Invalid cron schedule")
|
|
try:
|
|
croniter(body.schedule)
|
|
except Exception:
|
|
raise HTTPException(status_code=400, detail="Invalid cron expression")
|
|
if body.plan_type == "site":
|
|
r = await db.execute(select(Site).where(Site.id == body.target_id))
|
|
if not r.scalar_one_or_none():
|
|
raise HTTPException(status_code=404, detail="Site not found")
|
|
else:
|
|
r = await db.execute(select(Database).where(Database.id == body.target_id))
|
|
if not r.scalar_one_or_none():
|
|
raise HTTPException(status_code=404, detail="Database not found")
|
|
plan = BackupPlan(
|
|
name=body.name,
|
|
plan_type=body.plan_type,
|
|
target_id=body.target_id,
|
|
schedule=body.schedule,
|
|
enabled=body.enabled,
|
|
s3_bucket=(body.s3_bucket or "")[:256],
|
|
s3_endpoint=(body.s3_endpoint or "")[:512],
|
|
s3_key_prefix=(body.s3_key_prefix or "")[:256],
|
|
)
|
|
db.add(plan)
|
|
await db.commit()
|
|
return {"status": True, "msg": "Backup plan created", "id": plan.id}
|
|
|
|
|
|
@router.put("/plans/{plan_id}")
|
|
async def backup_plan_update(
|
|
plan_id: int,
|
|
body: CreateBackupPlanRequest,
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Update a backup plan"""
|
|
result = await db.execute(select(BackupPlan).where(BackupPlan.id == plan_id))
|
|
plan = result.scalar_one_or_none()
|
|
if not plan:
|
|
raise HTTPException(status_code=404, detail="Backup plan not found")
|
|
if body.schedule:
|
|
try:
|
|
croniter(body.schedule)
|
|
except Exception:
|
|
raise HTTPException(status_code=400, detail="Invalid cron expression")
|
|
plan.name = body.name
|
|
plan.plan_type = body.plan_type
|
|
plan.target_id = body.target_id
|
|
plan.schedule = body.schedule
|
|
plan.enabled = body.enabled
|
|
plan.s3_bucket = (body.s3_bucket or "")[:256]
|
|
plan.s3_endpoint = (body.s3_endpoint or "")[:512]
|
|
plan.s3_key_prefix = (body.s3_key_prefix or "")[:256]
|
|
await db.commit()
|
|
return {"status": True, "msg": "Updated"}
|
|
|
|
|
|
@router.delete("/plans/{plan_id}")
|
|
async def backup_plan_delete(
|
|
plan_id: int,
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Delete a backup plan"""
|
|
result = await db.execute(select(BackupPlan).where(BackupPlan.id == plan_id))
|
|
plan = result.scalar_one_or_none()
|
|
if not plan:
|
|
raise HTTPException(status_code=404, detail="Backup plan not found")
|
|
await db.delete(plan)
|
|
await db.commit()
|
|
return {"status": True, "msg": "Deleted"}
|
|
|
|
|
|
def _run_site_backup(site: Site) -> tuple[bool, str, str | None]:
|
|
"""Run site backup (sync, for use in run_scheduled). Returns (ok, msg, filename)."""
|
|
cfg = get_runtime_config()
|
|
backup_dir = cfg["backup_path"]
|
|
os.makedirs(backup_dir, exist_ok=True)
|
|
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
filename = f"{site.name}_{ts}.tar.gz"
|
|
dest = os.path.join(backup_dir, filename)
|
|
try:
|
|
with tarfile.open(dest, "w:gz") as tf:
|
|
tf.add(site.path, arcname=os.path.basename(site.path))
|
|
return True, "Backup created", filename
|
|
except Exception as e:
|
|
return False, str(e), None
|
|
|
|
|
|
def _maybe_upload_s3(local_file: str, plan: BackupPlan) -> tuple[bool, str]:
|
|
"""Copy backup file to S3-compatible bucket if plan.s3_bucket set. Uses AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY."""
|
|
bucket = (getattr(plan, "s3_bucket", None) or "").strip()
|
|
if not bucket or not os.path.isfile(local_file):
|
|
return True, ""
|
|
try:
|
|
import boto3
|
|
except ImportError:
|
|
return False, "boto3 not installed (pip install boto3)"
|
|
ep = (getattr(plan, "s3_endpoint", None) or "").strip() or None
|
|
prefix = (getattr(plan, "s3_key_prefix", None) or "").strip().strip("/")
|
|
key_base = os.path.basename(local_file)
|
|
key = f"{prefix}/{key_base}" if prefix else key_base
|
|
try:
|
|
client = boto3.client("s3", endpoint_url=ep)
|
|
client.upload_file(local_file, bucket, key)
|
|
return True, f"s3://{bucket}/{key}"
|
|
except Exception as e:
|
|
return False, str(e)
|
|
|
|
|
|
def _run_database_backup(dbo: Database) -> tuple[bool, str, str | None]:
|
|
"""Run database backup (sync). Returns (ok, msg, filename)."""
|
|
cfg = get_runtime_config()
|
|
backup_dir = os.path.join(cfg["backup_path"], "database")
|
|
if dbo.db_type == "MySQL":
|
|
return backup_mysql_database(dbo.name, backup_dir)
|
|
if dbo.db_type == "PostgreSQL":
|
|
return backup_postgresql_database(dbo.name, backup_dir)
|
|
if dbo.db_type == "MongoDB":
|
|
return backup_mongodb_database(dbo.name, backup_dir)
|
|
return False, "Unsupported database type", None
|
|
|
|
|
|
@router.post("/run-scheduled")
|
|
async def backup_run_scheduled(
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Run all due backup plans. Call this from cron (e.g. every hour) or manually."""
|
|
from datetime import datetime as dt
|
|
now = dt.utcnow()
|
|
cfg = get_runtime_config()
|
|
result = await db.execute(select(BackupPlan).where(BackupPlan.enabled == True))
|
|
plans = result.scalars().all()
|
|
results = []
|
|
for plan in plans:
|
|
ok = False
|
|
msg = ""
|
|
try:
|
|
prev_run = croniter(plan.schedule, now).get_prev(dt)
|
|
secs_since = (now - prev_run).total_seconds()
|
|
if secs_since > 900 or secs_since < 0:
|
|
continue
|
|
except Exception:
|
|
continue
|
|
if plan.plan_type == "site":
|
|
r = await db.execute(select(Site).where(Site.id == plan.target_id))
|
|
site = r.scalar_one_or_none()
|
|
if not site or not os.path.isdir(site.path):
|
|
results.append({"plan": plan.name, "status": "skipped", "msg": "Site not found or path invalid"})
|
|
continue
|
|
ok, msg, filename = _run_site_backup(site)
|
|
if ok and filename:
|
|
full = os.path.join(cfg["backup_path"], filename)
|
|
u_ok, u_msg = _maybe_upload_s3(full, plan)
|
|
if u_msg:
|
|
msg = f"{msg}; {u_msg}" if u_ok else f"{msg}; S3 failed: {u_msg}"
|
|
if ok:
|
|
send_email(
|
|
subject=f"YakPanel - Scheduled backup: {plan.name}",
|
|
body=f"Site backup completed: {filename}\nSite: {site.name}",
|
|
)
|
|
else:
|
|
r = await db.execute(select(Database).where(Database.id == plan.target_id))
|
|
dbo = r.scalar_one_or_none()
|
|
if not dbo:
|
|
results.append({"plan": plan.name, "status": "skipped", "msg": "Database not found"})
|
|
continue
|
|
ok, msg, filename = _run_database_backup(dbo)
|
|
if ok and filename:
|
|
full = os.path.join(cfg["backup_path"], "database", filename)
|
|
u_ok, u_msg = _maybe_upload_s3(full, plan)
|
|
if u_msg:
|
|
msg = f"{msg}; {u_msg}" if u_ok else f"{msg}; S3 failed: {u_msg}"
|
|
if ok:
|
|
send_email(
|
|
subject=f"YakPanel - Scheduled backup: {plan.name}",
|
|
body=f"Database backup completed: {filename}\nDatabase: {dbo.name}",
|
|
)
|
|
results.append({"plan": plan.name, "status": "ok" if ok else "failed", "msg": msg})
|
|
return {"status": True, "results": results}
|