diff --git a/backend/api/v1/modules/a76/transportation/drivers/dto.py b/backend/api/v1/modules/a76/transportation/drivers/dto.py index f07e4b4c..bfb3d8fc 100644 --- a/backend/api/v1/modules/a76/transportation/drivers/dto.py +++ b/backend/api/v1/modules/a76/transportation/drivers/dto.py @@ -6,28 +6,28 @@ from pydantic import BaseModel class DriverBaseDTO(BaseModel): transporter_key: str line: int - driver_name: Optional[str] - license_number: Optional[str] - express_line_id: Optional[str] - ace_id: Optional[str] - birth_date: Optional[int] - gender: Optional[str] - birth_country: Optional[str] - hazardous_material_auth: Optional[str] - hazardous_material_state: Optional[str] - first_name: Optional[str] - last_name: Optional[str] - id_key1: Optional[str] - id_number1: Optional[str] - id_state1: Optional[str] - id_country1: Optional[str] - id_key2: Optional[str] - id_number2: Optional[str] - id_state2: Optional[str] - id_country2: Optional[str] - badge_number: Optional[str] - class_type: Optional[str] - unique_badge_number: Optional[str] + driver_name: Optional[str] = None + license_number: Optional[str] = None + express_line_id: Optional[str] = None + ace_id: Optional[str] = None + birth_date: Optional[int] = None + gender: Optional[str] = None + birth_country: Optional[str] = None + hazardous_material_auth: Optional[str] = None + hazardous_material_state: Optional[str] = None + first_name: Optional[str] = None + last_name: Optional[str] = None + id_key1: Optional[str] = None + id_number1: Optional[str] = None + id_state1: Optional[str] = None + id_country1: Optional[str] = None + id_key2: Optional[str] = None + id_number2: Optional[str] = None + id_state2: Optional[str] = None + id_country2: Optional[str] = None + badge_number: Optional[str] = None + class_type: Optional[str] = None + unique_badge_number: Optional[str] = None company_id: int tenant_id: int diff --git a/backend/api/v1/modules/a76/transportation/drivers/imports/__init__.py b/backend/api/v1/modules/a76/transportation/drivers/imports/__init__.py new file mode 100644 index 00000000..98bef45c --- /dev/null +++ b/backend/api/v1/modules/a76/transportation/drivers/imports/__init__.py @@ -0,0 +1 @@ +# CSV import for Drivers / Conductores (upload -> scan -> status -> commit). diff --git a/backend/api/v1/modules/a76/transportation/drivers/imports/routes.py b/backend/api/v1/modules/a76/transportation/drivers/imports/routes.py new file mode 100644 index 00000000..ea5d2eb4 --- /dev/null +++ b/backend/api/v1/modules/a76/transportation/drivers/imports/routes.py @@ -0,0 +1,188 @@ +""" +Rutas de importacion CSV para Conductores. +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, + DRV_IMPORT_FILE_PREFIX, + DRV_IMPORT_META_PREFIX, + DRV_IMPORT_STATUS_PREFIX, + DRV_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"Drivers 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": "drivers", + } + + try: + r = _get_redis() + r.set( + f"{DRV_IMPORT_FILE_PREFIX}{job_id}", + base64.b64encode(contents), + ex=DRV_IMPORT_REDIS_TTL, + ) + r.set( + f"{DRV_IMPORT_META_PREFIX}{job_id}", + json.dumps(meta_data).encode("utf-8"), + ex=DRV_IMPORT_REDIS_TTL, + ) + except Exception as e: + logger.error(f"Drivers 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"drv_{job_id}.csv"), "wb") as f: + f.write(contents) + with open(os.path.join(upload_dir, f"drv_{job_id}.meta.json"), "w") as f: + json.dump(meta_data, f) + except Exception as e: + logger.warning(f"Drivers import: local file save failed: {e}") + + scan_file.apply_async(args=[job_id], task_id=job_id) + + def run_scan_background(): + try: + run_scan_sync(job_id) + except Exception as e: + logger.exception(f"Drivers 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): + try: + r = _get_redis() + raw = r.get(f"{DRV_IMPORT_STATUS_PREFIX}{job_id}") + if raw: + data = json.loads(raw.decode("utf-8")) + return data + except Exception as e: + logger.debug(f"Drivers 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("Drivers 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): + try: + r = _get_redis() + r.set( + f"{DRV_IMPORT_STATUS_PREFIX}{job_id}", + json.dumps({"status": "processing", "message": "Insertando..."}).encode("utf-8"), + ex=DRV_IMPORT_REDIS_TTL, + ) + except Exception as e: + logger.debug(f"Drivers import: could not write processing status: {e}") + + def run_commit_background(): + try: + run_commit_sync(job_id) + except Exception as e: + logger.exception(f"Drivers import: background commit failed: {e}") + + threading.Thread(target=run_commit_background, daemon=True).start() + + return { + "status": "committing", + "message": "Insercion iniciada.", + "commit_job_id": job_id, + } diff --git a/backend/api/v1/modules/a76/transportation/drivers/imports/schemas.py b/backend/api/v1/modules/a76/transportation/drivers/imports/schemas.py new file mode 100644 index 00000000..c5d2c2cc --- /dev/null +++ b/backend/api/v1/modules/a76/transportation/drivers/imports/schemas.py @@ -0,0 +1,23 @@ +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 + updated: Optional[int] = 0 + skipped_invalid: Optional[int] = 0 + skipped_missing_fk: Optional[int] = 0 + skipped_duplicate: Optional[int] = 0 + skipped_details: Optional[list] = None diff --git a/backend/api/v1/modules/a76/transportation/drivers/imports/tasks.py b/backend/api/v1/modules/a76/transportation/drivers/imports/tasks.py new file mode 100644 index 00000000..9234776e --- /dev/null +++ b/backend/api/v1/modules/a76/transportation/drivers/imports/tasks.py @@ -0,0 +1,640 @@ +""" +Tareas Celery para importacion CSV de Conductores. +Flujo: scan_file (validacion) -> insert_valid_rows (commit). +""" +import os +import base64 +import csv +import json +import logging +import re +import unicodedata +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__) + +DRV_IMPORT_FILE_PREFIX = "drv_import_file:" +DRV_IMPORT_META_PREFIX = "drv_import_meta:" +DRV_IMPORT_ERROR_LINES_PREFIX = "drv_import_error_lines:" +DRV_IMPORT_STATUS_PREFIX = "drv_import_status:" +DRV_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"{DRV_IMPORT_FILE_PREFIX}{job_id}") + if not data: + return None + try: + raw = base64.b64decode(data) + except Exception as e: + logger.warning(f"Drivers 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"drv_{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"{DRV_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"Drivers 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"{DRV_IMPORT_FILE_PREFIX}{job_id}", + f"{DRV_IMPORT_META_PREFIX}{job_id}", + f"{DRV_IMPORT_ERROR_LINES_PREFIX}{job_id}", + f"{DRV_IMPORT_STATUS_PREFIX}{job_id}", + ) + except Exception as e: + logger.warning(f"Drivers 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 = { + "transporter_key": 5, + "driver_name": 80, + "license_number": 29, + "express_line_id": 17, + "ace_id": 20, + "gender": 1, + "birth_country": 3, + "hazardous_material_auth": 2, + "hazardous_material_state": 30, + "first_name": 20, + "last_name": 20, + "id_key1": 40, + "id_number1": 20, + "id_state1": 30, + "id_country1": 3, + "id_key2": 40, + "id_number2": 20, + "id_state2": 30, + "id_country2": 3, + "badge_number": 20, + "class_type": 1, + "unique_badge_number": 100, +} + + +def _parse_int(val: Any) -> Optional[int]: + if val is None or (isinstance(val, str) and not val.strip()): + return None + s = str(val).strip() + if re.match(r"^\d+$", s): + return int(s) + if re.match(r"^\d+\.0+$", s): + try: + return int(float(s)) + except ValueError: + return None + return None + + +def _parse_birth_date(val: Any) -> Optional[int]: + if val is None or (isinstance(val, str) and not val.strip()): + return None + s = str(val).strip() + if re.match(r"^\d{8}$", s): + try: + return int(s) + except ValueError: + return None + if re.match(r"^\d+\.0+$", s): + try: + return int(float(s)) + except ValueError: + return None + 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: + return int(c) * 10000 + int(b) * 100 + int(a) + if len(a) == 4 and len(b) <= 2 and len(c) <= 2: + return int(a) * 10000 + int(b) * 100 + int(c) + except (ValueError, TypeError): + return None + break + return None + + +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 _dedupe_headers(headers: List[str]) -> List[str]: + counts: Dict[str, int] = {} + unique: List[str] = [] + for header in headers: + name = str(header or "").strip() or "COL" + count = counts.get(name, 0) + 1 + counts[name] = count + if count == 1: + unique.append(name) + else: + unique.append(f"{name} {count}") + return unique + + +def _validate_row_driver(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]: + transporter_key = (row.get("TRANSPORTISTA") or "").strip() + if not transporter_key: + return {"line": line_num, "col": "TRANSPORTISTA", "msg": "Requerido"} + if len(transporter_key) > _MAX["transporter_key"]: + return {"line": line_num, "col": "TRANSPORTISTA", "msg": f"Maximo {_MAX['transporter_key']} caracteres"} + + line_val = _parse_int(row.get("LINEA")) + if line_val is None: + return {"line": line_num, "col": "LINEA", "msg": "Debe ser numerico"} + if line_val <= 0: + return {"line": line_num, "col": "LINEA", "msg": "Debe ser mayor a 0"} + + driver_name = (row.get("CLAVE CONDUCTOR") or "").strip() + if driver_name and len(driver_name) > _MAX["driver_name"]: + return {"line": line_num, "col": "CLAVE CONDUCTOR", "msg": f"Maximo {_MAX['driver_name']} caracteres"} + + for col, max_len in [ + ("LICENCIA", _MAX["license_number"]), + ("PERMISO LINEA EXPRESS", _MAX["express_line_id"]), + ("IDENTIFICACION ACE", _MAX["ace_id"]), + ("SEXO", _MAX["gender"]), + ("PAIS NACIMIENTO", _MAX["birth_country"]), + ("TRANSPORTA MAT. PELIGROSO?", _MAX["hazardous_material_auth"]), + ("PERMISO MAT. PELIGROSO", _MAX["hazardous_material_state"]), + ("NOMBRE(S)", _MAX["first_name"]), + ("APELLIDO PATERNO", _MAX["last_name"]), + ("FORMA IDENTIFICACION 1", _MAX["id_key1"]), + ("NUM. IDENTIFICACION 1", _MAX["id_number1"]), + ("ESTADO", _MAX["id_state1"]), + ("PAIS", _MAX["id_country1"]), + ("FORMA IDENTIFICACION 2", _MAX["id_key2"]), + ("NUM. IDENTIFICACION 2", _MAX["id_number2"]), + ("ESTADO 2", _MAX["id_state2"]), + ("PAIS 2", _MAX["id_country2"]), + ]: + val = (row.get(col) or "").strip() + if val and len(val) > max_len: + return {"line": line_num, "col": col, "msg": f"Maximo {max_len} caracteres"} + + fecha = row.get("FECHA NACIMIENTO") + if fecha is not None and str(fecha).strip(): + if _parse_birth_date(fecha) is None: + return { + "line": line_num, + "col": "FECHA NACIMIENTO", + "msg": "Formato de fecha invalido (use YYYYMMDD o DD/MM/YYYY)", + } + + return None + + +def _row_to_driver_dto(row: Dict[str, Any], tenant_id: int, company_id: int) -> Dict[str, Any]: + transporter_key = _str_or_none(row.get("TRANSPORTISTA"), _MAX["transporter_key"]) + line = _parse_int(row.get("LINEA")) + if not transporter_key or line is None: + return {} + data = { + "transporter_key": transporter_key, + "line": line, + "driver_name": _str_or_none(row.get("CLAVE CONDUCTOR"), _MAX["driver_name"]), + "license_number": _str_or_none(row.get("LICENCIA"), _MAX["license_number"]), + "express_line_id": _str_or_none(row.get("PERMISO LINEA EXPRESS"), _MAX["express_line_id"]), + "ace_id": _str_or_none(row.get("IDENTIFICACION ACE"), _MAX["ace_id"]), + "birth_date": _parse_birth_date(row.get("FECHA NACIMIENTO")), + "gender": _str_or_none(row.get("SEXO"), _MAX["gender"]), + "birth_country": _str_or_none(row.get("PAIS NACIMIENTO"), _MAX["birth_country"]), + "hazardous_material_auth": _str_or_none( + row.get("TRANSPORTA MAT. PELIGROSO?"), _MAX["hazardous_material_auth"] + ), + "hazardous_material_state": _str_or_none( + row.get("PERMISO MAT. PELIGROSO"), _MAX["hazardous_material_state"] + ), + "first_name": _str_or_none(row.get("NOMBRE(S)"), _MAX["first_name"]), + "last_name": _str_or_none(row.get("APELLIDO PATERNO"), _MAX["last_name"]), + "id_key1": _str_or_none(row.get("FORMA IDENTIFICACION 1"), _MAX["id_key1"]), + "id_number1": _str_or_none(row.get("NUM. IDENTIFICACION 1"), _MAX["id_number1"]), + "id_state1": _str_or_none(row.get("ESTADO"), _MAX["id_state1"]), + "id_country1": _str_or_none(row.get("PAIS"), _MAX["id_country1"]), + "id_key2": _str_or_none(row.get("FORMA IDENTIFICACION 2"), _MAX["id_key2"]), + "id_number2": _str_or_none(row.get("NUM. IDENTIFICACION 2"), _MAX["id_number2"]), + "id_state2": _str_or_none(row.get("ESTADO 2"), _MAX["id_state2"]), + "id_country2": _str_or_none(row.get("PAIS 2"), _MAX["id_country2"]), + "company_id": company_id, + "tenant_id": tenant_id, + } + return data + + +def _do_scan(job_id: str, progress_callback: Optional[Any] = None) -> Dict[str, Any]: + 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"drv_{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"Drivers 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.reader(f_in, dialect=dialect) + try: + headers = next(reader) + except StopIteration: + headers = [] + headers = _dedupe_headers(headers) + dict_reader = csv.DictReader(f_in, fieldnames=headers, dialect=dialect) + + for i, row in enumerate(dict_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_driver(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"Drivers 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"{DRV_IMPORT_ERROR_LINES_PREFIX}{job_id}", + json.dumps(error_lines_list).encode("utf-8"), + ex=DRV_IMPORT_REDIS_TTL, + ) + except Exception as e: + logger.warning(f"Drivers import: failed to store error lines: {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]: + result = _do_scan(job_id, progress_callback=None) + try: + r = _get_redis() + r.set( + f"{DRV_IMPORT_STATUS_PREFIX}{job_id}", + json.dumps(result).encode("utf-8"), + ex=DRV_IMPORT_REDIS_TTL, + ) + except Exception as e: + logger.warning(f"Drivers 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): + logger.info(f"Drivers 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"{DRV_IMPORT_STATUS_PREFIX}{job_id}", + json.dumps(result).encode("utf-8"), + ex=DRV_IMPORT_REDIS_TTL, + ) + except Exception as e: + logger.warning(f"Drivers import: failed to store scan status in Redis: {e}") + return result + + +def _do_commit(job_id: str) -> Dict[str, Any]: + file_path = _ensure_worker_has_file_from_redis(job_id) + if not file_path: + alt_path = os.path.join(_worker_upload_dir(), f"drv_{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"drv_{job_id}.jsonl") + + error_lines = set() + try: + r = _get_redis() + raw = r.get(f"{DRV_IMPORT_ERROR_LINES_PREFIX}{job_id}") + if raw: + error_lines = set(json.loads(raw.decode("utf-8"))) + except Exception as e: + logger.debug(f"Drivers 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.drivers.services import DriverService + from api.v1.modules.a76.transportation.drivers.dto import DriverCreateDTO + + 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.reader(f, dialect=dialect) + try: + headers = next(reader) + except StopIteration: + headers = [] + headers = _dedupe_headers(headers) + dict_reader = csv.DictReader(f, fieldnames=headers, dialect=dialect) + + for i, row in enumerate(dict_reader, start=1): + if i in error_lines: + continue + + row_norm = row_from_template(row, normalize_header) + err = _validate_row_driver(row_norm, i) + if err: + skipped_invalid += 1 + skipped_details.append( + { + "line": i, + "driver_key": (row_norm.get("CLAVE CONDUCTOR") or "").strip()[:80] or "-", + "invoice": (row_norm.get("CLAVE CONDUCTOR") or "").strip()[:80] or "-", + "reason": f"{err.get('col', '')}: {err.get('msg', '')}", + } + ) + continue + + data = _row_to_driver_dto(row_norm, tenant_id, company_id) + if not data or not data.get("transporter_key") or data.get("line") is None: + skipped_invalid += 1 + continue + + key = f"{data['transporter_key']}:{data['line']}" + if key in seen_keys_in_file: + skipped_duplicate += 1 + skipped_details.append( + { + "line": i, + "driver_key": key, + "invoice": key, + "reason": "Clave duplicada en el archivo (se usa la primera)", + } + ) + continue + seen_keys_in_file[key] = i + + existing = DriverService.get_driver_by_key_and_line( + session, data["transporter_key"], data["line"], str(company_id), tenant_id + ) + try: + if existing: + update_fields = {k: v for k, v in data.items() if k not in ("transporter_key", "line", "company_id", "tenant_id")} + for field, value in update_fields.items(): + setattr(existing, field, value) + session.add(existing) + updated_count += 1 + else: + create_data = DriverCreateDTO(**data) + DriverService.create_driver(session, create_data) + inserted_count += 1 + except Exception as db_err: + session.rollback() + skipped_invalid += 1 + skipped_details.append( + {"line": i, "driver_key": key, "invoice": key, "reason": str(db_err)} + ) + continue + + try: + session.commit() + except Exception as db_err: + session.rollback() + logger.error(f"Drivers import DB error: {db_err}") + return {"status": "failed", "error": str(db_err)} + + except Exception as e: + logger.error(f"Drivers 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"Drivers 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 validos 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]: + result = _do_commit(job_id) + try: + r = _get_redis() + r.set( + f"{DRV_IMPORT_STATUS_PREFIX}{job_id}", + json.dumps(result).encode("utf-8"), + ex=DRV_IMPORT_REDIS_TTL, + ) + except Exception as e: + logger.warning(f"Drivers import: failed to store commit status in Redis: {e}") + return result + + +@celery_app.task(bind=True) +def insert_valid_rows(self, job_id: str): + logger.info(f"Drivers import: starting commit for job {job_id}") + result = _do_commit(job_id) + try: + r = _get_redis() + r.set( + f"{DRV_IMPORT_STATUS_PREFIX}{job_id}", + json.dumps(result).encode("utf-8"), + ex=DRV_IMPORT_REDIS_TTL, + ) + except Exception as e: + logger.warning(f"Drivers import: failed to store commit status in Redis: {e}") + return result diff --git a/backend/api/v1/modules/a76/transportation/drivers/imports/template_config.py b/backend/api/v1/modules/a76/transportation/drivers/imports/template_config.py new file mode 100644 index 00000000..cd9a3194 --- /dev/null +++ b/backend/api/v1/modules/a76/transportation/drivers/imports/template_config.py @@ -0,0 +1,56 @@ +""" +Configuracion de plantilla CSV para Conductores (EstructuraCatConductor.xls). +""" + +from typing import Dict, List, Any + +TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = { + "drivers": [ + {"canonical": "TRANSPORTISTA", "aliases": ["TRANSPORTISTA CLAVE", "CLAVE TRANSPORTISTA", "TRANSPORTER"]}, + {"canonical": "LINEA", "aliases": ["LINE", "LINEA CONDUCTOR"]}, + {"canonical": "CLAVE CONDUCTOR", "aliases": ["CONDUCTOR", "DRIVER", "NOMBRE CONDUCTOR"]}, + {"canonical": "LICENCIA", "aliases": ["LICENSE", "LICENCE"]}, + {"canonical": "PERMISO LINEA EXPRESS", "aliases": ["LINEA EXPRESS", "LINEA EXPRESS ID", "EXPRESS LINE"]}, + {"canonical": "IDENTIFICACION ACE", "aliases": ["ACE", "ACE ID", "IDENTIFICACION ACE ID"]}, + {"canonical": "FECHA NACIMIENTO", "aliases": ["FECHA NAC", "BIRTH DATE", "BIRTHDATE"]}, + {"canonical": "SEXO", "aliases": ["GENERO", "GENDER"]}, + {"canonical": "PAIS NACIMIENTO", "aliases": ["PAIS Nacimiento", "BIRTH COUNTRY"]}, + {"canonical": "TRANSPORTA MAT. PELIGROSO?", "aliases": ["MATERIAL PELIGROSO", "HAZMAT", "HAZARDOUS MATERIAL"]}, + {"canonical": "PERMISO MAT. PELIGROSO", "aliases": ["PERMISO MAT PELIGROSO", "HAZMAT PERMIT"]}, + {"canonical": "NOMBRE(S)", "aliases": ["NOMBRE", "NOMBRES", "FIRST NAME"]}, + {"canonical": "APELLIDO PATERNO", "aliases": ["APELLIDO", "LAST NAME", "APELLIDO P"]}, + {"canonical": "FORMA IDENTIFICACION 1", "aliases": ["FORMA IDENTIFICACION", "ID TIPO 1", "ID KEY 1"]}, + {"canonical": "NUM. IDENTIFICACION 1", "aliases": ["NUM IDENTIFICACION 1", "ID NUMERO 1", "ID NUMBER 1"]}, + {"canonical": "ESTADO", "aliases": ["ESTADO 1", "STATE 1"]}, + {"canonical": "PAIS", "aliases": ["PAIS 1", "COUNTRY 1"]}, + {"canonical": "FORMA IDENTIFICACION 2", "aliases": ["ID TIPO 2", "ID KEY 2"]}, + {"canonical": "NUM. IDENTIFICACION 2", "aliases": ["NUM IDENTIFICACION 2", "ID NUMERO 2", "ID NUMBER 2"]}, + {"canonical": "ESTADO 2", "aliases": ["STATE 2"]}, + {"canonical": "PAIS 2", "aliases": ["COUNTRY 2"]}, + ], +} + + +def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]: + cols = TEMPLATE_COLUMNS.get("drivers") + 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 diff --git a/backend/api/v1/modules/a76/transportation/drivers/routes.py b/backend/api/v1/modules/a76/transportation/drivers/routes.py index 981ab353..a07b5c55 100644 --- a/backend/api/v1/modules/a76/transportation/drivers/routes.py +++ b/backend/api/v1/modules/a76/transportation/drivers/routes.py @@ -8,9 +8,13 @@ from sqlalchemy.orm import Session from .dto import DriverCreateDTO, DriverResponseDTO from .models import Driver from .services import DriverService +from .imports.routes import router as imports_router router = APIRouter(prefix="/drivers") +# CSV import (upload -> scan -> status -> commit) +router.include_router(imports_router, prefix="/imports", tags=["a76 / drivers / csv_import"]) + @router.get("/", response_model=Dict[str, Any]) async def list_drivers( diff --git a/frontend/package.json b/frontend/package.json index b21c873b..e7cd1887 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -64,5 +64,6 @@ "lucide-svelte": "^0.553.0", "marked": "^12.0.0", "svelte-sonner": "^1.0.7" - } -} \ No newline at end of file + }, + "packageManager": "pnpm@10.30.3+sha512.c961d1e0a2d8e354ecaa5166b822516668b7f44cb5bd95122d590dd81922f606f5473b6d23ec4a5be05e7fcd18e8488d47d978bbe981872f1145d06e9a740017" +} diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index e7669c73..fde71cfe 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -479,6 +479,20 @@ export const api = { api.post(`/v1/a76/transportation/vehicles/imports/${jobId}/commit`, {}) }, + // CSV import for Conductores (drivers/imports) + driverImports: { + upload: (file: File, companyId: number) => { + const formData = new FormData(); + formData.append('file', file); + return fetchApi( + `/v1/a76/drivers/imports/upload?company_id=${companyId}`, + { method: 'POST', body: formData } + ); + }, + status: (jobId: string) => api.get(`/v1/a76/drivers/imports/${jobId}/status`), + commit: (jobId: string) => api.post(`/v1/a76/drivers/imports/${jobId}/commit`, {}) + }, + // Generic request for custom needs (like file uploads) request: (endpoint: string, options: RequestInit = {}) => fetchApi(endpoint, options) }; diff --git a/frontend/src/routes/dashboard/csv-upload/+page.svelte b/frontend/src/routes/dashboard/csv-upload/+page.svelte index a936bc8a..514fe09d 100644 --- a/frontend/src/routes/dashboard/csv-upload/+page.svelte +++ b/frontend/src/routes/dashboard/csv-upload/+page.svelte @@ -38,6 +38,8 @@ let useMaterialClassesImport = $state(false); // Cuando es true, usamos API de importación de Vehículos / Transportes (vehicles/imports) let useVehicleImport = $state(false); + // Cuando es true, usamos API de importación de Conductores (drivers/imports) + let useDriverImport = $state(false); // Initialize settings for all tabs upfront to avoid reactivity loops let allSettings = $state>(() => { @@ -63,6 +65,7 @@ usePedimentosImport = config.id === 'pedimentos'; useMaterialClassesImport = config.id === 'material_classes'; useVehicleImport = config.id === 'transports'; + useDriverImport = config.id === 'drivers'; const companyId = companyStore.activeCompany?.id || 1; @@ -192,6 +195,24 @@ return; } + if (useDriverImport) { + try { + const res = await api.driverImports.upload(file, companyId); + if (res.data?.job_id) { + currentJobId = res.data.job_id; + pollStatus(); + } else { + toast.error(res.error || 'Error al subir el archivo'); + isUploading = false; + } + } catch (e) { + console.error('Upload exception', e); + toast.error('Error inesperado al subir el archivo'); + isUploading = false; + } + return; + } + const currentSettings = allSettings[activeTab] || {}; const footerConfig = { ...currentSettings }; if (activeTab === 'importacion') { @@ -240,6 +261,8 @@ ? await api.materialClassImports.status(currentJobId) : useVehicleImport ? await api.vehicleImports.status(currentJobId) + : useDriverImport + ? await api.driverImports.status(currentJobId) : await api.imports.status(currentJobId); console.log('Poll response', res); if (res.error && !res.data) { @@ -398,6 +421,8 @@ ? await api.materialClassImports.commit(currentJobId) : useVehicleImport ? await api.vehicleImports.commit(currentJobId) + : useDriverImport + ? await api.driverImports.commit(currentJobId) : await api.imports.commit(currentJobId, activeModelTarget || ''); if (res.data?.commit_job_id) { currentJobId = res.data.commit_job_id;