feature/csv-pedimento

backend/api/v1/modules/a76/pedmientos/imports/__init__.py

@@ -0,0 +1 @@
+# CSV import for Pedimentos (upload → scan → commit)

backend/api/v1/modules/a76/pedmientos/imports/routes.py

@@ -0,0 +1,160 @@
+"""
+Rutas de importación CSV para Pedimentos.
+Mismo flujo que customs_brokers/imports: upload → scan → status (polling) → commit.
+"""
+import base64
+import json
+import logging
+import os
+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,
+    insert_valid_rows,
+    PED_IMPORT_FILE_PREFIX,
+    PED_IMPORT_META_PREFIX,
+    PED_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),
+):
+    """
+    Fase 1: Subir CSV, guardar en Redis, encolar tarea de escaneo.
+    """
+    try:
+        tenant_id = validate_access_to_resource(db, company_id, current_user)
+    except Exception as e:
+        logger.error(f"Pedimentos 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": "pedimentos",
+    }
+
+    try:
+        r = _get_redis()
+        r.set(
+            f"{PED_IMPORT_FILE_PREFIX}{job_id}",
+            base64.b64encode(contents),
+            ex=PED_IMPORT_REDIS_TTL,
+        )
+        r.set(
+            f"{PED_IMPORT_META_PREFIX}{job_id}",
+            json.dumps(meta_data).encode("utf-8"),
+            ex=PED_IMPORT_REDIS_TTL,
+        )
+    except Exception as e:
+        logger.error(f"Pedimentos 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"ped_{job_id}.csv"), "wb") as f:
+            f.write(contents)
+        with open(os.path.join(upload_dir, f"ped_{job_id}.meta.json"), "w") as f:
+            json.dump(meta_data, f)
+    except Exception as e:
+        logger.warning(f"Pedimentos import: local file save failed: {e}")
+
+    scan_file.apply_async(args=[job_id], task_id=job_id)
+
+    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):
+    """
+    Polling: estado del escaneo o del commit.
+    """
+    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("Pedimentos 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):
+    """
+    Fase 2: Usuario confirma; se encola la inserción de filas válidas.
+    """
+    task = insert_valid_rows.delay(job_id)
+    return {
+        "status": "committing",
+        "message": "Inserción iniciada.",
+        "commit_job_id": task.id,
+    }

backend/api/v1/modules/a76/pedmientos/imports/schemas.py

@@ -0,0 +1,25 @@
+from pydantic import BaseModel
+from typing import Optional
+
+
+class ImportJobResponse(BaseModel):
+    job_id: str
+    status: str
+    message: str
+
+
+class CommitRequest(BaseModel):
+    pass  # no body needed for single model
+
+
+class ImportJobStatus(BaseModel):
+    status: str
+    job_id: Optional[str] = None
+    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_details: Optional[list] = None

backend/api/v1/modules/a76/pedmientos/imports/tasks.py

@@ -0,0 +1,563 @@
+"""
+Tareas Celery para importación CSV de Pedimentos.
+Flujo en dos fases: scan_file (validación) → insert_valid_rows (commit).
+"""
+import os
+import base64
+import csv
+import json
+import logging
+import re
+import unicodedata
+from decimal import Decimal, InvalidOperation
+from datetime import datetime
+from typing import Dict, Any, Optional, List, Set
+
+from core.celery_app import celery_app
+from core.database import CoreSessionLocal
+
+from .template_config import row_from_template
+
+logger = logging.getLogger(__name__)
+
+# Redis keys (prefijo propio para no colisionar con otros imports)
+PED_IMPORT_FILE_PREFIX = "ped_import_file:"
+PED_IMPORT_META_PREFIX = "ped_import_meta:"
+PED_IMPORT_ERROR_LINES_PREFIX = "ped_import_error_lines:"
+PED_IMPORT_REDIS_TTL = 3600  # 1 hour
+
+
+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"{PED_IMPORT_FILE_PREFIX}{job_id}")
+    if not data:
+        return None
+    try:
+        raw = base64.b64decode(data)
+    except Exception as e:
+        logger.warning(f"Pedimentos 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"ped_{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"{PED_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"Pedimentos 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"{PED_IMPORT_FILE_PREFIX}{job_id}",
+            f"{PED_IMPORT_META_PREFIX}{job_id}",
+            f"{PED_IMPORT_ERROR_LINES_PREFIX}{job_id}",
+        )
+    except Exception as e:
+        logger.warning(f"Pedimentos 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()
+
+
+def _validate_row_pedimento(
+    row: Dict[str, Any],
+    line_num: int,
+    valid_client_ids: Set[int],
+    valid_regimes: Set[str],
+    valid_pedimento_codes: Set[str],
+) -> Optional[Dict[str, Any]]:
+    """Valida una fila para Pedimento. Retorna error dict o None."""
+    # Required: AÑO, ADUANA, PATENTE, NUMERO, CLIENTE_ID, CODIGO_PEDIMENTO, REGIMEN
+    year = (row.get("AÑO") or "").strip()
+    if not year:
+        return {"line": line_num, "col": "AÑO", "msg": "Requerido"}
+    if len(year) > 2:
+        return {"line": line_num, "col": "AÑO", "msg": "Máximo 2 caracteres"}
+
+    customs_office = (row.get("ADUANA") or "").strip()
+    if not customs_office:
+        return {"line": line_num, "col": "ADUANA", "msg": "Requerido"}
+    if len(customs_office) > 3:
+        return {"line": line_num, "col": "ADUANA", "msg": "Máximo 3 caracteres"}
+
+    license_val = (row.get("PATENTE") or "").strip()
+    if not license_val:
+        return {"line": line_num, "col": "PATENTE", "msg": "Requerido"}
+    if len(license_val) > 4:
+        return {"line": line_num, "col": "PATENTE", "msg": "Máximo 4 caracteres"}
+
+    pedimento_number = (row.get("NUMERO") or "").strip()
+    if not pedimento_number:
+        return {"line": line_num, "col": "NUMERO", "msg": "Requerido"}
+    if len(pedimento_number) > 7:
+        return {"line": line_num, "col": "NUMERO", "msg": "Máximo 7 caracteres"}
+
+    client_id_str = (row.get("CLIENTE_ID") or "").strip()
+    if not client_id_str:
+        return {"line": line_num, "col": "CLIENTE_ID", "msg": "Requerido"}
+    try:
+        client_id = int(client_id_str)
+    except ValueError:
+        return {"line": line_num, "col": "CLIENTE_ID", "msg": "Debe ser número entero"}
+    if client_id not in valid_client_ids:
+        return {"line": line_num, "col": "CLIENTE_ID", "msg": "Cliente no existe en catálogo"}
+
+    pedimento_code = (row.get("CODIGO_PEDIMENTO") or "").strip()
+    if not pedimento_code:
+        return {"line": line_num, "col": "CODIGO_PEDIMENTO", "msg": "Requerido"}
+    if len(pedimento_code) > 2:
+        return {"line": line_num, "col": "CODIGO_PEDIMENTO", "msg": "Máximo 2 caracteres"}
+    if pedimento_code not in valid_pedimento_codes:
+        return {"line": line_num, "col": "CODIGO_PEDIMENTO", "msg": "Código no existe en catálogo"}
+
+    regime = (row.get("REGIMEN") or "").strip()
+    if not regime:
+        return {"line": line_num, "col": "REGIMEN", "msg": "Requerido"}
+    if len(regime) > 3:
+        return {"line": line_num, "col": "REGIMEN", "msg": "Máximo 3 caracteres"}
+    if regime not in valid_regimes:
+        return {"line": line_num, "col": "REGIMEN", "msg": "Régimen no existe en catálogo"}
+
+    # Optional numeric/string fields - validate format if present
+    status = (row.get("ESTATUS") or "").strip()
+    if status and len(status) > 30:
+        return {"line": line_num, "col": "ESTATUS", "msg": "Máximo 30 caracteres"}
+
+    for col, max_len in [("VALOR_USD", 17), ("PRECIO_PAGADO", 17), ("PESO_BRUTO", 19), ("TIPO_CAMBIO", 9)]:
+        val = (row.get(col) or "").strip()
+        if not val:
+            continue
+        try:
+            Decimal(val.replace(",", "."))
+        except (InvalidOperation, ValueError):
+            return {"line": line_num, "col": col, "msg": "Valor numérico inválido"}
+
+    operation_type = (row.get("TIPO_OPERACION") or "").strip().lower()
+    if operation_type and operation_type not in ("imp", "exp", ""):
+        return {"line": line_num, "col": "TIPO_OPERACION", "msg": "Debe ser imp o exp"}
+
+    pedimento_type = (row.get("TIPO_PEDIMENTO") or "").strip().lower()
+    if pedimento_type and pedimento_type not in ("normal", "consolidated", "complementary", "automobile", ""):
+        return {"line": line_num, "col": "TIPO_PEDIMENTO", "msg": "Tipo no válido (normal, consolidated, complementary, automobile)"}
+
+    return None
+
+
+def _parse_decimal(val: Any) -> Optional[Decimal]:
+    if val is None or (isinstance(val, str) and not val.strip()):
+        return None
+    try:
+        return Decimal(str(val).strip().replace(",", "."))
+    except (InvalidOperation, ValueError):
+        return None
+
+
+def _row_to_pedimentos_create(row: Dict[str, Any]) -> Dict[str, Any]:
+    """Build dict for PedimentosCreate from normalized CSV row (canonical names)."""
+    year = (row.get("AÑO") or "").strip()[:2]
+    customs_office = (row.get("ADUANA") or "").strip()[:3]
+    license_val = (row.get("PATENTE") or "").strip()[:4]
+    pedimento_number = (row.get("NUMERO") or "").strip()[:7]
+    client_id_str = (row.get("CLIENTE_ID") or "").strip()
+    client_id = int(client_id_str) if client_id_str else None
+    pedimento_code = (row.get("CODIGO_PEDIMENTO") or "").strip()[:2]
+    regime = (row.get("REGIMEN") or "").strip()[:3]
+
+    data = {
+        "year": year,
+        "customs_office": customs_office,
+        "license": license_val,
+        "pedimento_number": pedimento_number,
+        "client_id": client_id,
+        "pedimento_code": pedimento_code,
+        "regime": regime,
+    }
+
+    op = (row.get("TIPO_OPERACION") or "").strip().lower()
+    if op in ("imp", "exp"):
+        data["operation_type"] = op
+
+    ptype = (row.get("TIPO_PEDIMENTO") or "").strip().lower()
+    if ptype in ("normal", "consolidated", "complementary", "automobile"):
+        data["pedimento_type"] = ptype
+
+    status = (row.get("ESTATUS") or "").strip()
+    if status:
+        data["status"] = status[:30]
+
+    data["usd_value"] = _parse_decimal(row.get("VALOR_USD"))
+    data["paid_price"] = _parse_decimal(row.get("PRECIO_PAGADO"))
+    data["gross_weight"] = _parse_decimal(row.get("PESO_BRUTO"))
+    data["exchange_rate"] = _parse_decimal(row.get("TIPO_CAMBIO"))
+
+    obs = (row.get("OBSERVACIONES") or "").strip()
+    if obs:
+        data["observations"] = obs
+
+    return data
+
+
+@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"Pedimentos import: starting scan for job {job_id}")
+
+    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"ped_{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"Pedimentos 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)"}
+
+    # Load valid FK sets for validation
+    valid_client_ids: Set[int] = set()
+    valid_regimes: Set[str] = set()
+    valid_pedimento_codes: Set[str] = set()
+    try:
+        with CoreSessionLocal() as session:
+            from api.v1.modules.a76.clients_and_providers.models import ClientProvider
+            from api.v1.modules.public.reference_data.pedimento_regimens.models import RegimenPedimento
+            from api.v1.modules.public.reference_data.pedimento_codes.models import PedimentoCode
+
+            for cp in session.query(ClientProvider).filter(
+                ClientProvider.tenant_id == tenant_id,
+                ClientProvider.company_id == company_id,
+            ).all():
+                valid_client_ids.add(cp.id)
+            for r in session.query(RegimenPedimento).all():
+                valid_regimes.add(r.code)
+            for pc in session.query(PedimentoCode).all():
+                valid_pedimento_codes.add(pc.code)
+    except Exception as e:
+        logger.error(f"Pedimentos import: failed to load FK sets: {e}")
+        return {"status": "failed", "error": f"No se pudo cargar catálogos: {e}"}
+
+    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 i % 500 == 0:
+                    self.update_state(
+                        state="PROGRESS",
+                        meta={"current": i, "total": total_rows, "errors": error_count},
+                    )
+
+                row_norm = row_from_template(row, normalize_header, "pedimentos")
+                err = _validate_row_pedimento(
+                    row_norm, i, valid_client_ids, valid_regimes, valid_pedimento_codes
+                )
+                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"Pedimentos 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"{PED_IMPORT_ERROR_LINES_PREFIX}{job_id}",
+                json.dumps(error_lines_list).encode("utf-8"),
+                ex=PED_IMPORT_REDIS_TTL,
+            )
+    except Exception as e:
+        logger.warning(f"Pedimentos import: failed to store error lines in Redis: {e}")
+
+    return {
+        "status": "waiting_confirmation",
+        "job_id": job_id,
+        "total_rows": processed_rows,
+        "error_count": error_count,
+        "valid_rows": processed_rows - error_count,
+        "errors": errors_detail,
+    }
+
+
+@celery_app.task(bind=True)
+def insert_valid_rows(self, job_id: str):
+    """
+    Fase 2: Re-leer CSV, omitir filas con error, insertar Pedimentos vía PedimentosService.create.
+    """
+    logger.info(f"Pedimentos import: starting commit for job {job_id}")
+
+    file_path = _ensure_worker_has_file_from_redis(job_id)
+    if not file_path:
+        alt_path = os.path.join(_worker_upload_dir(), f"ped_{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"ped_{job_id}.jsonl")
+
+    error_lines = set()
+    try:
+        r = _get_redis()
+        raw = r.get(f"{PED_IMPORT_ERROR_LINES_PREFIX}{job_id}")
+        if raw:
+            error_lines = set(json.loads(raw.decode("utf-8")))
+    except Exception as e:
+        logger.debug(f"Pedimentos 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)"}
+
+    # Reload FK sets for commit-time validation
+    valid_client_ids: Set[int] = set()
+    valid_regimes: Set[str] = set()
+    valid_pedimento_codes: Set[str] = set()
+    try:
+        with CoreSessionLocal() as session:
+            from api.v1.modules.a76.clients_and_providers.models import ClientProvider
+            from api.v1.modules.public.reference_data.pedimento_regimens.models import RegimenPedimento
+            from api.v1.modules.public.reference_data.pedimento_codes.models import PedimentoCode
+
+            for cp in session.query(ClientProvider).filter(
+                ClientProvider.tenant_id == tenant_id,
+                ClientProvider.company_id == company_id,
+            ).all():
+                valid_client_ids.add(cp.id)
+            for r in session.query(RegimenPedimento).all():
+                valid_regimes.add(r.code)
+            for pc in session.query(PedimentoCode).all():
+                valid_pedimento_codes.add(pc.code)
+    except Exception as e:
+        logger.error(f"Pedimentos import: failed to load FK sets: {e}")
+        return {"status": "failed", "error": str(e)}
+
+    from api.v1.modules.a76.pedmientos.dtos.pedimentos import PedimentosCreate
+    from api.v1.modules.a76.pedmientos.dtos.pedimento_dates import PedimentoDatesCreate
+    from api.v1.modules.a76.pedmientos.services.pedimentos import PedimentosService
+
+    inserted_count = 0
+    skipped_invalid = 0
+    skipped_missing_fk = 0
+    skipped_duplicate = 0
+    skipped_details: List[Dict[str, Any]] = []
+    response = None
+
+    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, "pedimentos")
+                    err = _validate_row_pedimento(
+                        row_norm, i, valid_client_ids, valid_regimes, valid_pedimento_codes
+                    )
+                    if err:
+                        skipped_invalid += 1
+                        skipped_details.append(
+                            {"line": i, "reason": f"{err.get('col', '')}: {err.get('msg', '')}"}
+                        )
+                        continue
+
+                    try:
+                        data = _row_to_pedimentos_create(row_norm)
+                        # Service expects at least pedimento_dates with entry_date/end_date for DB NOT NULL
+                        if "pedimento_dates" not in data or data.get("pedimento_dates") is None:
+                            data["pedimento_dates"] = PedimentoDatesCreate(
+                                entry_date=datetime.now(),
+                                end_date=datetime.now(),
+                            )
+                        create_data = PedimentosCreate(**data)
+                        PedimentosService.create(session, create_data, tenant_id, company_id)
+                        inserted_count += 1
+                    except ValueError as ve:
+                        if "Ya existe" in str(ve) or "duplicate" in str(ve).lower():
+                            skipped_duplicate += 1
+                            skipped_details.append({"line": i, "reason": str(ve)})
+                        else:
+                            skipped_invalid += 1
+                            skipped_details.append({"line": i, "reason": str(ve)})
+                    except Exception as e:
+                        logger.warning(f"Pedimentos import line {i}: {e}")
+                        skipped_invalid += 1
+                        skipped_details.append({"line": i, "reason": str(e)})
+
+        total_skipped = skipped_invalid + skipped_missing_fk + skipped_duplicate
+        if inserted_count == 0 and total_skipped > 0:
+            response = {
+                "status": "warning",
+                "inserted": 0,
+                "skipped_invalid": skipped_invalid,
+                "skipped_missing_fk": skipped_missing_fk,
+                "skipped_duplicate": skipped_duplicate,
+                "skipped_details": skipped_details,
+                "message": f"No se insertaron registros. {total_skipped} rechazados.",
+            }
+        elif inserted_count == 0:
+            response = {
+                "status": "failed",
+                "error": "No hay registros válidos en el archivo CSV",
+                "inserted": 0,
+                "skipped_invalid": skipped_invalid,
+                "skipped_missing_fk": skipped_missing_fk,
+                "skipped_duplicate": skipped_duplicate,
+                "skipped_details": skipped_details,
+            }
+        else:
+            response = {
+                "status": "finished",
+                "inserted": inserted_count,
+                "skipped_invalid": skipped_invalid,
+                "skipped_missing_fk": skipped_missing_fk,
+                "skipped_duplicate": skipped_duplicate,
+                "skipped_details": skipped_details,
+            }
+
+    except Exception as e:
+        logger.error(f"Pedimentos 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)
+        meta_path = file_path.replace(".csv", ".meta.json")
+        if os.path.exists(meta_path):
+            os.remove(meta_path)
+        _delete_import_from_redis(job_id)
+    except Exception as cleanup_err:
+        logger.warning(f"Pedimentos import cleanup failed: {cleanup_err}")
+
+    if response is None:
+        response = {
+            "status": "failed",
+            "error": "Error inesperado",
+            "inserted": 0,
+            "skipped_invalid": skipped_invalid,
+            "skipped_missing_fk": skipped_missing_fk,
+            "skipped_duplicate": skipped_duplicate,
+            "skipped_details": skipped_details,
+        }
+    return response

backend/api/v1/modules/a76/pedmientos/imports/template_config.py

@@ -0,0 +1,53 @@
+"""
+Configuración de plantilla CSV para Pedimentos (EstructuraCatPedimentos.xls).
+Solo se leen columnas definidas aquí; el resto se ignora.
+"""
+
+from typing import Dict, List, Any, Optional
+
+TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
+    "pedimentos": [
+        {"canonical": "AÑO", "aliases": ["YEAR", "ANIO"]},
+        {"canonical": "ADUANA", "aliases": ["CUSTOMS_OFFICE", "CUSTOMS OFFICE"]},
+        {"canonical": "PATENTE", "aliases": ["LICENCIA", "LICENSE", "LIC"]},
+        {"canonical": "NUMERO", "aliases": ["PEDIMENTO_NUMBER", "PEDIMENTO NUMBER", "NUMERO PEDIMENTO"]},
+        {"canonical": "CLIENTE_ID", "aliases": ["CLIENT_ID", "CLIENTE", "ID CLIENTE"]},
+        {"canonical": "TIPO_OPERACION", "aliases": ["OPERATION_TYPE", "OPERACION"]},
+        {"canonical": "TIPO_PEDIMENTO", "aliases": ["PEDIMENTO_TYPE", "TIPO"]},
+        {"canonical": "CODIGO_PEDIMENTO", "aliases": ["PEDIMENTO_CODE", "CODIGO", "CLAVE PEDIMENTO"]},
+        {"canonical": "REGIMEN", "aliases": ["REGIME"]},
+        {"canonical": "ESTATUS", "aliases": ["STATUS", "ESTADO"]},
+        {"canonical": "VALOR_USD", "aliases": ["USD_VALUE", "VALOR USD", "USD"]},
+        {"canonical": "PRECIO_PAGADO", "aliases": ["PAID_PRICE", "PRECIO PAGADO"]},
+        {"canonical": "PESO_BRUTO", "aliases": ["GROSS_WEIGHT", "PESO BRUTO", "PESO"]},
+        {"canonical": "TIPO_CAMBIO", "aliases": ["EXCHANGE_RATE", "TIPO CAMBIO", "CAMBIO"]},
+        {"canonical": "OBSERVACIONES", "aliases": ["OBSERVATIONS", "OBS", "NOTAS"]},
+    ],
+}
+
+
+def build_normalized_lookup(normalize_header_fn, template_id: str = "pedimentos") -> Dict[str, str]:
+    """normalized_header -> canonical_name para plantilla pedimentos."""
+    cols = TEMPLATE_COLUMNS.get(template_id)
+    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, template_id: str = "pedimentos") -> Dict[str, Any]:
+    """Fila CSV con solo columnas de la plantilla, en nombres canónicos."""
+    lookup = build_normalized_lookup(normalize_header_fn, template_id)
+    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/pedmientos/router.py

