feature/csv-expo

backend/api/v1/modules/a76/layouts_csv/exportacion/__init__.py

@@ -0,0 +1 @@
+# layouts_csv.exportacion — carga CSV exportación (encabezado y partidas)

backend/api/v1/modules/a76/layouts_csv/exportacion/routes.py

@@ -0,0 +1,151 @@
+"""
+Rutas de importación CSV para Exportación (encabezado y partidas).
+Flujo: upload → scan → status (polling) → commit. Sin validaciones ni inserción aún.
+"""
+import base64
+import json
+import logging
+import os
+from uuid import uuid4
+
+from fastapi import APIRouter, File, HTTPException, UploadFile, Depends, Form, Query
+from sqlalchemy.orm import Session
+from typing import Literal, Optional, Dict, Any
+
+from core.celery_app import celery_app
+from core.database import get_core_db
+from core.paths import layout_path
+from core.security import get_current_user, validate_access_to_resource
+
+from .schemas import ImportJobResponse, CommitRequest
+from .tasks import scan_file, insert_valid_rows, JOB_TYPE, EXP_IMPORT_REDIS_TTL
+from ..common import storage as common_storage
+
+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/{model_target}", response_model=ImportJobResponse)
+async def upload_import_file(
+    model_target: Literal["invoice_header", "invoice_details"],
+    file: UploadFile = File(...),
+    footer_config: Optional[str] = Form(None),
+    template_id: Optional[str] = Form(None),
+    company_id: int = Query(..., description="Company ID"),
+    operation_type: Optional[str] = Query("exp"),
+    db: Session = Depends(get_core_db),
+    current_user: Dict[str, Any] = Depends(get_current_user),
+):
+    """Subir CSV, guardar en Redis, encolar scan. operation_type=exp para exportación."""
+    try:
+        tenant_id = validate_access_to_resource(db, company_id, current_user)
+    except Exception as e:
+        logger.error("Exportación import: access validation failed: %s", 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()
+
+    file_key, meta_key, _ = common_storage.storage_keys(JOB_TYPE, job_id)
+    meta_data = {
+        "tenant_id": tenant_id,
+        "company_id": company_id,
+        "user_id": current_user.get("id"),
+        "footer_config": footer_config,
+        "operation_type": operation_type or "exp",
+        "template_id": template_id or ("exp_def_header" if model_target == "invoice_header" else "exp_def_details"),
+    }
+
+    try:
+        r = _get_redis()
+        r.set(file_key, base64.b64encode(contents), ex=EXP_IMPORT_REDIS_TTL)
+        r.set(meta_key, json.dumps(meta_data).encode("utf-8"), ex=EXP_IMPORT_REDIS_TTL)
+    except Exception as e:
+        logger.error("Exportación import: Redis store error: %s", e)
+        raise HTTPException(status_code=500, detail="No se pudo encolar el archivo.")
+
+    try:
+        upload_dir = layout_path("imports", "temp")
+        os.makedirs(upload_dir, exist_ok=True)
+        csv_path = common_storage.file_path_for_job(JOB_TYPE, job_id)
+        with open(csv_path, "wb") as f:
+            f.write(contents)
+        meta_path = csv_path.replace(".csv", ".meta.json")
+        with open(meta_path, "w") as f:
+            json.dump(meta_data, f)
+    except Exception as e:
+        logger.warning("Exportación import: local file save failed: %s", e)
+
+    scan_file.apply_async(args=[job_id, model_target, footer_config], 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}
+
+    if isinstance(getattr(task_result, "result", None), dict) and task_result.result.get("status") in ("finished", "warning"):
+        return task_result.result
+
+    logger.warning("Exportación 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:
+        result = getattr(task_result, "result", None)
+        if result is not None and not isinstance(result, dict):
+            err_msg = str(result)
+        elif isinstance(result, dict) and (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, body: CommitRequest):
+    """Usuario confirma; se encola la tarea de commit (por ahora sin inserción real)."""
+    task = insert_valid_rows.delay(job_id, body.model_target)
+    return {
+        "status": "committing",
+        "message": "Proceso de commit iniciado.",
+        "commit_job_id": task.id,
+    }

backend/api/v1/modules/a76/layouts_csv/exportacion/schemas.py

@@ -0,0 +1,23 @@
+from pydantic import BaseModel
+from typing import Optional, Literal
+
+
+class ImportJobResponse(BaseModel):
+    job_id: str
+    status: str
+    message: str
+
+
+class CommitRequest(BaseModel):
+    model_target: Literal["invoice_header", "invoice_details"]
+
+
+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
+    error_file: Optional[str] = None

backend/api/v1/modules/a76/layouts_csv/exportacion/tasks.py

@@ -0,0 +1,139 @@
+"""
+Tareas Celery para importación CSV de Exportación (encabezado y partidas).
+Flujo: scan_file (sin validaciones) → insert_valid_rows (sin inserción en BD).
+Usa layouts_csv.common (storage, normalize, meta, responses, csv_reader).
+"""
+import logging
+import os
+from typing import Dict, Any, Optional
+
+from core.celery_app import celery_app
+
+from ..common import storage as common_storage
+from ..common import normalize as common_normalize
+from ..common import meta as common_meta
+from ..common import responses as common_responses
+from ..common import csv_reader as common_csv_reader
+from .template_config import row_from_template
+
+logger = logging.getLogger(__name__)
+
+JOB_TYPE = "exp"
+EXP_IMPORT_REDIS_TTL = common_storage.IMPORT_REDIS_TTL
+
+
+def _ensure_file(job_id: str) -> Optional[str]:
+    return common_storage.ensure_file_from_redis(JOB_TYPE, job_id, "Exportación import")
+
+
+def _ensure_meta(job_id: str, file_path: str) -> bool:
+    return common_storage.ensure_meta_from_redis(JOB_TYPE, job_id, file_path, "Exportación import")
+
+
+def _norm_row(row: Dict[str, Any], template_id: str) -> Dict[str, Any]:
+    return row_from_template(row, template_id, common_normalize.normalize_header)
+
+
+@celery_app.task(bind=True, name="api.v1.modules.a76.layouts_csv.exportacion.tasks.scan_file")
+def scan_file(self, job_id: str, model_target: str, config: str = None):
+    """
+    Scan CSV sin validaciones: leer, normalizar con plantilla, devolver total_rows y 0 errores.
+    """
+    logger.info("Exportación import: starting scan for job %s target %s", job_id, model_target)
+
+    file_path = _ensure_file(job_id)
+    if not file_path:
+        return {"status": "failed", "error": "Archivo no encontrado (expirado o no subido). Sube de nuevo."}
+    if os.path.getsize(file_path) == 0:
+        return {"status": "failed", "error": "El archivo está vacío."}
+    _ensure_meta(job_id, file_path)
+
+    try:
+        common_meta.require_tenant_context(file_path)
+    except ValueError as e:
+        return {"status": "failed", "error": str(e)}
+
+    meta = common_meta.load_meta(file_path) or {}
+    template_id = meta.get("template_id") or (
+        "exp_def_header" if model_target == "invoice_header" else "exp_def_details"
+    )
+
+    total_rows = 0
+    processed_rows = 0
+
+    try:
+        total_rows = common_csv_reader.count_csv_rows(file_path, has_header=True)
+    except Exception as e:
+        return {"status": "failed", "error": str(e)}
+
+    def on_progress(current: int, total: int) -> None:
+        self.update_state(state="PROGRESS", meta={"current": current, "total": total})
+
+    try:
+        for i, row in common_csv_reader.iter_csv_rows(file_path, fieldnames=None):
+            if i % 500 == 0:
+                on_progress(i, total_rows)
+            _norm_row(row, template_id)
+            processed_rows += 1
+    except Exception as e:
+        logger.error("Exportación import scan failed: %s", e)
+        return {"status": "failed", "error": str(e)}
+
+    return common_responses.scan_result(job_id, processed_rows, 0, [])
+
+
+@celery_app.task(bind=True, name="api.v1.modules.a76.layouts_csv.exportacion.tasks.insert_valid_rows")
+def insert_valid_rows(self, job_id: str, model_target: str):
+    """
+    Commit sin inserción en BD: leer CSV, omitir líneas de error (vacío por ahora), cleanup, devolver finished con inserted=0.
+    """
+    logger.info("Exportación import: starting commit for job %s target %s", job_id, model_target)
+
+    file_path = _ensure_file(job_id)
+    if not file_path:
+        alt_path = common_storage.file_path_for_job(JOB_TYPE, job_id)
+        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_meta(job_id, file_path)
+
+    try:
+        common_meta.require_tenant_context(file_path)
+    except ValueError as e:
+        return {"status": "failed", "error": str(e)}
+
+    meta = common_meta.load_meta(file_path) or {}
+    meta_path = common_meta.get_meta_path(file_path)
+    error_path = common_storage.error_path_for_job(JOB_TYPE, job_id)
+    error_lines = common_storage.get_error_lines(JOB_TYPE, job_id, error_path)
+
+    template_id = meta.get("template_id") or (
+        "exp_def_header" if model_target == "invoice_header" else "exp_def_details"
+    )
+
+    try:
+        for i, row in common_csv_reader.iter_csv_rows(file_path, fieldnames=None):
+            if i in error_lines:
+                continue
+            _norm_row(row, template_id)
+    except Exception as e:
+        logger.error("Exportación import commit read failed: %s", e)
+        return {"status": "failed", "error": str(e)}
+
+    common_storage.cleanup_import_job(
+        JOB_TYPE, job_id,
+        file_path=file_path,
+        error_path=error_path,
+        meta_path=meta_path,
+    )
+
+    return {
+        "status": "finished",
+        "inserted": 0,
+        "skipped_invalid": 0,
+        "skipped_missing_fk": 0,
+        "skipped_duplicate": 0,
+        "skipped_details": [],
+        "message": "Proceso base listo; validaciones e inserción pendientes.",
+    }

backend/api/v1/modules/a76/layouts_csv/exportacion/template_config.py

@@ -0,0 +1,89 @@
+"""
+Plantillas CSV para Exportación (encabezado y partidas).
+Misma estructura que facturas exp_def_header / exp_def_details; módulo autocontenido.
+"""
+
+from typing import Dict, List, Any, Optional
+
+# Columnas para encabezado y partidas de exportación (EstructuraEncFacExpoCamReg / EstructuraParExpoCamReg)
+TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
+    "exp_def_header": [
+        {"canonical": "NUMERO FACTURA", "aliases": ["NUM FACTURA", "FACTURA", "ID"]},
+        {"canonical": "FECHA FACTURA", "aliases": ["FECHA"]},
+        {"canonical": "FECHA EMISION"},
+        {"canonical": "CLAVE PROVEEDOR"},
+        {"canonical": "CLAVE VENDIDO A"},
+        {"canonical": "CLAVE ENVIADO A"},
+        {"canonical": "REGIMEN", "aliases": ["CLAVEDOCUMENTO"]},
+        {"canonical": "ADUANA DE CRUCE"},
+        {"canonical": "CLAVE MONEDA"},
+        {"canonical": "CLAVE INCOTERM"},
+        {"canonical": "TIPO MONEDA"},
+        {"canonical": "TIPO DE CAMBIO"},
+        {"canonical": "TIPO PESO"},
+        {"canonical": "TIPO TRANSPORTE"},
+        {"canonical": "REMESA"},
+        {"canonical": "AGENTE ADUANAL"},
+        {"canonical": "FLETES"},
+        {"canonical": "VALOR SEGUROS"},
+        {"canonical": "SEGUROS"},
+        {"canonical": "EMBALAJES"},
+        {"canonical": "OTROS INCREMENTABLES"},
+        {"canonical": "NUM PROYECTO", "aliases": ["NUMPROYECTO"]},
+        {"canonical": "ORDEN COMPRA", "aliases": ["ORDENCOMPRA"]},
+        {"canonical": "FACTURA ALTERNA"},
+        {"canonical": "FACTURA EXPO REF", "aliases": ["FACTURAEXPOREF"]},
+        {"canonical": "OBSERVACIONES E"},
+        {"canonical": "OBSERVACIONES I"},
+        {"canonical": "E DOCUMENT"},
+        {"canonical": "NUM OPERACION"},
+        {"canonical": "CLAVE TRANSPORTISTA"},
+        {"canonical": "NOMBRE CONDUCTOR"},
+        {"canonical": "NUMERO TRANSPORTE"},
+        {"canonical": "PRECINTO"},
+    ],
+    "exp_def_details": [
+        {"canonical": "NUMERO FACTURA", "aliases": ["NUM FACTURA", "FACTURA"]},
+        {"canonical": "LINEA", "aliases": ["RENGLON", "PARTIDA"]},
+        {"canonical": "NUMPARTE", "aliases": ["NUMERO PARTE"]},
+        {"canonical": "PRECIO UNITARIO", "aliases": ["PRECIOUNITARIO"]},
+        {"canonical": "VALOR COMERCIAL", "aliases": ["VALORCOMERCIAL"]},
+        {"canonical": "CANTIDAD"},
+        {"canonical": "CANTIDAD BULTOS", "aliases": ["CANTIDADBULTOS"]},
+        {"canonical": "DESCRIPCION"},
+        {"canonical": "PAIS ORIGEN", "aliases": ["PAISORIGEN"]},
+        {"canonical": "FRACCION"},
+        {"canonical": "ORDEN DE COMPRA", "aliases": ["ORDENCOMPRA"]},
+    ],
+}
+
+
+def _resolve_template_columns(template_id: str) -> Optional[List[Dict[str, Any]]]:
+    return TEMPLATE_COLUMNS.get(template_id)
+
+
+def build_normalized_lookup(template_id: str, normalize_header_fn) -> Dict[str, str]:
+    """normalized_header -> canonical_name."""
+    cols = _resolve_template_columns(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], template_id: str, normalize_header_fn) -> Dict[str, Any]:
+    """Fila CSV -> dict con nombres canónicos de la plantilla."""
+    lookup = build_normalized_lookup(template_id, 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/router.py

@@ -15,6 +15,7 @@ from .classes import router as classes_router
 
 from .clients_and_providers import router as client_and_provider_router
 from .layouts_csv.facturas.routes import router as imports_router
+from .layouts_csv.exportacion.routes import router as exportacion_imports_router
 from .csv_templates.routes import router as csv_templates_router
 from .invoice_settings.routes import router as invoice_settings_router
 from .item_presets.routes import router as item_presets_router
@@ -56,6 +57,7 @@ router.include_router(general_catalogs_router, prefix="/a76", tags=["a76 / gener
 router.include_router(invoices_router, prefix="/a76", tags=["a76 / invoices"])
 router.include_router(items_router, prefix="/a76", tags=["a76 / items"])
 router.include_router(imports_router, prefix="/a76/imports", tags=["a76 / imports"])
+router.include_router(exportacion_imports_router, prefix="/a76/imports/exportacion", tags=["a76 / imports / exportacion"])
 router.include_router(csv_templates_router, prefix="/a76/csv-templates", tags=["a76 / csv_templates"])
 router.include_router(invoice_settings_router)
 router.include_router(item_presets_router, prefix="/a76/item-presets", tags=["a76 / item_presets"])