""" app.py - Splunk Packaging Toolkit FastAPI Service All settings imported from config.py """ import uuid import logging from pathlib import Path from datetime import datetime from concurrent.futures import ThreadPoolExecutor from fastapi import FastAPI, File, UploadFile, HTTPException, BackgroundTasks from fastapi.responses import FileResponse, HTMLResponse from fastapi.middleware.cors import CORSMiddleware from build import build_packages from config import ( APP_TITLE, APP_VERSION, APP_DESCRIPTION, ALLOWED_EXTENSIONS, MAX_UPLOAD_BYTES, BASE_WORK_DIR, MAX_CONCURRENT_JOBS, JOB_ID_LENGTH, CORS_ALLOW_ORIGINS, CORS_ALLOW_METHODS, CORS_ALLOW_HEADERS, POLL_INTERVAL_MS, LOG_TAIL_LINES, PROGRESS_STEP_PCT, ) logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = FastAPI(title=APP_TITLE, description=APP_DESCRIPTION, version=APP_VERSION) app.add_middleware( CORSMiddleware, allow_origins=CORS_ALLOW_ORIGINS, allow_methods=CORS_ALLOW_METHODS, allow_headers=CORS_ALLOW_HEADERS, ) JOBS: dict = {} BASE_DIR = Path(BASE_WORK_DIR) BASE_DIR.mkdir(parents=True, exist_ok=True) executor = ThreadPoolExecutor(max_workers=MAX_CONCURRENT_JOBS) ALLOWED_EXT_STR = " · ".join(ALLOWED_EXTENSIONS) ACCEPT_ATTR = ",".join(ALLOWED_EXTENSIONS) def run_build(job_id: str, input_file: Path, work_dir: Path): job = JOBS[job_id] job["status"] = "running" job["started_at"] = datetime.utcnow().isoformat() try: result = build_packages(input_file, work_dir, job["logs"]) job["status"] = "done" job["result"] = result job["done_at"] = datetime.utcnow().isoformat() except Exception as e: job["status"] = "failed" job["error"] = str(e) job["logs"].append(f"[FAIL] {e}") # ── HTML stored as plain string — no f-string ──────────────── HTML = """ __TITLE__

__TITLE__

__DESC__
Drop app here or click to browse
__EXTS__
-
Search Head
-
Indexer
-
Forwarder
Download role packages
""" def render_html() -> str: return (HTML .replace("__TITLE__", APP_TITLE) .replace("__DESC__", APP_DESCRIPTION) .replace("__ACCEPT__", ACCEPT_ATTR) .replace("__EXTS__", ALLOWED_EXT_STR) .replace("__POLL_MS__", str(POLL_INTERVAL_MS)) .replace("__LOG_TAIL__", str(LOG_TAIL_LINES)) .replace("__PROG_STEP__", str(PROGRESS_STEP_PCT)) ) # ════════════════════════════════════════════════════════════ # ROUTES — plain strings, no f-string, no double braces # ════════════════════════════════════════════════════════════ @app.get("/", response_class=HTMLResponse) async def index(): return render_html() @app.post("/upload") async def upload(background_tasks: BackgroundTasks, file: UploadFile = File(...)): filename = file.filename or "" if not any(filename.endswith(ext) for ext in ALLOWED_EXTENSIONS): raise HTTPException(400, f"Only {', '.join(ALLOWED_EXTENSIONS)} files supported") content = await file.read() if len(content) > MAX_UPLOAD_BYTES: raise HTTPException(413, f"File too large. Max {MAX_UPLOAD_BYTES // (1024*1024)}MB") job_id = str(uuid.uuid4())[:JOB_ID_LENGTH] work_dir = BASE_DIR / job_id work_dir.mkdir(parents=True, exist_ok=True) input_path = work_dir / filename input_path.write_bytes(content) JOBS[job_id] = { "job_id": job_id, "filename": filename, "status": "queued", "created_at": datetime.utcnow().isoformat(), "started_at": None, "done_at": None, "logs": [], "result": None, "error": None, } background_tasks.add_task(run_build, job_id, input_path, work_dir) return {"job_id": job_id, "filename": filename, "status": "queued"} @app.get("/status/{job_id}") async def status(job_id: str): job = JOBS.get(job_id) if not job: raise HTTPException(404, "Job not found") return job @app.get("/download/{job_id}/{role}") async def download(job_id: str, role: str): job = JOBS.get(job_id) if not job: raise HTTPException(404, "Job not found") if job["status"] != "done": raise HTTPException(400, "Build not complete yet") pkgs = job["result"].get("packages", {}) if role not in pkgs: raise HTTPException(404, f"Role '{role}' not found") pkg_path = Path(pkgs[role]["path"]) if not pkg_path.exists(): raise HTTPException(404, "Package file missing on disk") return FileResponse( str(pkg_path), filename=pkgs[role]["filename"], media_type="application/gzip" ) @app.get("/logs/{job_id}") async def logs(job_id: str): job = JOBS.get(job_id) if not job: raise HTTPException(404, "Job not found") return {"job_id": job_id, "status": job["status"], "logs": job["logs"]} @app.get("/health") async def health(): return { "status": "ok", "version": APP_VERSION, "title": APP_TITLE, "jobs": len(JOBS), }