from __future__ import annotations import os from pathlib import Path from typing import Any, Dict, List, Optional from fastapi import Depends, FastAPI, File, Form, HTTPException, Query, UploadFile from fastapi.responses import Response from pydantic import BaseModel, Field from domain.models import JobStatus from infrastructure.auth import require_bearer from infrastructure.storage_minio import MinioSettings, MinioStorage from infrastructure.store import JobStore store = JobStore() minio = MinioStorage(MinioSettings.from_env()) app = FastAPI(title="BotDescarga Remote API", version="1.0.0") class RegisterWorkerBody(BaseModel): worker_id: str hostname: str = "" max_concurrent: int = 1 class HeartbeatBody(BaseModel): worker_id: str class EventBody(BaseModel): event_type: str message: str radicado: Optional[str] = None nit: Optional[str] = None index: Optional[int] = None total: Optional[int] = None metadata: Dict[str, Any] = Field(default_factory=dict) timestamp: Optional[str] = None class PatchJobBody(BaseModel): status: Optional[str] = None result: Optional[Dict[str, Any]] = None error: Optional[str] = None @app.get("/health") def health() -> dict: return {"status": "ok", "service": "bot-descarga"} @app.post("/workers/register") def register_worker(body: RegisterWorkerBody, _: str = Depends(require_bearer)) -> dict: worker = store.register_worker( body.worker_id, hostname=body.hostname, max_concurrent=body.max_concurrent, ) return worker.to_dict() @app.post("/workers/heartbeat") def heartbeat(body: HeartbeatBody, _: str = Depends(require_bearer)) -> dict: worker = store.heartbeat(body.worker_id) if not worker: raise HTTPException(status_code=404, detail="Worker no registrado") return worker.to_dict() @app.get("/workers") def list_workers(_: str = Depends(require_bearer)) -> List[dict]: return [w.to_dict() for w in store.list_workers()] @app.get("/workers/me/pending") def claim_pending( worker_id: str = Query(...), _: str = Depends(require_bearer), ) -> dict: job = store.claim_pending(worker_id) if not job: return {"job": None} return {"job": job.to_dict()} @app.post("/jobs") async def create_job( worker_id: str = Form(...), job_id: str = Form(...), title: str = Form(""), file: UploadFile = File(...), _: str = Depends(require_bearer), ) -> dict: if not worker_id.strip(): raise HTTPException(status_code=400, detail="worker_id obligatorio") parent_job_id = job_id.strip() if not parent_job_id: raise HTTPException(status_code=400, detail="job_id (Job Desktop) obligatorio") if not store.parent_job_exists(parent_job_id): raise HTTPException(status_code=404, detail="Trabajo (job) no encontrado") if not store.is_worker_online(worker_id): raise HTTPException( status_code=409, detail=f"Worker '{worker_id}' no está online. Enciende el agent primero.", ) data = await file.read() if not data: raise HTTPException(status_code=400, detail="Archivo vacío") original = file.filename or "input.xlsx" content_type = file.content_type or "application/octet-stream" title_clean = title.strip() job = store.create_job( worker_id=worker_id.strip(), excel_key="", original_filename=original, title=title_clean, job_id=parent_job_id, ) task_id = store.create_downloads_task( parent_job_id=parent_job_id, download_job_id=job.id, worker_id=worker_id.strip(), title=title_clean or f"Bot descarga ({original})", original_filename=original, ) job = store.get_job(job.id) assert job is not None try: key = minio.upload_excel(job.id, data, original, content_type=content_type) except Exception as e: store.patch_job(job.id, status=JobStatus.FAILED.value, error=f"MinIO upload failed: {e}") raise HTTPException(status_code=502, detail=f"Error subiendo a MinIO: {e}") from e job = store.update_excel_key(job.id, key) assert job is not None store.add_event( job.id, "run_started", f"Job creado y asignado a {worker_id}", metadata={ "excel_key": key, "worker_id": worker_id, "job_id": parent_job_id, "task_id": task_id, }, ) return job.to_dict() @app.get("/jobs") def list_jobs( limit: int = Query(100, ge=1, le=500), job_id: Optional[str] = Query(None, description="Filtrar por Job Desktop"), _: str = Depends(require_bearer), ) -> List[dict]: return [j.to_dict() for j in store.list_jobs(limit=limit, job_id=job_id)] @app.get("/jobs/{download_job_id}") def get_job(download_job_id: str, _: str = Depends(require_bearer)) -> dict: job = store.get_job(download_job_id) if not job: raise HTTPException(status_code=404, detail="Job no encontrado") return job.to_dict() @app.get("/jobs/{download_job_id}/files") def list_job_files(download_job_id: str, _: str = Depends(require_bearer)) -> dict: job = store.get_job(download_job_id) if not job: raise HTTPException(status_code=404, detail="Job no encontrado") try: return minio.list_download_files(download_job_id, prefix=job.downloads_prefix) except Exception as e: raise HTTPException(status_code=502, detail=f"Error listando MinIO: {e}") from e @app.get("/jobs/{download_job_id}/object") def get_job_object( download_job_id: str, key: str = Query(..., min_length=1, description="Object key bajo downloads_prefix"), _: str = Depends(require_bearer), ): """Stream de un PDF del resultado MinIO (solo keys del prefijo del job).""" job = store.get_job(download_job_id) if not job: raise HTTPException(status_code=404, detail="Job no encontrado") prefix = (job.downloads_prefix or f"downloads/{download_job_id}/").lstrip("/") if not prefix.endswith("/"): prefix += "/" normalized = key.lstrip("/") if not normalized.startswith(prefix): raise HTTPException(status_code=403, detail="Key fuera del prefijo del job") if not normalized.lower().endswith(".pdf"): raise HTTPException(status_code=400, detail="Solo se permiten objetos PDF") if ".." in normalized.split("/"): raise HTTPException(status_code=400, detail="Key no permitida") if normalized.endswith("_manifest.json"): raise HTTPException(status_code=400, detail="Key no permitida") try: data = minio.get_bytes(normalized) except Exception as e: raise HTTPException(status_code=502, detail=f"Error leyendo MinIO: {e}") from e filename = Path(normalized).name return Response( content=data, media_type="application/pdf", headers={"Content-Disposition": f'inline; filename="{filename}"'}, ) @app.get("/jobs/{download_job_id}/events") def get_events( download_job_id: str, since: int = Query(0, ge=0), _: str = Depends(require_bearer), ) -> dict: if not store.get_job(download_job_id): raise HTTPException(status_code=404, detail="Job no encontrado") events = store.list_events(download_job_id, since_id=since) return {"events": [e.to_dict() for e in events]} @app.post("/jobs/{download_job_id}/cancel") def cancel_job(download_job_id: str, _: str = Depends(require_bearer)) -> dict: job = store.request_cancel(download_job_id) if not job: raise HTTPException(status_code=404, detail="Job no encontrado") store.add_event(download_job_id, "warning", "Cancelación solicitada por el usuario") return job.to_dict() @app.get("/jobs/{download_job_id}/file") def get_job_file(download_job_id: str, _: str = Depends(require_bearer)): job = store.get_job(download_job_id) if not job: raise HTTPException(status_code=404, detail="Job no encontrado") if not job.excel_key: raise HTTPException(status_code=404, detail="Excel no disponible") try: data = minio.get_bytes(job.excel_key) except Exception as e: raise HTTPException(status_code=502, detail=f"Error leyendo MinIO: {e}") from e filename = Path(job.excel_key).name return Response( content=data, media_type="application/octet-stream", headers={"Content-Disposition": f'attachment; filename="{filename}"'}, ) @app.get("/jobs/{download_job_id}/file-url") def get_job_file_url(download_job_id: str, _: str = Depends(require_bearer)) -> dict: job = store.get_job(download_job_id) if not job or not job.excel_key: raise HTTPException(status_code=404, detail="Job/archivo no encontrado") try: url = minio.presigned_get(job.excel_key) except Exception as e: raise HTTPException(status_code=502, detail=str(e)) from e return {"url": url, "excel_key": job.excel_key} @app.post("/jobs/{download_job_id}/events") def post_event( download_job_id: str, body: EventBody, _: str = Depends(require_bearer), ) -> dict: if not store.get_job(download_job_id): raise HTTPException(status_code=404, detail="Job no encontrado") event = store.add_event( download_job_id, body.event_type, body.message, radicado=body.radicado, nit=body.nit, index=body.index, total=body.total, metadata=body.metadata, timestamp=body.timestamp, ) job = store.get_job(download_job_id) if job and body.index is not None and body.total is not None: result = dict(job.result or {}) result["processed"] = body.index result["total"] = body.total if body.nit: result["nit_actual"] = body.nit store.patch_job(download_job_id, result=result) return event.to_dict() @app.patch("/jobs/{download_job_id}") def patch_job( download_job_id: str, body: PatchJobBody, _: str = Depends(require_bearer), ) -> dict: if body.status and body.status not in {s.value for s in JobStatus}: raise HTTPException(status_code=400, detail="status inválido") job = store.patch_job( download_job_id, status=body.status, result=body.result, error=body.error, ) if not job: raise HTTPException(status_code=404, detail="Job no encontrado") return job.to_dict() @app.get("/jobs/{download_job_id}/cancel-flag") def cancel_flag(download_job_id: str, _: str = Depends(require_bearer)) -> dict: job = store.get_job(download_job_id) if not job: raise HTTPException(status_code=404, detail="Job no encontrado") return {"cancel_requested": job.cancel_requested, "status": job.status}