@@ -31,6 +31,7 @@ from .routes.pedimento_rectification_origin import (
 from .routes.pedimento_transport_means import router as pedimento_transport_means_router
 from .routes.pedimento_validation import router as pedimento_validation_router
 from .routes.pedimentos import router as pedimentos_router
+from .imports.routes import router as pedimentos_imports_router
 
 router = APIRouter()
 
@@ -117,3 +118,8 @@ router.include_router(
 router.include_router(
     pedimentos_router, prefix="/pedimentos", tags=["a76 / pedimentos"]
 )
+router.include_router(
+    pedimentos_imports_router,
+    prefix="/pedimentos/imports",
+    tags=["a76 / pedimentos / csv_import"],
+)

backend/core/celery_app.py

@@ -35,6 +35,7 @@ celery_app.conf.update(
         "api.v1.modules.a76.imports.tasks",
         "api.v1.modules.a76.customs_brokers.imports.tasks",
         "api.v1.modules.a76.clients_and_providers.imports.tasks",
+        "api.v1.modules.a76.pedmientos.imports.tasks",
         "api.v1.modules.a76.general_catalogs.exchange_rate.imports.tasks",
         "api.v1.modules.a76.general_catalogs.fractions.us_tariff_fractions.imports.tasks",
         "api.v1.modules.a76.reports.exportacion.transmission.MAINX30.task",

frontend/src/lib/api.ts

@@ -434,6 +434,21 @@ export const api = {
 			api.post(`/v1/a76/us-tariff-fractions/imports/${jobId}/commit`, {})
 	},
 
+	// CSV import for Pedimentos (pedimentos/imports)
+	pedimentosImports: {
+		upload: (file: File, companyId: number) => {
+			const formData = new FormData();
+			formData.append('file', file);
+			return fetchApi(
+				`/v1/a76/pedimentos/imports/upload?company_id=${companyId}`,
+				{ method: 'POST', body: formData }
+			);
+		},
+		status: (jobId: string) => api.get(`/v1/a76/pedimentos/imports/${jobId}/status`),
+		commit: (jobId: string) =>
+			api.post(`/v1/a76/pedimentos/imports/${jobId}/commit`, {})
+	},
+
 	// Generic request for custom needs (like file uploads)
 	request: <T = any>(endpoint: string, options: RequestInit = {}) => fetchApi<T>(endpoint, options)
 };

frontend/src/routes/dashboard/csv-upload/+page.svelte

@@ -32,6 +32,8 @@
 	let useExchangeRateImport = $state(false);
 	// Cuando es true, usamos API de importación de Fracción Americana (us_tariff_fractions/imports)
 	let useAmericanFractionImport = $state(false);
+	// Cuando es true, usamos API de importación de Pedimentos (pedimentos/imports)
+	let usePedimentosImport = $state(false);
 
 	// Initialize settings for all tabs upfront to avoid reactivity loops
 	let allSettings = $state<Record<string, any>>(() => {
@@ -54,6 +56,7 @@
 		useClientProviderImport = config.id === 'clients_providers';
 		useExchangeRateImport = config.id === 'exchange_rates';
 		useAmericanFractionImport = config.id === 'american_fractions';
+		usePedimentosImport = config.id === 'pedimentos';
 
 		const companyId = companyStore.activeCompany?.id || 1;
 
@@ -129,6 +132,24 @@
 			return;
 		}
 
+		if (usePedimentosImport) {
+			try {
+				const res = await api.pedimentosImports.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') {
@@ -171,7 +192,9 @@
 						? await api.exchangeRateImports.status(currentJobId)
 						: useAmericanFractionImport
 							? await api.americanFractionImports.status(currentJobId)
-							: await api.imports.status(currentJobId);
+							: usePedimentosImport
+								? await api.pedimentosImports.status(currentJobId)
+								: await api.imports.status(currentJobId);
 			console.log('Poll response', res);
 			if (res.error && !res.data) {
 				toast.error(res.error || 'Error al consultar el estado');
@@ -205,7 +228,8 @@
 				const inserted = res.data?.inserted || 0;
 				const skippedInvalid = res.data?.skipped_invalid || 0;
 				const skippedFk = res.data?.skipped_missing_fk || 0;
-				const totalSkipped = skippedInvalid + skippedFk;
+				const skippedDup = res.data?.skipped_duplicate || 0;
+				const totalSkipped = skippedInvalid + skippedFk + skippedDup;
 
 				if (inserted === 0) {
 					toast.error(`No se insertaron registros. ${totalSkipped} fueron rechazados.`);
@@ -219,12 +243,13 @@
 				const inserted = res.data?.inserted || 0;
 				const skippedInvalid = res.data?.skipped_invalid || 0;
 				const skippedFk = res.data?.skipped_missing_fk || 0;
+				const skippedDup = res.data?.skipped_duplicate || 0;
 				const skippedDetails = res.data?.skipped_details || [];
 
 				if (inserted > 0) {
 					toast.success(`Importación completada: ${inserted} registros insertados`);
-					if (skippedInvalid > 0 || skippedFk > 0) {
-						const totalSkipped = skippedInvalid + skippedFk;
+					if (skippedInvalid > 0 || skippedFk > 0 || skippedDup > 0) {
+						const totalSkipped = skippedInvalid + skippedFk + skippedDup;
 						toast.warning(`${totalSkipped} registros fueron rechazados`);
 					}
 				} else {
@@ -321,7 +346,9 @@
 							? await api.exchangeRateImports.commit(currentJobId)
 							: useAmericanFractionImport
 								? await api.americanFractionImports.commit(currentJobId)
-								: await api.imports.commit(currentJobId, activeModelTarget || '');
+								: usePedimentosImport
+									? await api.pedimentosImports.commit(currentJobId)
+									: await api.imports.commit(currentJobId, activeModelTarget || '');
 				if (res.data?.commit_job_id) {
 					currentJobId = res.data.commit_job_id;
 					pollStatus();

scripts/backend-entrypoint.sh

@@ -19,7 +19,7 @@ wait_for_tcp() {
   echo "Esperando a que $service esté disponible en ${host}:${port}..."
   
   while [ $attempt -le $max_attempts ]; do
-    if python -c "import socket; s = socket.socket(); s.settimeout(3); s.connect(('$host', $port)); s.close()" 2>/dev/null; then
+    if python3 -c "import socket; s = socket.socket(); s.settimeout(3); s.connect(('$host', $port)); s.close()" 2>/dev/null; then
       echo "✓ $service está listo y accesible"
       return 0
     fi