feature/csv-drivers

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
 

backend/api/v1/modules/a76/transportation/drivers/imports/__init__.py

@@ -0,0 +1 @@
+# CSV import for Drivers / Conductores (upload -> scan -> status -> commit).

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,
+    }

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

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

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

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(