207 lines
7.2 KiB
Python
207 lines
7.2 KiB
Python
"""
|
|
Rutas de importación CSV para Vehículos (Transportes).
|
|
Flujo: upload → scan → status (polling) → commit.
|
|
"""
|
|
import json
|
|
import logging
|
|
import os
|
|
import threading
|
|
from uuid import uuid4
|
|
|
|
from fastapi import APIRouter, File, HTTPException, Query, UploadFile, Depends
|
|
from sqlalchemy.orm import Session
|
|
from typing import Dict, Any
|
|
|
|
from core.celery_app import celery_app
|
|
from core.database import get_core_db
|
|
from core.security import get_current_user, validate_access_to_resource
|
|
from api.v1.modules.core.tasks_tracking import track_and_dispatch
|
|
|
|
from .schemas import ImportJobResponse
|
|
from .tasks import (
|
|
scan_file,
|
|
run_scan_sync,
|
|
run_commit_sync,
|
|
JOB_TYPE,
|
|
VEHL_IMPORT_STATUS_PREFIX,
|
|
VEHL_IMPORT_REDIS_TTL,
|
|
)
|
|
from ..common import storage as common_storage
|
|
from ..common.error_csv import download_scan_errors_csv_stream
|
|
from ..common.responses import normalize_commit_status_payload
|
|
|
|
router = APIRouter()
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _get_redis():
|
|
import redis
|
|
url = os.getenv("VALKEY_URL", os.getenv("REDIS_URL", "redis://valkey:6379/0"))
|
|
return redis.Redis.from_url(url, decode_responses=False)
|
|
|
|
|
|
@router.post("/upload", response_model=ImportJobResponse)
|
|
async def upload_import_file(
|
|
file: UploadFile = File(...),
|
|
company_id: int = Query(..., description="Company ID"),
|
|
actualizar: bool = Query(False, description="Modo Agregar/Actualizar (ACT); si True, validación parcial para claves existentes"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
try:
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
except Exception as e:
|
|
logger.error(f"Vehicles import: access validation failed: {e}")
|
|
raise HTTPException(status_code=403, detail="Invalid company access")
|
|
|
|
if not file.filename or not file.filename.lower().endswith(".csv"):
|
|
raise HTTPException(status_code=400, detail="Solo se permiten archivos .csv")
|
|
|
|
job_id = str(uuid4())
|
|
contents = await file.read()
|
|
|
|
meta_data = {
|
|
"tenant_id": tenant_id,
|
|
"company_id": company_id,
|
|
"user_id": current_user.get("id"),
|
|
"template_id": "vehicles",
|
|
"actualizar": actualizar,
|
|
}
|
|
|
|
try:
|
|
common_storage.store_import_file(
|
|
JOB_TYPE,
|
|
job_id,
|
|
contents,
|
|
meta_data,
|
|
tenant_id=int(tenant_id),
|
|
company_id=company_id,
|
|
ttl=VEHL_IMPORT_REDIS_TTL,
|
|
log_label="Vehicles import",
|
|
)
|
|
except common_storage.ImportStoreError as e:
|
|
logger.error(f"Vehicles import: store error: {e}")
|
|
raise HTTPException(status_code=500, detail="No se pudo encolar el archivo.")
|
|
except Exception as e:
|
|
logger.error(f"Vehicles import: Redis store error: {e}")
|
|
raise HTTPException(status_code=500, detail="No se pudo encolar el archivo.")
|
|
|
|
track_and_dispatch(
|
|
db=db,
|
|
task=scan_file,
|
|
tenant_id=int(tenant_id),
|
|
company_id=company_id,
|
|
requested_by_user=current_user.get("preferred_username") or current_user.get("email") or current_user.get("sub"),
|
|
task_name="vehicles_scan_file",
|
|
task_group="layouts_csv",
|
|
task_origin="a76/layouts_csv/vehicles/upload",
|
|
args=[job_id],
|
|
task_id=job_id,
|
|
)
|
|
# Fallback sin worker: ejecutar scan en un hilo y guardar resultado en Redis
|
|
def run_scan_background():
|
|
try:
|
|
run_scan_sync(job_id)
|
|
except Exception as e:
|
|
logger.exception(f"Vehicles import: background scan failed: {e}")
|
|
|
|
threading.Thread(target=run_scan_background, daemon=True).start()
|
|
|
|
return ImportJobResponse(
|
|
job_id=job_id,
|
|
status="queued",
|
|
message="Archivo subido. Escaneo iniciado.",
|
|
)
|
|
|
|
|
|
@router.get("/{job_id}/status")
|
|
async def get_import_status(job_id: str):
|
|
# Si el resultado del scan está en Redis (worker o hilo de fallback), usarlo
|
|
try:
|
|
r = _get_redis()
|
|
raw = r.get(f"{VEHL_IMPORT_STATUS_PREFIX}{job_id}")
|
|
if raw:
|
|
data = json.loads(raw.decode("utf-8"))
|
|
if isinstance(data, dict) and data.get("status") in ("finished", "warning"):
|
|
return normalize_commit_status_payload(data)
|
|
return data
|
|
except Exception as e:
|
|
logger.debug(f"Vehicles import: could not read status from Redis: {e}")
|
|
|
|
task_result = celery_app.AsyncResult(job_id)
|
|
|
|
if task_result.state == "PENDING":
|
|
return {"status": "processing", "progress": 0}
|
|
if task_result.state == "PROGRESS":
|
|
info = (task_result.info or {})
|
|
return {
|
|
"status": "processing",
|
|
"progress": info.get("current", 0),
|
|
"total": info.get("total", 0),
|
|
}
|
|
if task_result.state == "SUCCESS":
|
|
result = task_result.result
|
|
if isinstance(result, dict) and "status" in result:
|
|
return normalize_commit_status_payload(result)
|
|
return {"status": "finished", "result": result}
|
|
|
|
result = getattr(task_result, "result", None)
|
|
if isinstance(result, dict) and result.get("status") in ("finished", "warning"):
|
|
return normalize_commit_status_payload(result)
|
|
|
|
logger.warning("Vehicles import task %s failed: state=%s", job_id, task_result.state)
|
|
err_msg = None
|
|
tb = getattr(task_result, "traceback", None)
|
|
if tb and isinstance(tb, str):
|
|
lines = [l.strip() for l in tb.strip().split("\n") if l.strip()]
|
|
if lines:
|
|
err_msg = lines[-1]
|
|
if not err_msg:
|
|
try:
|
|
exc = task_result.get(propagate=False)
|
|
if exc is not None:
|
|
err_msg = str(exc)
|
|
except Exception:
|
|
pass
|
|
if not err_msg and result is not None:
|
|
if not isinstance(result, dict):
|
|
err_msg = str(result)
|
|
elif result.get("error") or result.get("message"):
|
|
err_msg = result.get("error") or result.get("message")
|
|
return {"status": "failed", "error": err_msg or "Task failed"}
|
|
|
|
|
|
@router.post("/{job_id}/commit")
|
|
async def commit_import_job(job_id: str):
|
|
# Escribir "processing" en Redis para que el polling muestre estado hasta que termine el commit
|
|
try:
|
|
r = _get_redis()
|
|
r.set(
|
|
f"{VEHL_IMPORT_STATUS_PREFIX}{job_id}",
|
|
json.dumps({"status": "processing", "message": "Insertando..."}).encode("utf-8"),
|
|
ex=VEHL_IMPORT_REDIS_TTL,
|
|
)
|
|
except Exception as e:
|
|
logger.debug(f"Vehicles import: could not write processing status: {e}")
|
|
|
|
# Ejecutar commit en hilo (mismo fallback que el scan: no depende del worker de Celery)
|
|
def run_commit_background():
|
|
try:
|
|
run_commit_sync(job_id)
|
|
except Exception as e:
|
|
logger.exception(f"Vehicles import: background commit failed: {e}")
|
|
|
|
threading.Thread(target=run_commit_background, daemon=True).start()
|
|
|
|
# Mismo job_id para que el front siga haciendo polling y reciba el resultado de Redis
|
|
return {
|
|
"status": "committing",
|
|
"message": "Inserción iniciada.",
|
|
"commit_job_id": job_id,
|
|
}
|
|
|
|
|
|
@router.get("/{job_id}/errors/scan-csv")
|
|
async def download_scan_errors_csv(job_id: str):
|
|
return download_scan_errors_csv_stream("veh", job_id)
|