190 lines
7.5 KiB
Python
190 lines
7.5 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status, BackgroundTasks
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select, desc
|
|
from typing import List
|
|
import time
|
|
import subprocess
|
|
import asyncio
|
|
from pathlib import Path
|
|
from datetime import datetime, timezone
|
|
|
|
from app.core.database import get_db, AsyncSessionLocal
|
|
from app.models.models import User, ScriptJob, ScriptJobRun
|
|
from app.schemas.schemas import ScriptJobResponse, ScriptJobCreate, ScriptJobUpdate, ScriptJobRunResponse
|
|
from app.api.deps import require_admin
|
|
|
|
router = APIRouter(prefix="/script-jobs", tags=["Script Jobs Management"], dependencies=[Depends(require_admin)])
|
|
|
|
async def execute_script_in_background(job_id: int):
|
|
"""
|
|
Executes a script (python/bash) in a separate thread and saves status and logs to the DB.
|
|
"""
|
|
async with AsyncSessionLocal() as db:
|
|
result = await db.execute(select(ScriptJob).where(ScriptJob.id == job_id))
|
|
job = result.scalar_one_or_none()
|
|
if not job:
|
|
return
|
|
|
|
run = ScriptJobRun(
|
|
script_job_id=job.id,
|
|
status="RUNNING",
|
|
started_at=datetime.now(timezone.utc)
|
|
)
|
|
db.add(run)
|
|
await db.commit()
|
|
await db.refresh(run)
|
|
|
|
start_time = time.time()
|
|
base_dir = Path(__file__).resolve().parent.parent.parent.parent
|
|
script_full_path = base_dir / job.script_path
|
|
|
|
# Find virtualenv python interpreter or fallback
|
|
if job.script_type == "python":
|
|
venv_python_win = base_dir / "backend" / "venv" / "Scripts" / "python.exe"
|
|
venv_python_lin = base_dir / "backend" / "venv" / "bin" / "python"
|
|
|
|
if venv_python_win.exists():
|
|
venv_python = venv_python_win
|
|
elif venv_python_lin.exists():
|
|
venv_python = venv_python_lin
|
|
else:
|
|
import sys
|
|
venv_python = sys.executable
|
|
cmd = [str(venv_python), str(script_full_path)]
|
|
else:
|
|
import platform
|
|
if platform.system().lower() == "windows":
|
|
# For Windows bash/sh script, run PowerShell as wrapper
|
|
cmd = ["powershell.exe", "-Command", str(script_full_path)]
|
|
else:
|
|
cmd = ["bash", str(script_full_path)]
|
|
|
|
try:
|
|
loop = asyncio.get_running_loop()
|
|
|
|
def run_subprocess():
|
|
return subprocess.run(
|
|
cmd,
|
|
capture_output=True,
|
|
text=True,
|
|
cwd=str(base_dir),
|
|
encoding='utf-8',
|
|
errors='replace' # Handle Karen-style unicode error logs safely!
|
|
)
|
|
|
|
res = await loop.run_in_executor(None, run_subprocess)
|
|
duration = time.time() - start_time
|
|
|
|
run.completed_at = datetime.now(timezone.utc)
|
|
run.duration_seconds = round(duration, 2)
|
|
run.log_output = f"--- STDOUT ---\n{res.stdout}\n\n--- STDERR ---\n{res.stderr}"
|
|
run.status = "SUCCESS" if res.returncode == 0 else "FAILED"
|
|
if res.returncode != 0:
|
|
run.log_output += f"\n\nProcess exited with return code: {res.returncode}"
|
|
|
|
except Exception as e:
|
|
duration = time.time() - start_time
|
|
run.completed_at = datetime.now(timezone.utc)
|
|
run.duration_seconds = round(duration, 2)
|
|
run.status = "FAILED"
|
|
run.log_output = f"Execution failed due to launcher error:\n{str(e)}"
|
|
|
|
db.add(run)
|
|
await db.commit()
|
|
|
|
@router.get("", response_model=List[ScriptJobResponse])
|
|
async def list_script_jobs(db: AsyncSession = Depends(get_db)):
|
|
"""List all configured server-side script backup jobs."""
|
|
result = await db.execute(select(ScriptJob).order_by(ScriptJob.id.asc()))
|
|
return result.scalars().all()
|
|
|
|
@router.post("", response_model=ScriptJobResponse)
|
|
async def create_script_job(payload: ScriptJobCreate, db: AsyncSession = Depends(get_db)):
|
|
"""Create a new server-side script job (cron schedule)."""
|
|
# Verify path exists
|
|
base_dir = Path(__file__).resolve().parent.parent.parent.parent
|
|
target_path = base_dir / payload.script_path
|
|
if not target_path.exists():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Script file '{payload.script_path}' was not found on the server."
|
|
)
|
|
|
|
job = ScriptJob(
|
|
name=payload.name,
|
|
script_type=payload.script_type,
|
|
script_path=payload.script_path,
|
|
schedule_cron=payload.schedule_cron,
|
|
is_active=payload.is_active
|
|
)
|
|
db.add(job)
|
|
await db.commit()
|
|
await db.refresh(job)
|
|
return job
|
|
|
|
@router.put("/{job_id}", response_model=ScriptJobResponse)
|
|
async def update_script_job(job_id: int, payload: ScriptJobUpdate, db: AsyncSession = Depends(get_db)):
|
|
"""Update a script job configuration."""
|
|
result = await db.execute(select(ScriptJob).where(ScriptJob.id == job_id))
|
|
job = result.scalar_one_or_none()
|
|
if not job:
|
|
raise HTTPException(status_code=404, detail="Script job not found.")
|
|
|
|
if payload.name is not None:
|
|
job.name = payload.name
|
|
if payload.script_type is not None:
|
|
job.script_type = payload.script_type
|
|
if payload.script_path is not None:
|
|
# Verify path
|
|
base_dir = Path(__file__).resolve().parent.parent.parent.parent
|
|
target_path = base_dir / payload.script_path
|
|
if not target_path.exists():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Script file '{payload.script_path}' was not found on the server."
|
|
)
|
|
job.script_path = payload.script_path
|
|
if payload.schedule_cron is not None:
|
|
job.schedule_cron = payload.schedule_cron
|
|
if payload.is_active is not None:
|
|
job.is_active = payload.is_active
|
|
|
|
db.add(job)
|
|
await db.commit()
|
|
await db.refresh(job)
|
|
return job
|
|
|
|
@router.delete("/{job_id}")
|
|
async def delete_script_job(job_id: int, db: AsyncSession = Depends(get_db)):
|
|
"""Deletes a script job and its execution history."""
|
|
result = await db.execute(select(ScriptJob).where(ScriptJob.id == job_id))
|
|
job = result.scalar_one_or_none()
|
|
if not job:
|
|
raise HTTPException(status_code=404, detail="Script job not found.")
|
|
|
|
await db.delete(job)
|
|
await db.commit()
|
|
return {"message": f"Script job '{job.name}' deleted successfully."}
|
|
|
|
@router.post("/{job_id}/trigger")
|
|
async def trigger_script_job(job_id: int, background_tasks: BackgroundTasks, db: AsyncSession = Depends(get_db)):
|
|
"""Triggers immediate execution of a script job in the background."""
|
|
result = await db.execute(select(ScriptJob).where(ScriptJob.id == job_id))
|
|
job = result.scalar_one_or_none()
|
|
if not job:
|
|
raise HTTPException(status_code=404, detail="Script job not found.")
|
|
|
|
background_tasks.add_task(execute_script_in_background, job.id)
|
|
return {"message": f"Script execution for '{job.name}' triggered successfully in background."}
|
|
|
|
@router.get("/{job_id}/runs", response_model=List[ScriptJobRunResponse])
|
|
async def get_script_job_runs(job_id: int, db: AsyncSession = Depends(get_db)):
|
|
"""List execution history runs for a specific script job."""
|
|
result = await db.execute(
|
|
select(ScriptJobRun)
|
|
.where(ScriptJobRun.script_job_id == job_id)
|
|
.order_by(desc(ScriptJobRun.started_at))
|
|
.limit(50)
|
|
)
|
|
return result.scalars().all()
|