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()