feature/csv-trasportes
This commit is contained in:
hreyes
@@ -0,0 +1 @@
|
||||
# CSV import for Vehicles / Transportes (upload → scan → status → commit).
|
||||
@@ -0,0 +1,192 @@
|
||||
"""
|
||||
Rutas de importación CSV para Vehículos (Transportes).
|
||||
Flujo: upload → scan → status (polling) → commit.
|
||||
"""
|
||||
import base64
|
||||
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 .schemas import ImportJobResponse
|
||||
from .tasks import (
|
||||
scan_file,
|
||||
run_scan_sync,
|
||||
run_commit_sync,
|
||||
VEHL_IMPORT_FILE_PREFIX,
|
||||
VEHL_IMPORT_META_PREFIX,
|
||||
VEHL_IMPORT_STATUS_PREFIX,
|
||||
VEHL_IMPORT_REDIS_TTL,
|
||||
)
|
||||
|
||||
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"),
|
||||
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",
|
||||
}
|
||||
|
||||
try:
|
||||
r = _get_redis()
|
||||
r.set(
|
||||
f"{VEHL_IMPORT_FILE_PREFIX}{job_id}",
|
||||
base64.b64encode(contents),
|
||||
ex=VEHL_IMPORT_REDIS_TTL,
|
||||
)
|
||||
r.set(
|
||||
f"{VEHL_IMPORT_META_PREFIX}{job_id}",
|
||||
json.dumps(meta_data).encode("utf-8"),
|
||||
ex=VEHL_IMPORT_REDIS_TTL,
|
||||
)
|
||||
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.")
|
||||
|
||||
try:
|
||||
upload_dir = os.path.join(os.getcwd(), "uploads", "temp")
|
||||
os.makedirs(upload_dir, exist_ok=True)
|
||||
with open(os.path.join(upload_dir, f"veh_{job_id}.csv"), "wb") as f:
|
||||
f.write(contents)
|
||||
with open(os.path.join(upload_dir, f"veh_{job_id}.meta.json"), "w") as f:
|
||||
json.dump(meta_data, f)
|
||||
except Exception as e:
|
||||
logger.warning(f"Vehicles import: local file save failed: {e}")
|
||||
|
||||
scan_file.apply_async(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"))
|
||||
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 result
|
||||
return {"status": "finished", "result": result}
|
||||
|
||||
result = getattr(task_result, "result", None)
|
||||
if isinstance(result, dict) and result.get("status") in ("finished", "warning"):
|
||||
return 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,
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
from pydantic import BaseModel
|
||||
from typing import Optional
|
||||
|
||||
|
||||
class ImportJobResponse(BaseModel):
|
||||
job_id: str
|
||||
status: str
|
||||
message: str
|
||||
|
||||
|
||||
class ImportJobStatus(BaseModel):
|
||||
status: str
|
||||
job_id: str
|
||||
total_rows: Optional[int] = 0
|
||||
error_count: Optional[int] = 0
|
||||
valid_rows: Optional[int] = 0
|
||||
error: Optional[str] = None
|
||||
inserted: Optional[int] = 0
|
||||
skipped_invalid: Optional[int] = 0
|
||||
skipped_missing_fk: Optional[int] = 0
|
||||
skipped_duplicate: Optional[int] = 0
|
||||
skipped_details: Optional[list] = None
|
||||
@@ -0,0 +1,607 @@
|
||||
"""
|
||||
Tareas Celery para importación CSV de Vehículos (Transportes).
|
||||
Flujo: scan_file (validación) → insert_valid_rows (commit).
|
||||
Respeta la lógica manual: vehicle_key requerido, resto opcional; upsert por vehicle_key.
|
||||
"""
|
||||
import os
|
||||
import base64
|
||||
import csv
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
import unicodedata
|
||||
from decimal import Decimal
|
||||
from typing import Dict, Any, Optional, List
|
||||
|
||||
from core.celery_app import celery_app
|
||||
from core.database import CoreSessionLocal
|
||||
|
||||
from .template_config import row_from_template
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
VEHL_IMPORT_FILE_PREFIX = "veh_import_file:"
|
||||
VEHL_IMPORT_META_PREFIX = "veh_import_meta:"
|
||||
VEHL_IMPORT_ERROR_LINES_PREFIX = "veh_import_error_lines:"
|
||||
VEHL_IMPORT_STATUS_PREFIX = "veh_import_status:"
|
||||
VEHL_IMPORT_REDIS_TTL = 3600
|
||||
|
||||
|
||||
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)
|
||||
|
||||
|
||||
def _worker_upload_dir() -> str:
|
||||
return os.path.join(os.getcwd(), "uploads", "temp")
|
||||
|
||||
|
||||
def _ensure_worker_has_file_from_redis(job_id: str) -> Optional[str]:
|
||||
r = _get_redis()
|
||||
data = r.get(f"{VEHL_IMPORT_FILE_PREFIX}{job_id}")
|
||||
if not data:
|
||||
return None
|
||||
try:
|
||||
raw = base64.b64decode(data)
|
||||
except Exception as e:
|
||||
logger.warning(f"Vehicles import: failed to decode file from Redis: {e}")
|
||||
return None
|
||||
upload_dir = _worker_upload_dir()
|
||||
os.makedirs(upload_dir, exist_ok=True)
|
||||
file_path = os.path.join(upload_dir, f"veh_{job_id}.csv")
|
||||
with open(file_path, "wb") as f:
|
||||
f.write(raw)
|
||||
return file_path
|
||||
|
||||
|
||||
def _ensure_worker_has_meta_from_redis(job_id: str, file_path: str) -> bool:
|
||||
r = _get_redis()
|
||||
data = r.get(f"{VEHL_IMPORT_META_PREFIX}{job_id}")
|
||||
if not data:
|
||||
return False
|
||||
try:
|
||||
meta = json.loads(data.decode("utf-8"))
|
||||
except Exception as e:
|
||||
logger.warning(f"Vehicles import: failed to decode meta from Redis: {e}")
|
||||
return False
|
||||
meta_path = file_path.replace(".csv", ".meta.json")
|
||||
with open(meta_path, "w", encoding="utf-8") as f:
|
||||
json.dump(meta, f)
|
||||
return True
|
||||
|
||||
|
||||
def _delete_import_from_redis(job_id: str) -> None:
|
||||
try:
|
||||
r = _get_redis()
|
||||
r.delete(
|
||||
f"{VEHL_IMPORT_FILE_PREFIX}{job_id}",
|
||||
f"{VEHL_IMPORT_META_PREFIX}{job_id}",
|
||||
f"{VEHL_IMPORT_ERROR_LINES_PREFIX}{job_id}",
|
||||
f"{VEHL_IMPORT_STATUS_PREFIX}{job_id}",
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(f"Vehicles import: failed to delete Redis keys: {e}")
|
||||
|
||||
|
||||
def normalize_header(name: Optional[str]) -> str:
|
||||
if not name:
|
||||
return ""
|
||||
name = unicodedata.normalize("NFKD", str(name)).upper()
|
||||
name = "".join(ch for ch in name if not unicodedata.combining(ch))
|
||||
name = re.sub(r"[^A-Z0-9]+", " ", name)
|
||||
return re.sub(r"\s+", " ", name).strip()
|
||||
|
||||
|
||||
# Max lengths from Vehicle model (a76.vehicle)
|
||||
_MAX = {
|
||||
"vehicle_key": 14,
|
||||
"ace_vehicle_key": 10,
|
||||
"transporter_key": 23,
|
||||
"transport_identifier": 30,
|
||||
"transport_type": 2,
|
||||
"entity_code": 1,
|
||||
"transponder_number": 16,
|
||||
"dot_number": 8,
|
||||
"plate_number": 17,
|
||||
"city": 30,
|
||||
"state": 30,
|
||||
"country": 3,
|
||||
"seal": 49,
|
||||
"insurance_company_name": 30,
|
||||
"insurance_number": 20,
|
||||
"series": 30,
|
||||
}
|
||||
|
||||
|
||||
def _parse_decimal(val: Any) -> Optional[Decimal]:
|
||||
if val is None or (isinstance(val, str) and not val.strip()):
|
||||
return None
|
||||
try:
|
||||
s = str(val).strip().replace(",", "")
|
||||
return Decimal(s)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def _parse_insurance_date(val: Any) -> Optional[int]:
|
||||
"""Parse FECHA DE ASEGURADORA to integer yyyymmdd. Tolerates DD/MM/YYYY, YYYY-MM-DD, or YYYYMMDD."""
|
||||
if val is None or (isinstance(val, str) and not val.strip()):
|
||||
return None
|
||||
s = str(val).strip()
|
||||
if not s:
|
||||
return None
|
||||
# Already integer-like
|
||||
if re.match(r"^\d{8}$", s):
|
||||
try:
|
||||
return int(s)
|
||||
except ValueError:
|
||||
pass
|
||||
# Try DD/MM/YYYY or similar
|
||||
for sep in ["/", "-", "."]:
|
||||
if sep in s:
|
||||
parts = s.split(sep)
|
||||
if len(parts) == 3:
|
||||
try:
|
||||
a, b, c = [p.strip() for p in parts]
|
||||
if len(c) == 4 and len(a) <= 2 and len(b) <= 2: # c=year
|
||||
return int(c) * 10000 + int(b) * 100 + int(a)
|
||||
if len(a) == 4 and len(b) <= 2 and len(c) <= 2: # a=year
|
||||
return int(a) * 10000 + int(b) * 100 + int(c)
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
break
|
||||
return None
|
||||
|
||||
|
||||
def _validate_row_vehicle(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
|
||||
"""Valida una fila para Vehículo. Retorna error dict o None."""
|
||||
clave = (row.get("CLAVE") or "").strip()
|
||||
if not clave:
|
||||
return {"line": line_num, "col": "CLAVE", "msg": "Requerido"}
|
||||
if len(clave) > _MAX["vehicle_key"]:
|
||||
return {"line": line_num, "col": "CLAVE", "msg": f"Máximo {_MAX['vehicle_key']} caracteres"}
|
||||
|
||||
# Optional fields: max lengths only
|
||||
for col, max_len in [
|
||||
("CLAVE ACE", _MAX["ace_vehicle_key"]),
|
||||
("CLAVE TRANSPORTE", _MAX["transporter_key"]),
|
||||
("VIN", _MAX["series"]),
|
||||
("TIPO TRANSPORTE", _MAX["transport_type"]),
|
||||
("CODIGO DE ENTIDAD", _MAX["entity_code"]),
|
||||
("TRANSPONDEDOR", _MAX["transponder_number"]),
|
||||
("NUMERO DOT", _MAX["dot_number"]),
|
||||
("PLACAS", _MAX["plate_number"]),
|
||||
("CIUDAD", _MAX["city"]),
|
||||
("ESTADO", _MAX["state"]),
|
||||
("PAIS", _MAX["country"]),
|
||||
("PRECINTO", _MAX["seal"]),
|
||||
("EMPRESA ASEGURADORA", _MAX["insurance_company_name"]),
|
||||
("NUM. ASEGURADORA", _MAX["insurance_number"]),
|
||||
]:
|
||||
val = (row.get(col) or "").strip()
|
||||
if val and len(val) > max_len:
|
||||
return {"line": line_num, "col": col, "msg": f"Máximo {max_len} caracteres"}
|
||||
|
||||
# MONTO ASEGURADO: must be numeric if present
|
||||
monto = row.get("MONTO ASEGURADO") or row.get("MONTO")
|
||||
if monto is not None and str(monto).strip():
|
||||
if _parse_decimal(monto) is None:
|
||||
return {"line": line_num, "col": "MONTO ASEGURADO", "msg": "Debe ser numérico"}
|
||||
|
||||
# FECHA DE ASEGURADORA: optional; if present try parse (do not fail row if invalid, set None)
|
||||
# Plan says: "en caso de formato inválido, marcar error pero no rechazar toda la fila" -> we can either
|
||||
# reject or set None. We reject invalid date to keep data quality.
|
||||
fecha = row.get("FECHA DE ASEGURADORA") or row.get("FECHA ASEGURADORA")
|
||||
if fecha is not None and str(fecha).strip():
|
||||
if _parse_insurance_date(fecha) is None:
|
||||
return {"line": line_num, "col": "FECHA DE ASEGURADORA", "msg": "Formato de fecha inválido (use YYYYMMDD o DD/MM/YYYY)"}
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def _do_scan(
|
||||
job_id: str,
|
||||
progress_callback: Optional[Any] = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Lógica de escaneo (sin Celery). Usado por la tarea scan_file y por run_scan_sync.
|
||||
progress_callback(current, total, error_count) opcional.
|
||||
"""
|
||||
file_path = _ensure_worker_has_file_from_redis(job_id)
|
||||
if not file_path:
|
||||
return {"status": "failed", "error": "Archivo no encontrado (expirado o no subido). Sube de nuevo."}
|
||||
_ensure_worker_has_meta_from_redis(job_id, file_path)
|
||||
|
||||
error_dir = os.path.join(os.path.dirname(file_path).replace("temp", "errors"), "")
|
||||
os.makedirs(error_dir, exist_ok=True)
|
||||
error_path = os.path.join(error_dir, f"veh_{job_id}.jsonl")
|
||||
|
||||
total_rows = 0
|
||||
try:
|
||||
with open(file_path, "r", encoding="utf-8-sig") as f:
|
||||
total_rows = sum(1 for _ in f) - 1
|
||||
except Exception as e:
|
||||
return {"status": "failed", "error": str(e)}
|
||||
|
||||
meta_path = file_path.replace(".csv", ".meta.json")
|
||||
meta = {}
|
||||
if os.path.exists(meta_path):
|
||||
try:
|
||||
with open(meta_path, "r", encoding="utf-8") as f:
|
||||
meta = json.load(f) or {}
|
||||
except Exception as e:
|
||||
logger.warning(f"Vehicles import: failed to read meta: {e}")
|
||||
|
||||
tenant_id = meta.get("tenant_id")
|
||||
company_id = meta.get("company_id")
|
||||
if not tenant_id or not company_id:
|
||||
return {"status": "failed", "error": "Falta contexto (tenant/company)"}
|
||||
|
||||
error_count = 0
|
||||
processed_rows = 0
|
||||
errors_detail: List[Dict[str, Any]] = []
|
||||
|
||||
try:
|
||||
with open(file_path, "r", encoding="utf-8-sig") as f_in, open(
|
||||
error_path, "w", encoding="utf-8"
|
||||
) as f_err:
|
||||
sample = f_in.read(2048)
|
||||
f_in.seek(0)
|
||||
try:
|
||||
dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
|
||||
except Exception:
|
||||
dialect = "excel"
|
||||
reader = csv.DictReader(f_in, dialect=dialect)
|
||||
|
||||
for i, row in enumerate(reader, start=1):
|
||||
if progress_callback and i % 500 == 0:
|
||||
progress_callback(i, total_rows, error_count)
|
||||
|
||||
row_norm = row_from_template(row, normalize_header)
|
||||
err = _validate_row_vehicle(row_norm, i)
|
||||
if err:
|
||||
error_count += 1
|
||||
f_err.write(json.dumps(err) + "\n")
|
||||
if len(errors_detail) < 500:
|
||||
errors_detail.append(
|
||||
{"line": err["line"], "col": err.get("col", ""), "msg": err.get("msg", "")}
|
||||
)
|
||||
processed_rows += 1
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Vehicles import scan failed: {e}")
|
||||
return {"status": "failed", "error": str(e)}
|
||||
|
||||
error_lines_list = []
|
||||
try:
|
||||
if os.path.exists(error_path):
|
||||
with open(error_path, "r", encoding="utf-8") as f:
|
||||
for line in f:
|
||||
try:
|
||||
err = json.loads(line)
|
||||
if "line" in err:
|
||||
error_lines_list.append(err["line"])
|
||||
except Exception:
|
||||
pass
|
||||
if error_lines_list:
|
||||
r = _get_redis()
|
||||
r.set(
|
||||
f"{VEHL_IMPORT_ERROR_LINES_PREFIX}{job_id}",
|
||||
json.dumps(error_lines_list).encode("utf-8"),
|
||||
ex=VEHL_IMPORT_REDIS_TTL,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(f"Vehicles import: failed to store error lines in Redis: {e}")
|
||||
|
||||
result = {
|
||||
"status": "waiting_confirmation",
|
||||
"job_id": job_id,
|
||||
"total_rows": processed_rows,
|
||||
"error_count": error_count,
|
||||
"valid_rows": processed_rows - error_count,
|
||||
"errors": errors_detail,
|
||||
}
|
||||
return result
|
||||
|
||||
|
||||
def run_scan_sync(job_id: str) -> Dict[str, Any]:
|
||||
"""
|
||||
Ejecuta el escaneo en el proceso actual y guarda el resultado en Redis.
|
||||
Usado desde el endpoint de upload en un hilo cuando no hay worker de Celery.
|
||||
"""
|
||||
result = _do_scan(job_id, progress_callback=None)
|
||||
try:
|
||||
r = _get_redis()
|
||||
r.set(
|
||||
f"{VEHL_IMPORT_STATUS_PREFIX}{job_id}",
|
||||
json.dumps(result).encode("utf-8"),
|
||||
ex=VEHL_IMPORT_REDIS_TTL,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(f"Vehicles import: failed to store scan status in Redis: {e}")
|
||||
return result
|
||||
|
||||
|
||||
@celery_app.task(bind=True)
|
||||
def scan_file(self, job_id: str, config: str = None):
|
||||
"""
|
||||
Fase 1: Leer CSV, validar filas, escribir errores en JSONL.
|
||||
Devuelve waiting_confirmation con total_rows, error_count, valid_rows, errors.
|
||||
"""
|
||||
logger.info(f"Vehicles import: starting scan for job {job_id}")
|
||||
|
||||
def on_progress(current: int, total: int, errors: int) -> None:
|
||||
self.update_state(
|
||||
state="PROGRESS",
|
||||
meta={"current": current, "total": total, "errors": errors},
|
||||
)
|
||||
|
||||
result = _do_scan(job_id, progress_callback=on_progress)
|
||||
try:
|
||||
r = _get_redis()
|
||||
r.set(
|
||||
f"{VEHL_IMPORT_STATUS_PREFIX}{job_id}",
|
||||
json.dumps(result).encode("utf-8"),
|
||||
ex=VEHL_IMPORT_REDIS_TTL,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(f"Vehicles import: failed to store scan status in Redis: {e}")
|
||||
return result
|
||||
|
||||
|
||||
def _str_or_none(val: Any, max_len: Optional[int] = None) -> Optional[str]:
|
||||
if val is None:
|
||||
return None
|
||||
s = str(val).strip()
|
||||
if not s:
|
||||
return None
|
||||
if max_len and len(s) > max_len:
|
||||
return s[:max_len]
|
||||
return s
|
||||
|
||||
|
||||
def _row_to_vehicle_dto(row: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Build dict suitable for VehicleCreateDTO / VehicleUpdateDTO from normalized row."""
|
||||
vehicle_key = _str_or_none(row.get("CLAVE"), _MAX["vehicle_key"])
|
||||
if not vehicle_key:
|
||||
return {}
|
||||
data = {
|
||||
"vehicle_key": vehicle_key,
|
||||
"ace_vehicle_key": _str_or_none(row.get("CLAVE ACE"), _MAX["ace_vehicle_key"]),
|
||||
"transporter_key": _str_or_none(row.get("CLAVE TRANSPORTE"), _MAX["transporter_key"]),
|
||||
"series": _str_or_none(row.get("VIN"), _MAX["series"]),
|
||||
"transport_type": _str_or_none(row.get("TIPO TRANSPORTE"), _MAX["transport_type"]),
|
||||
"entity_code": _str_or_none(row.get("CODIGO DE ENTIDAD"), _MAX["entity_code"]),
|
||||
"transponder_number": _str_or_none(row.get("TRANSPONDEDOR"), _MAX["transponder_number"]),
|
||||
"dot_number": _str_or_none(row.get("NUMERO DOT"), _MAX["dot_number"]),
|
||||
"plate_number": _str_or_none(row.get("PLACAS"), _MAX["plate_number"]),
|
||||
"city": _str_or_none(row.get("CIUDAD"), _MAX["city"]),
|
||||
"state": _str_or_none(row.get("ESTADO"), _MAX["state"]),
|
||||
"country": _str_or_none(row.get("PAIS"), _MAX["country"]),
|
||||
"seal": _str_or_none(row.get("PRECINTO"), _MAX["seal"]),
|
||||
"insurance_company_name": _str_or_none(row.get("EMPRESA ASEGURADORA"), _MAX["insurance_company_name"]),
|
||||
"insurance_number": _str_or_none(row.get("NUM. ASEGURADORA"), _MAX["insurance_number"]),
|
||||
"insurance_amount": _parse_decimal(row.get("MONTO ASEGURADO") or row.get("MONTO")),
|
||||
"insurance_date": _parse_insurance_date(row.get("FECHA DE ASEGURADORA") or row.get("FECHA ASEGURADORA")),
|
||||
}
|
||||
return data
|
||||
|
||||
|
||||
def _do_commit(job_id: str) -> Dict[str, Any]:
|
||||
"""
|
||||
Lógica de commit (inserción/actualización). Usado por la tarea insert_valid_rows y por run_commit_sync.
|
||||
"""
|
||||
file_path = _ensure_worker_has_file_from_redis(job_id)
|
||||
if not file_path:
|
||||
alt_path = os.path.join(_worker_upload_dir(), f"veh_{job_id}.csv")
|
||||
if not os.path.exists(alt_path):
|
||||
return {
|
||||
"status": "failed",
|
||||
"error": "Archivo no encontrado (expirado). Sube y confirma de nuevo.",
|
||||
}
|
||||
file_path = alt_path
|
||||
else:
|
||||
_ensure_worker_has_meta_from_redis(job_id, file_path)
|
||||
|
||||
base_dir = os.path.dirname(file_path)
|
||||
error_dir = base_dir.replace("temp", "errors")
|
||||
error_path = os.path.join(error_dir, f"veh_{job_id}.jsonl")
|
||||
|
||||
error_lines = set()
|
||||
try:
|
||||
r = _get_redis()
|
||||
raw = r.get(f"{VEHL_IMPORT_ERROR_LINES_PREFIX}{job_id}")
|
||||
if raw:
|
||||
error_lines = set(json.loads(raw.decode("utf-8")))
|
||||
except Exception as e:
|
||||
logger.debug(f"Vehicles import: could not load error lines from Redis: {e}")
|
||||
if not error_lines and os.path.exists(error_path):
|
||||
with open(error_path, "r", encoding="utf-8") as f:
|
||||
for line in f:
|
||||
try:
|
||||
err = json.loads(line)
|
||||
error_lines.add(err["line"])
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
meta_path = file_path.replace(".csv", ".meta.json")
|
||||
tenant_id = None
|
||||
company_id = None
|
||||
meta = {}
|
||||
if os.path.exists(meta_path):
|
||||
try:
|
||||
with open(meta_path, "r", encoding="utf-8") as f:
|
||||
meta = json.load(f) or {}
|
||||
tenant_id = meta.get("tenant_id")
|
||||
company_id = meta.get("company_id")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if not tenant_id or not company_id:
|
||||
return {"status": "failed", "error": "Falta contexto (tenant/company)"}
|
||||
|
||||
from api.v1.modules.a76.transportation.vehicles.services import VehicleService
|
||||
from api.v1.modules.a76.transportation.vehicles.dto import VehicleCreateDTO, VehicleUpdateDTO
|
||||
|
||||
inserted_count = 0
|
||||
updated_count = 0
|
||||
skipped_invalid = 0
|
||||
skipped_duplicate = 0
|
||||
skipped_details: List[Dict[str, Any]] = []
|
||||
seen_keys_in_file: Dict[str, int] = {}
|
||||
|
||||
try:
|
||||
with CoreSessionLocal() as session:
|
||||
with open(file_path, "r", encoding="utf-8-sig") as f:
|
||||
sample = f.read(2048)
|
||||
f.seek(0)
|
||||
try:
|
||||
dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
|
||||
except Exception:
|
||||
dialect = "excel"
|
||||
reader = csv.DictReader(f, dialect=dialect)
|
||||
|
||||
for i, row in enumerate(reader, start=1):
|
||||
if i in error_lines:
|
||||
continue
|
||||
|
||||
row_norm = row_from_template(row, normalize_header)
|
||||
err = _validate_row_vehicle(row_norm, i)
|
||||
if err:
|
||||
skipped_invalid += 1
|
||||
skipped_details.append(
|
||||
{
|
||||
"line": i,
|
||||
"vehicle_key": (row_norm.get("CLAVE") or "").strip()[:14] or "-",
|
||||
"invoice": (row_norm.get("CLAVE") or "").strip()[:14] or "-",
|
||||
"reason": f"{err.get('col', '')}: {err.get('msg', '')}",
|
||||
}
|
||||
)
|
||||
continue
|
||||
|
||||
data = _row_to_vehicle_dto(row_norm)
|
||||
if not data or not data.get("vehicle_key"):
|
||||
skipped_invalid += 1
|
||||
continue
|
||||
|
||||
vk = data["vehicle_key"]
|
||||
if vk in seen_keys_in_file:
|
||||
skipped_duplicate += 1
|
||||
skipped_details.append(
|
||||
{"line": i, "vehicle_key": vk, "invoice": vk, "reason": "Clave duplicada en el archivo (se usa la primera)"}
|
||||
)
|
||||
continue
|
||||
seen_keys_in_file[vk] = i
|
||||
|
||||
existing = VehicleService.get_by_id(session, vk, tenant_id, company_id)
|
||||
try:
|
||||
if existing:
|
||||
update_data = VehicleUpdateDTO(**{k: v for k, v in data.items() if k != "vehicle_key"})
|
||||
VehicleService.update(session, vk, tenant_id, update_data, company_id)
|
||||
updated_count += 1
|
||||
else:
|
||||
create_data = VehicleCreateDTO(**data)
|
||||
VehicleService.create(session, create_data, tenant_id, company_id)
|
||||
inserted_count += 1
|
||||
except Exception as db_err:
|
||||
session.rollback()
|
||||
skipped_invalid += 1
|
||||
skipped_details.append(
|
||||
{"line": i, "vehicle_key": vk, "invoice": vk, "reason": str(db_err)}
|
||||
)
|
||||
continue
|
||||
|
||||
try:
|
||||
session.commit()
|
||||
except Exception as db_err:
|
||||
session.rollback()
|
||||
logger.error(f"Vehicles import DB error: {db_err}")
|
||||
return {"status": "failed", "error": str(db_err)}
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Vehicles import task failed: {e}")
|
||||
import traceback
|
||||
logger.error(traceback.format_exc())
|
||||
return {"status": "failed", "error": str(e)}
|
||||
|
||||
try:
|
||||
if file_path and os.path.exists(file_path):
|
||||
os.remove(file_path)
|
||||
if os.path.exists(error_path):
|
||||
os.remove(error_path)
|
||||
if os.path.exists(meta_path):
|
||||
os.remove(meta_path)
|
||||
_delete_import_from_redis(job_id)
|
||||
except Exception as cleanup_err:
|
||||
logger.warning(f"Vehicles import cleanup failed: {cleanup_err}")
|
||||
|
||||
total_ok = inserted_count + updated_count
|
||||
if total_ok == 0 and (skipped_invalid + skipped_duplicate) > 0:
|
||||
return {
|
||||
"status": "warning",
|
||||
"inserted": inserted_count,
|
||||
"updated": updated_count,
|
||||
"skipped_invalid": skipped_invalid,
|
||||
"skipped_duplicate": skipped_duplicate,
|
||||
"skipped_missing_fk": 0,
|
||||
"skipped_details": skipped_details,
|
||||
"message": f"No se insertaron registros. {skipped_invalid + skipped_duplicate} rechazados.",
|
||||
}
|
||||
if total_ok == 0:
|
||||
return {
|
||||
"status": "failed",
|
||||
"error": "No hay registros válidos en el archivo CSV",
|
||||
"inserted": 0,
|
||||
"updated": 0,
|
||||
"skipped_invalid": skipped_invalid,
|
||||
"skipped_duplicate": skipped_duplicate,
|
||||
"skipped_missing_fk": 0,
|
||||
"skipped_details": skipped_details,
|
||||
}
|
||||
return {
|
||||
"status": "finished",
|
||||
"inserted": inserted_count,
|
||||
"updated": updated_count,
|
||||
"skipped_invalid": skipped_invalid,
|
||||
"skipped_duplicate": skipped_duplicate,
|
||||
"skipped_missing_fk": 0,
|
||||
"skipped_details": skipped_details,
|
||||
}
|
||||
|
||||
|
||||
def run_commit_sync(job_id: str) -> Dict[str, Any]:
|
||||
"""
|
||||
Ejecuta el commit en el proceso actual y guarda el resultado en Redis.
|
||||
Usado desde el endpoint de commit en un hilo cuando no hay worker de Celery.
|
||||
"""
|
||||
result = _do_commit(job_id)
|
||||
try:
|
||||
r = _get_redis()
|
||||
r.set(
|
||||
f"{VEHL_IMPORT_STATUS_PREFIX}{job_id}",
|
||||
json.dumps(result).encode("utf-8"),
|
||||
ex=VEHL_IMPORT_REDIS_TTL,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(f"Vehicles import: failed to store commit status in Redis: {e}")
|
||||
return result
|
||||
|
||||
|
||||
@celery_app.task(bind=True)
|
||||
def insert_valid_rows(self, job_id: str):
|
||||
"""
|
||||
Fase 2: Re-leer CSV, omitir filas con error, create/update via VehicleService.
|
||||
"""
|
||||
logger.info(f"Vehicles import: starting commit for job {job_id}")
|
||||
result = _do_commit(job_id)
|
||||
try:
|
||||
r = _get_redis()
|
||||
r.set(
|
||||
f"{VEHL_IMPORT_STATUS_PREFIX}{job_id}",
|
||||
json.dumps(result).encode("utf-8"),
|
||||
ex=VEHL_IMPORT_REDIS_TTL,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(f"Vehicles import: failed to store commit status in Redis: {e}")
|
||||
return result
|
||||
@@ -0,0 +1,53 @@
|
||||
"""
|
||||
Configuración de plantilla CSV para Vehículos / Transportes (EstructuraCatTransportes.xls).
|
||||
Mapeo: CLAVE → vehicle_key, CLAVE ACE → ace_vehicle_key, etc.
|
||||
"""
|
||||
|
||||
from typing import Dict, List, Any
|
||||
|
||||
TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
|
||||
"vehicles": [
|
||||
{"canonical": "CLAVE", "aliases": ["CLAVE VEHICULO", "VEHICLE KEY", "KEY"]},
|
||||
{"canonical": "CLAVE ACE", "aliases": ["ACE", "CLAVEACE"]},
|
||||
{"canonical": "CLAVE TRANSPORTE", "aliases": ["TRANSPORTE", "TRANSPORTISTA"]},
|
||||
{"canonical": "VIN", "aliases": ["SERIE", "SERIES", "NUMERO SERIE"]},
|
||||
{"canonical": "TIPO TRANSPORTE", "aliases": ["TIPO", "TRANSPORT TYPE"]},
|
||||
{"canonical": "CODIGO DE ENTIDAD", "aliases": ["ENTIDAD", "ENTITY CODE"]},
|
||||
{"canonical": "TRANSPONDEDOR", "aliases": ["TRANSPONDER", "TRANSPONDER NUMBER"]},
|
||||
{"canonical": "NUMERO DOT", "aliases": ["DOT", "NUM DOT"]},
|
||||
{"canonical": "PLACAS", "aliases": ["PLACA", "PLATE", "PLATE NUMBER"]},
|
||||
{"canonical": "CIUDAD", "aliases": ["CITY"]},
|
||||
{"canonical": "ESTADO", "aliases": ["STATE"]},
|
||||
{"canonical": "PAIS", "aliases": ["COUNTRY", "Pais"]},
|
||||
{"canonical": "PRECINTO", "aliases": ["SEAL"]},
|
||||
{"canonical": "EMPRESA ASEGURADORA", "aliases": ["ASEGURADORA", "INSURANCE COMPANY"]},
|
||||
{"canonical": "NUM. ASEGURADORA", "aliases": ["NUM ASEGURADORA", "POLIZA", "INSURANCE NUMBER"]},
|
||||
{"canonical": "MONTO ASEGURADO", "aliases": ["MONTO", "INSURANCE AMOUNT"]},
|
||||
{"canonical": "FECHA DE ASEGURADORA", "aliases": ["FECHA ASEGURADORA", "INSURANCE DATE"]},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]:
|
||||
cols = TEMPLATE_COLUMNS.get("vehicles")
|
||||
if not cols:
|
||||
return {}
|
||||
lookup: Dict[str, str] = {}
|
||||
for item in cols:
|
||||
canonical = item["canonical"]
|
||||
lookup[normalize_header_fn(canonical)] = canonical
|
||||
for alias in item.get("aliases") or []:
|
||||
lookup[normalize_header_fn(alias)] = canonical
|
||||
return lookup
|
||||
|
||||
|
||||
def row_from_template(row: Dict[str, Any], normalize_header_fn) -> Dict[str, Any]:
|
||||
lookup = build_normalized_lookup(normalize_header_fn)
|
||||
if not lookup:
|
||||
return {normalize_header_fn(k): v for k, v in row.items()}
|
||||
out: Dict[str, Any] = {}
|
||||
for csv_header, value in row.items():
|
||||
key_norm = normalize_header_fn(csv_header)
|
||||
if key_norm in lookup:
|
||||
out[lookup[key_norm]] = value
|
||||
return out
|
||||
@@ -1,11 +1,19 @@
|
||||
from fastapi import APIRouter
|
||||
|
||||
from api.v1.common.tenant_crud_routes import TenantCRUDRoutes
|
||||
|
||||
from .dto import VehicleCreateDTO, VehicleResponseDTO, VehicleUpdateDTO
|
||||
from .services import VehicleService
|
||||
from .imports.routes import router as imports_router
|
||||
|
||||
# Create router using TenantCRUDRoutes factory
|
||||
# Note: vehicle_key is a string (not int) and is used as the primary key
|
||||
router = TenantCRUDRoutes(
|
||||
# Main router: vehicles CRUD + CSV imports
|
||||
router = APIRouter()
|
||||
|
||||
# CSV import (upload → scan → status → commit)
|
||||
router.include_router(imports_router, prefix="/vehicles/imports", tags=["a76 / vehicles / csv_import"])
|
||||
|
||||
# CRUD routes
|
||||
crud_router = TenantCRUDRoutes(
|
||||
service=VehicleService,
|
||||
create_schema=VehicleCreateDTO,
|
||||
update_schema=VehicleUpdateDTO,
|
||||
@@ -13,10 +21,11 @@ router = TenantCRUDRoutes(
|
||||
prefix="/vehicles",
|
||||
tags=[],
|
||||
resource_name="Vehicle",
|
||||
id_name="vehicle_key", # Using vehicle_key instead of numeric ID
|
||||
id_type=str, # Specify that the ID is a string
|
||||
enable_list=True, # Enable GET /vehicles with pagination
|
||||
enable_filters=True, # Enable filtering by plate_number and transport_type
|
||||
id_name="vehicle_key",
|
||||
id_type=str,
|
||||
enable_list=True,
|
||||
enable_filters=True,
|
||||
default_page_size=50,
|
||||
max_page_size=100,
|
||||
).router
|
||||
router.include_router(crud_router)
|
||||
|
||||
Reference in New Issue
Block a user