feature/api-cove-integration

2026-04-06 13:52:33 -06:00
parent c71741077f 0c25572d70
commit fe74481786
61 changed files with 4218 additions and 627 deletions
--- a/backend/api/v1/modules/a76/audit_log/router.py
+++ b/backend/api/v1/modules/a76/audit_log/router.py
@@ -1,19 +1,184 @@
 """
 Audit Log Router
 """
-from typing import List, Optional
 from datetime import date
-from fastapi import APIRouter, Depends, Query, HTTPException
+from typing import Any, Dict, List, Optional
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from fastapi.responses import StreamingResponse
 from sqlalchemy.orm import Session
 from sqlalchemy import or_, desc, distinct

 from core.database import get_core_db
-from core.security import get_current_user # Assuming this exists
+from core.security import get_current_user, get_tenant_from_token
+from core.storage_s3 import get_object_bytes, list_objects_tree, should_ensure_s3_bucket
 from .models import AuditLog
-from .schemas import AuditLogListResponse, AuditLogResponse, AuditLogDetailResponse
+from .schemas import (
+    AuditFileBreadcrumb,
+    AuditFileBrowserResponse,
+    AuditFileFolderItem,
+    AuditFileObjectItem,
+    AuditLogDetailResponse,
+    AuditLogListResponse,
+)
+from api.v1.modules.a76.general_catalogs.company.models import Company

 router = APIRouter()

+_SEGMENT_LABELS = {
+    "tenants": "Espacio",
+    "companies": "Companias",
+    "users": "Usuarios",
+    "imports": "Importaciones",
+    "csv": "Archivos CSV",
+    "branding": "Logotipos",
+    "certificates": "Certificados",
+    "customs_brokers": "Agentes aduanales",
+    "keys": "Llaves",
+    "cove": "COVE",
+    "doda": "DODA",
+    "system": "Sistema",
+    "help": "Ayuda",
+}
+
+
+def _tenant_id_from_user(current_user: Dict[str, Any]) -> int:
+    tenant_id = get_tenant_from_token(current_user) or current_user.get("tenant_id")
+    if not tenant_id:
+        raise HTTPException(status_code=401, detail="User context is invalid")
+    return int(tenant_id)
+
+
+def _normalize_relative_path(raw: Optional[str]) -> str:
+    if not raw:
+        return ""
+    val = raw.strip().strip("/")
+    if not val:
+        return ""
+    if ".." in val or "\\" in val:
+        raise HTTPException(status_code=400, detail="Invalid path")
+    parts = [p for p in val.split("/") if p]
+    for part in parts:
+        if part in (".", ".."):
+            raise HTTPException(status_code=400, detail="Invalid path segment")
+    return "/".join(parts)
+
+
+def _tenant_prefix(tenant_id: int) -> str:
+    return f"tenants/{tenant_id}/"
+
+
+def _relative_from_tenant_prefix(key: str, tenant_prefix: str) -> str:
+    if not key.startswith(tenant_prefix):
+        raise HTTPException(status_code=403, detail="Access denied to object key")
+    return key[len(tenant_prefix) :].strip("/")
+
+
+def _companies_map(db: Session, tenant_id: int) -> Dict[str, str]:
+    rows = (
+        db.query(Company.id, Company.name)
+        .filter(Company.tenant_id == tenant_id, Company.deleted_at.is_(None))
+        .all()
+    )
+    out: Dict[str, str] = {}
+    for company_id, company_name in rows:
+        if company_id is None:
+            continue
+        safe_name = (company_name or "").strip()
+        out[str(company_id)] = safe_name or "Compania"
+    return out
+
+
+def _display_segment(
+    part: str,
+    prev_part: Optional[str],
+    company_names: Dict[str, str],
+    current_user_id: Optional[str] = None,
+    current_user_label: Optional[str] = None,
+) -> str:
+    if prev_part == "companies":
+        return company_names.get(part, "Compania")
+    if prev_part == "users":
+        # Para carpetas de usuarios, mostrar un nombre amigable:
+        # - Si es el propio usuario actual, usar preferred_username/email/nombre.
+        # - Para otros IDs (UUIDs) mostrar un label genérico.
+        if current_user_id and part == str(current_user_id):
+            return (current_user_label or "").strip() or "Usuario"
+        return "Usuario"
+    if part in _SEGMENT_LABELS:
+        return _SEGMENT_LABELS[part]
+    # Evita exponer IDs puros en UI.
+    if part.isdigit():
+        return "Elemento"
+    return part.replace("_", " ").strip().title() or "Elemento"
+
+
+def _display_path(
+    rel_path: str,
+    company_names: Dict[str, str],
+    current_user_id: Optional[str] = None,
+    current_user_label: Optional[str] = None,
+) -> str:
+    if not rel_path:
+        return "Raiz de archivos"
+    parts = [p for p in rel_path.split("/") if p]
+    labels: List[str] = []
+    prev: Optional[str] = None
+    for part in parts:
+        labels.append(
+            _display_segment(
+                part,
+                prev,
+                company_names,
+                current_user_id=current_user_id,
+                current_user_label=current_user_label,
+            )
+        )
+        prev = part
+    return " / ".join(labels)
+
+
+def _display_file_name(filename: str) -> str:
+    stem, dot, ext = filename.rpartition(".")
+    if not dot:
+        stem = filename
+        ext = ""
+    if stem.isdigit():
+        return f"Archivo{f'.{ext}' if ext else ''}"
+    return filename
+
+
+def _build_breadcrumbs(
+    rel_path: str,
+    company_names: Dict[str, str],
+    current_user_id: Optional[str] = None,
+    current_user_label: Optional[str] = None,
+) -> List[AuditFileBreadcrumb]:
+    breadcrumbs: List[AuditFileBreadcrumb] = [
+        AuditFileBreadcrumb(path="", display_name="Raiz de archivos")
+    ]
+    if not rel_path:
+        return breadcrumbs
+    parts = [p for p in rel_path.split("/") if p]
+    prev: Optional[str] = None
+    acc: List[str] = []
+    for part in parts:
+        acc.append(part)
+        breadcrumbs.append(
+            AuditFileBreadcrumb(
+                path="/".join(acc),
+                display_name=_display_segment(
+                    part,
+                    prev,
+                    company_names,
+                    current_user_id=current_user_id,
+                    current_user_label=current_user_label,
+                ),
+            )
+        )
+        prev = part
+    return breadcrumbs
+
@router.get("/bitacora", response_model=AuditLogListResponse)
 async def get_bitacora(
    page: int = Query(1, ge=1),
@@ -91,3 +256,127 @@ async def get_audit_detail(spec_id: int, db: Session = Depends(get_core_db)):
    if not log:
        raise HTTPException(status_code=404, detail="Log entry not found")
    return log
+
+
+@router.get("/files", response_model=AuditFileBrowserResponse)
+async def list_tenant_files(
+    path: Optional[str] = Query(default="", description="Ruta relativa de navegación."),
+    continuation_token: Optional[str] = Query(default=None),
+    max_keys: int = Query(default=100, ge=1, le=500),
+    db: Session = Depends(get_core_db),
+    current_user: Dict[str, Any] = Depends(get_current_user),
+):
+    """
+    Explorador de archivos de solo lectura para Auditoría.
+    """
+    if not should_ensure_s3_bucket():
+        raise HTTPException(status_code=400, detail="S3 storage is disabled")
+
+    tenant_id = _tenant_id_from_user(current_user)
+    tenant_prefix = _tenant_prefix(tenant_id)
+    rel_path = _normalize_relative_path(path)
+    list_prefix = f"{tenant_prefix}{rel_path}/" if rel_path else tenant_prefix
+
+    # Datos del usuario actual para etiquetas amigables bajo /users/{id}/...
+    current_user_id = str(current_user.get("sub") or "")
+    current_user_label = (
+        (current_user.get("preferred_username") or "").strip()
+        or (current_user.get("name") or "").strip()
+        or (current_user.get("email") or "").strip()
+        or "Usuario"
+    )
+
+    data = list_objects_tree(
+        prefix=list_prefix,
+        delimiter="/",
+        max_keys=max_keys,
+        continuation_token=continuation_token,
+    )
+
+    company_names = _companies_map(db, tenant_id)
+    folders: List[AuditFileFolderItem] = []
+    for prefix in data.get("prefixes", []):
+        rel = _relative_from_tenant_prefix(prefix, tenant_prefix)
+        folders.append(
+            AuditFileFolderItem(
+                path=rel,
+                # Para el gestor de archivos mostramos el nombre amigable del último segmento
+                # (empresa, usuario actual, etc.), no el ID bruto.
+                display_name=_display_path(
+                    rel,
+                    company_names,
+                    current_user_id=current_user_id,
+                    current_user_label=current_user_label,
+                ).split(" / ")[-1],
+            )
+        )
+
+    files: List[AuditFileObjectItem] = []
+    for obj in data.get("objects", []):
+        key = obj.get("key")
+        if not key:
+            continue
+        rel = _relative_from_tenant_prefix(key, tenant_prefix)
+        name = rel.rsplit("/", 1)[-1]
+        files.append(
+            AuditFileObjectItem(
+                path=rel,
+                display_name=_display_file_name(name),
+                size=int(obj.get("size", 0) or 0),
+                last_modified=obj.get("last_modified"),
+            )
+        )
+
+    return AuditFileBrowserResponse(
+        current_path=rel_path,
+        display_path=_display_path(
+            rel_path,
+            company_names,
+            current_user_id=current_user_id,
+            current_user_label=current_user_label,
+        ),
+        breadcrumbs=_build_breadcrumbs(
+            rel_path,
+            company_names,
+            current_user_id=current_user_id,
+            current_user_label=current_user_label,
+        ),
+        folders=sorted(folders, key=lambda x: x.display_name.lower()),
+        files=sorted(files, key=lambda x: x.display_name.lower()),
+        next_token=data.get("next_continuation_token"),
+    )
+
+
+@router.get("/files/download")
+async def download_tenant_file(
+    path: str = Query(..., description="Ruta relativa del archivo a descargar."),
+    current_user: Dict[str, Any] = Depends(get_current_user),
+):
+    """
+    Descarga segura (backend streaming) de archivos autorizados.
+    """
+    if not should_ensure_s3_bucket():
+        raise HTTPException(status_code=400, detail="S3 storage is disabled")
+
+    tenant_id = _tenant_id_from_user(current_user)
+    tenant_prefix = _tenant_prefix(tenant_id)
+    rel_path = _normalize_relative_path(path)
+    if not rel_path or rel_path.endswith("/"):
+        raise HTTPException(status_code=400, detail="A file path is required")
+
+    object_key = f"{tenant_prefix}{rel_path}"
+    if not object_key.startswith(tenant_prefix):
+        raise HTTPException(status_code=403, detail="Access denied to object key")
+
+    try:
+        body = get_object_bytes(object_key)
+    except Exception as e:
+        raise HTTPException(status_code=404, detail=f"File not found: {e}") from e
+
+    filename = rel_path.rsplit("/", 1)[-1]
+    headers = {"Content-Disposition": f'attachment; filename="{filename}"'}
+    return StreamingResponse(
+        iter([body]),
+        media_type="application/octet-stream",
+        headers=headers,
+    )
--- a/backend/api/v1/modules/a76/audit_log/schemas.py
+++ b/backend/api/v1/modules/a76/audit_log/schemas.py
@@ -50,3 +50,29 @@ class AuditLogListResponse(BaseModel):
    total: int
    page: int
    page_size: int
+
+
+class AuditFileBreadcrumb(BaseModel):
+    path: str = Field(default="")
+    display_name: str
+
+
+class AuditFileFolderItem(BaseModel):
+    path: str = Field(description="Ruta relativa interna del archivo, para navegación.")
+    display_name: str
+
+
+class AuditFileObjectItem(BaseModel):
+    path: str = Field(description="Ruta relativa interna del archivo, para descarga.")
+    display_name: str
+    size: int
+    last_modified: Optional[datetime] = None
+
+
+class AuditFileBrowserResponse(BaseModel):
+    current_path: str = Field(default="")
+    display_path: str = Field(default="")
+    breadcrumbs: List[AuditFileBreadcrumb]
+    folders: List[AuditFileFolderItem]
+    files: List[AuditFileObjectItem]
+    next_token: Optional[str] = None
--- a/backend/api/v1/modules/a76/customs_brokers/routes.py
+++ b/backend/api/v1/modules/a76/customs_brokers/routes.py
@@ -1,16 +1,69 @@
-from typing import Dict, Any
-from fastapi import APIRouter, Depends, HTTPException, Query
+import logging
+import mimetypes
+import os
+from datetime import datetime
+from typing import Any, Dict, Optional
+
+from fastapi import APIRouter, Depends, File, HTTPException, Query, UploadFile, status
 from sqlalchemy.orm import Session

+from core.config import settings
 from core.database import get_core_db
-from core.security import get_current_user, validate_access_to_resource
+from core.s3_keys import (
+    customs_broker_vu_certificate_key,
+    customs_broker_vu_cove_key,
+    customs_broker_vu_doda_certificate_key,
+    customs_broker_vu_doda_cove_key,
+    customs_broker_vu_doda_private_key_key,
+    customs_broker_vu_private_key_key,
+)
+from core.security import get_current_user, get_tenant_from_token, validate_access_to_resource
+from core.storage_s3 import delete_object_if_exists, put_object_bytes

 from api.v1.common.tenant_crud_routes import TenantCRUDRoutes
 from . import dto, services
 from ..layouts_csv.customs_brokers.routes import router as imports_router

+logger = logging.getLogger(__name__)
+
 router = APIRouter()

+MAX_VU_CER_KEY_BYTES = 5 * 1024 * 1024  # 5 MB
+MAX_COVE_BYTES = 15 * 1024 * 1024  # 15 MB (xml/zip)
+
+
+def _resolve_tenant_id_int(current_user: dict) -> int:
+    tid = get_tenant_from_token(current_user)
+    if tid is not None:
+        return int(tid)
+    raw = current_user.get("tenant_id")
+    if isinstance(raw, list) and raw:
+        raw = raw[0]
+    if raw is None:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Tenant ID not found in user data",
+        )
+    try:
+        return int(raw)
+    except (TypeError, ValueError):
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Invalid tenant ID in token",
+        )
+
+
+def _remove_stored_vu_path(ref: Optional[str]) -> None:
+    if not ref:
+        return
+    if ref.startswith("tenants/"):
+        delete_object_if_exists(ref)
+    elif os.path.isfile(ref):
+        try:
+            os.remove(ref)
+        except OSError:
+            pass
+
 # CSV import (mismo flujo que a76.imports: upload → scan → commit)
 router.include_router(imports_router, prefix="/customs-brokers/imports", tags=["customs_brokers / csv_import"])

@@ -112,4 +165,246 @@ def update_customs_broker_personnel(
        raise HTTPException(
            status_code=404, detail="Customs Broker Personnel not found"
        )
-    return updated_personnel
+    return updated_personnel
+
+
+@router.post(
+    "/customs-brokers/{broker_key}/vu/upload",
+    summary="Sube VU/DODA (CER, KEY, COVE) al bucket bajo tenants/.../customs_brokers/{id}/...",
+)
+async def upload_customs_broker_vu_file(
+    broker_key: str,
+    file_kind: str = Query(
+        ...,
+        description=(
+            "certificate (.cer), key (.key), cove (xml/zip/txt/pdf/json), "
+            "doda_certificate (.cer), doda_key (.key), doda_cove (xml/zip/txt/pdf/json)"
+        ),
+    ),
+    company_id: int = Query(..., description="Company ID"),
+    file: UploadFile = File(...),
+    db: Session = Depends(get_core_db),
+    current_user: Dict[str, Any] = Depends(get_current_user),
+):
+    """
+    Persiste el archivo bajo la misma jerarquía que logos/avatares (tenant/company/...).
+    Guarda la clave S3 o ruta local en:
+    - VU: certificate_path, key_path, xml_files_path
+    - DODA: doda_certificate_path, doda_key_path, doda_xml_files_path
+    """
+    validate_access_to_resource(db, company_id, current_user)
+    tenant_id = _resolve_tenant_id_int(current_user)
+
+    broker = services.CustomsBrokerService.get_by_id(db, broker_key, tenant_id, company_id)
+    if not broker:
+        raise HTTPException(status_code=404, detail="Customs Broker not found")
+
+    fk = file_kind.lower().strip()
+    if fk not in (
+        "certificate",
+        "key",
+        "cove",
+        "doda_certificate",
+        "doda_key",
+        "doda_cove",
+    ):
+        raise HTTPException(
+            status_code=400,
+            detail=(
+                "file_kind must be certificate, key, cove, "
+                "doda_certificate, doda_key, or doda_cove"
+            ),
+        )
+
+    content = await file.read()
+    max_bytes = MAX_COVE_BYTES if fk in ("cove", "doda_cove") else MAX_VU_CER_KEY_BYTES
+    if len(content) > max_bytes:
+        raise HTTPException(
+            status_code=400,
+            detail=f"File too large (max {max_bytes // (1024 * 1024)} MB)",
+        )
+
+    file_ext = os.path.splitext(file.filename or "")[1].lower()
+    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+    vu = services.CustomsBrokerVUService.ensure_vu_for_broker(db, broker)
+    broker_id = broker.id
+
+    field_name: str
+    stored: str
+
+    try:
+        if settings.use_s3_object_storage:
+            if fk == "certificate":
+                if file_ext != ".cer":
+                    raise HTTPException(status_code=400, detail="certificate must be .cer")
+                key = customs_broker_vu_certificate_key(
+                    tenant_id, company_id, broker_id, timestamp, file_ext
+                )
+                ct = "application/x-x509-ca-cert"
+                field_name = "certificate_path"
+            elif fk == "key":
+                if file_ext != ".key":
+                    raise HTTPException(status_code=400, detail="key must be .key")
+                key = customs_broker_vu_private_key_key(
+                    tenant_id, company_id, broker_id, timestamp, file_ext
+                )
+                ct = "application/pkcs8"
+                field_name = "key_path"
+            elif fk == "cove":
+                key = customs_broker_vu_cove_key(
+                    tenant_id,
+                    company_id,
+                    broker_id,
+                    timestamp,
+                    file.filename or "cove.xml",
+                )
+                ct = (
+                    file.content_type
+                    or mimetypes.guess_type(file.filename or "")[0]
+                    or "application/octet-stream"
+                )
+                field_name = "xml_files_path"
+            elif fk == "doda_certificate":
+                if file_ext != ".cer":
+                    raise HTTPException(status_code=400, detail="doda_certificate must be .cer")
+                key = customs_broker_vu_doda_certificate_key(
+                    tenant_id, company_id, broker_id, timestamp, file_ext
+                )
+                ct = "application/x-x509-ca-cert"
+                field_name = "doda_certificate_path"
+            elif fk == "doda_key":
+                if file_ext != ".key":
+                    raise HTTPException(status_code=400, detail="doda_key must be .key")
+                key = customs_broker_vu_doda_private_key_key(
+                    tenant_id, company_id, broker_id, timestamp, file_ext
+                )
+                ct = "application/pkcs8"
+                field_name = "doda_key_path"
+            else:
+                key = customs_broker_vu_doda_cove_key(
+                    tenant_id,
+                    company_id,
+                    broker_id,
+                    timestamp,
+                    file.filename or "doda.xml",
+                )
+                ct = (
+                    file.content_type
+                    or mimetypes.guess_type(file.filename or "")[0]
+                    or "application/octet-stream"
+                )
+                field_name = "doda_xml_files_path"
+
+            old = getattr(vu, field_name)
+            _remove_stored_vu_path(old)
+            put_object_bytes(key, content, content_type=ct)
+            logger.info(
+                "Customs broker VU upload kind=%s key=%s bytes=%s",
+                fk,
+                key,
+                len(content),
+            )
+            stored = key
+        else:
+            base = os.path.join(
+                "uploads", "customs_brokers", str(company_id), str(broker_id)
+            )
+            if fk == "certificate":
+                if file_ext != ".cer":
+                    raise HTTPException(status_code=400, detail="certificate must be .cer")
+                key = customs_broker_vu_certificate_key(
+                    tenant_id, company_id, broker_id, timestamp, file_ext
+                )
+                field_name = "certificate_path"
+                subdir = "certificates"
+            elif fk == "key":
+                if file_ext != ".key":
+                    raise HTTPException(status_code=400, detail="key must be .key")
+                key = customs_broker_vu_private_key_key(
+                    tenant_id, company_id, broker_id, timestamp, file_ext
+                )
+                field_name = "key_path"
+                subdir = "keys"
+            elif fk == "cove":
+                try:
+                    key = customs_broker_vu_cove_key(
+                        tenant_id,
+                        company_id,
+                        broker_id,
+                        timestamp,
+                        file.filename or "cove.xml",
+                    )
+                except ValueError as e:
+                    raise HTTPException(status_code=400, detail=str(e)) from e
+                field_name = "xml_files_path"
+                subdir = "cove"
+            elif fk == "doda_certificate":
+                if file_ext != ".cer":
+                    raise HTTPException(status_code=400, detail="doda_certificate must be .cer")
+                key = customs_broker_vu_doda_certificate_key(
+                    tenant_id, company_id, broker_id, timestamp, file_ext
+                )
+                field_name = "doda_certificate_path"
+                subdir = "doda/certificates"
+            elif fk == "doda_key":
+                if file_ext != ".key":
+                    raise HTTPException(status_code=400, detail="doda_key must be .key")
+                key = customs_broker_vu_doda_private_key_key(
+                    tenant_id, company_id, broker_id, timestamp, file_ext
+                )
+                field_name = "doda_key_path"
+                subdir = "doda/keys"
+            else:
+                try:
+                    key = customs_broker_vu_doda_cove_key(
+                        tenant_id,
+                        company_id,
+                        broker_id,
+                        timestamp,
+                        file.filename or "doda.xml",
+                    )
+                except ValueError as e:
+                    raise HTTPException(status_code=400, detail=str(e)) from e
+                field_name = "doda_xml_files_path"
+                subdir = "doda/cove"
+
+            fname = key.rsplit("/", 1)[-1]
+            dest_dir = os.path.join(base, subdir)
+            os.makedirs(dest_dir, exist_ok=True)
+            path = os.path.join(dest_dir, fname)
+            old = getattr(vu, field_name)
+            _remove_stored_vu_path(old)
+            with open(path, "wb") as f:
+                f.write(content)
+            logger.info(
+                "Customs broker VU upload kind=%s path=%s bytes=%s",
+                fk,
+                path,
+                len(content),
+            )
+            stored = path
+
+        setattr(vu, field_name, stored)
+        db.add(vu)
+        db.commit()
+        db.refresh(vu)
+    except HTTPException:
+        db.rollback()
+        raise
+    except ValueError as e:
+        db.rollback()
+        raise HTTPException(status_code=400, detail=str(e)) from e
+    except Exception as e:
+        db.rollback()
+        raise HTTPException(
+            status_code=500, detail=f"Error saving file: {str(e)}"
+        ) from e
+
+    return {
+        "message": "File uploaded successfully",
+        "file_kind": fk,
+        "field": field_name,
+        "path": stored,
+        "broker_key": broker_key,
+        "company_id": company_id,
+    }
--- a/backend/api/v1/modules/a76/customs_brokers/services.py
+++ b/backend/api/v1/modules/a76/customs_brokers/services.py
@@ -160,6 +160,26 @@ class CustomsBrokerVUService:
            db.commit()
        return vu

+    @staticmethod
+    def ensure_vu_for_broker(db: Session, broker: models.CustomsBroker) -> models.CustomsBrokerVU:
+        """Crea fila VU vacía si no existe (p. ej. antes de subir CER/KEY/COVE al bucket)."""
+        vu = (
+            db.query(models.CustomsBrokerVU)
+            .filter(models.CustomsBrokerVU.customs_broker_id == broker.id)
+            .first()
+        )
+        if vu:
+            return vu
+        vu = models.CustomsBrokerVU(
+            customs_broker_id=broker.id,
+            tenant_id=broker.tenant_id,
+            company_id=broker.company_id,
+        )
+        db.add(vu)
+        db.commit()
+        db.refresh(vu)
+        return vu
+

 class CustomsBrokerPersonnelService:
    @staticmethod
--- a/backend/api/v1/modules/a76/factura_cove/external_service.py
+++ b/backend/api/v1/modules/a76/factura_cove/external_service.py
@@ -0,0 +1,178 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Dict
+
+import httpx
+
+from core.config import settings
+from .schemas import FacturaCoveRequest
+
+
+@dataclass
+class CoveExternalResult:
+    """
+    Resultado simplificado de la llamada al servicio externo de COVE.
+
+    Por ahora usamos un stub que simula una respuesta exitosa y devuelve
+    un número de COVE ficticio para poder probar el flujo end-to-end
+    (task_id + cove_number) sin depender del ambiente externo.
+    """
+
+    status: str
+    message: str | None = None
+    cove_number: str | None = None
+    vucem_operation_num: str | None = None
+    raw_response: Dict[str, Any] | None = None
+
+
+class CoveExternalService:
+    """
+    Cliente del API externo de COVE.
+
+    NOTA IMPORTANTE:
+    ----------------
+    Esta implementación es, por ahora, un stub que:
+      - No realiza la llamada HTTP real.
+      - Genera un número de COVE ficticio basado en los datos de la factura.
+
+    Cuando se tenga disponible la URL y contrato exacto del servicio COVE,
+    este stub se puede reemplazar por una implementación con httpx/requests
+    que:
+      - Serialice el FacturaCoveRequest al JSON requerido.
+      - Realice la petición HTTP.
+      - Mapee la respuesta real a CoveExternalResult.
+    """
+
+    def __init__(self) -> None:
+        """
+        Inicializa el cliente usando COVE_API_URL si está definido; de lo contrario,
+        usa por defecto el endpoint público documentado en:
+        https://api.vu.aduanasoft.com/docs#/Factura%20COVE/generar_factura_cove_endpoint_api_v1_factura_cove_generar_factura_cove_post
+        """
+        self.base_url = (settings.COVE_API_URL or "").strip() or "https://api.vu.aduanasoft.com"
+
+    def generate_cove(self, payload: FacturaCoveRequest) -> CoveExternalResult:
+        """
+        Llama al endpoint externo /api/v1/factura-cove/generar-factura-cove
+        con el FacturaCoveRequest completo y retorna el resultado tal como
+        lo reporta el servicio remoto.
+        """
+        url = f"{self.base_url.rstrip('/')}/api/v1/factura-cove/generar-factura-cove"
+        json_payload = payload.model_dump(mode="json")
+
+        # NOTA: verify=False desactiva la validación de certificado SSL.
+        # Esto es útil en entornos de desarrollo o cuando el entorno no confía
+        # en el certificado del endpoint externo. En producción, idealmente
+        # se debería habilitar la verificación SSL.
+        with httpx.Client(timeout=30.0, verify=False) as client:
+            resp = client.post(url, json=json_payload)
+
+        # Intentar parsear JSON siempre, incluso en errores 4xx/5xx
+        try:
+            data: Dict[str, Any] = resp.json()
+        except Exception:
+            data = {}
+
+        # Si el servicio externo respondió con error (por ejemplo 422 Validation Error),
+        # devolvemos un resultado de error rico en información para que la UI pueda
+        # mostrar el detalle completo.
+        if resp.status_code >= 400:
+            # Intentar construir un mensaje amigable
+            message = None
+            if isinstance(data, dict):
+                message = data.get("message")
+                if not message and "detail" in data:
+                    # FastAPI ValidationError-style: detail: [{loc, msg, type}, ...]
+                    try:
+                        parts = [str(d.get("msg")) for d in data["detail"] if isinstance(d, dict)]
+                        message = "; ".join([p for p in parts if p])
+                    except Exception:
+                        pass
+            if not message:
+                message = resp.text or f"HTTP {resp.status_code}"
+
+            status = "validation_error" if resp.status_code == 422 else "error"
+
+            return CoveExternalResult(
+                status=status,
+                message=message,
+                cove_number=None,
+                vucem_operation_num=None,
+                raw_response={
+                    "status_code": resp.status_code,
+                    "body": data,
+                },
+            )
+
+        # 2xx: según la especificación del servicio externo, al menos devuelve:
+        # { "task_id": "...", "status": "...", "message": "..." }
+        raw_status = str(data.get("status") or "queued")
+        message = data.get("message")
+        external_task_id = data.get("task_id")
+
+        # Caso especial: algunos ambientes de VU regresan status="error" pero un mensaje
+        # tipo "Factura COVE iniciada para: ... Use el task_id para consultar el estado."
+        # que en realidad indica que la factura fue aceptada y quedó encolada en VU.
+        # En ese caso NO lo tratamos como error de negocio, sino como "en cola".
+        normalized_status = raw_status.lower()
+        if (
+            normalized_status == "error"
+            and isinstance(message, str)
+            and "Factura COVE iniciada para" in message
+        ):
+            status = "external_queued"
+        else:
+            status = raw_status
+
+        # El número de COVE normalmente se obtendrá vía /status/{task_id}; por ahora
+        # lo dejamos en None y exponemos la respuesta completa para inspección en UI.
+        return CoveExternalResult(
+            status=status,
+            message=message,
+            cove_number=None,
+            vucem_operation_num=None,
+            raw_response={
+                "task_id": external_task_id,
+                "status": status,
+                "message": message,
+                "raw": data,
+            },
+        )
+
+    def get_status(self, task_id: str) -> Dict[str, Any]:
+        """
+        Consulta el endpoint externo /api/v1/factura-cove/status/{task_id}
+        y devuelve el JSON de progreso/resultado tal cual lo envía el servicio.
+
+        Ejemplo de respuesta esperada (simplificada):
+        {
+          "task_id": "...",
+          "state": "PROGRESS" | "SUCCESS" | "FAILURE",
+          "result": null | {...},
+          "error": null | "...",
+          "progress": {
+            "current_step": "Consultando respuesta COVE con número de operación",
+            "progress": 10.5,
+            "total_steps": 12,
+            "task_id": "...",
+            "numero_operacion": "306658625"
+          }
+        }
+        """
+        url = f"{self.base_url.rstrip('/')}/api/v1/factura-cove/status/{task_id}"
+
+        # Igual que en generate_cove, desactivamos verify solo para entornos de dev.
+        with httpx.Client(timeout=30.0, verify=False) as client:
+            resp = client.get(url)
+
+        try:
+            data: Dict[str, Any] = resp.json()
+        except Exception:
+            data = {}
+
+        # Adjuntar metadatos mínimos de respuesta HTTP
+        data.setdefault("status_code", resp.status_code)
+
+        return data
+
--- a/backend/api/v1/modules/a76/factura_cove/routes.py
+++ b/backend/api/v1/modules/a76/factura_cove/routes.py
@@ -0,0 +1,149 @@
+from __future__ import annotations
+
+from typing import Any, Dict
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy.orm import Session
+
+from core.celery_app import celery_app
+from core.database import get_core_db
+from core.security import get_current_user, get_tenant_from_token, validate_access_to_resource
+
+from api.v1.modules.core.tasks_tracking import track_and_dispatch
+
+from .schemas import (
+    CoveEligibilityResponse,
+    FacturaCoveResponse,
+    GenerateCoveFromInvoiceRequest,
+)
+from .service import FacturaCoveDomainService
+from .tasks import factura_cove_generate
+
+router = APIRouter()
+
+
+@router.post(
+    "/invoices/{invoice_id}/cove",
+    response_model=FacturaCoveResponse,
+    summary="Generar COVE a partir de una factura (asíncrono)",
+)
+def trigger_cove_for_invoice(
+    invoice_id: int,
+    body: GenerateCoveFromInvoiceRequest,
+    db: Session = Depends(get_core_db),
+    current_user: Dict[str, Any] = Depends(get_current_user),
+):
+    """
+    Dispara la tarea Celery `factura_cove_generate` para una factura específica.
+
+    - Valida acceso a la compañía.
+    - Registra la tarea en el tracker de tareas.
+    - Retorna el `task_id` para hacer polling de estado desde el frontend.
+    """
+    tenant_id = validate_access_to_resource(db, body.company_id, current_user)
+
+    # Validación mínima de existencia/propiedad de la factura (el dominio hará validaciones más profundas).
+    from api.v1.modules.a76.invoices.models import InvoiceHeader
+
+    invoice: InvoiceHeader | None = db.get(InvoiceHeader, invoice_id)
+    if not invoice:
+        raise HTTPException(status_code=404, detail=f"Factura {invoice_id} no encontrada.")
+    if invoice.company_id != body.company_id or invoice.tenant_id != tenant_id:
+        raise HTTPException(
+            status_code=400,
+            detail="La factura no pertenece a la compañía o tenant actuales.",
+        )
+
+    task = track_and_dispatch(
+        db=db,
+        task=factura_cove_generate,
+        tenant_id=tenant_id,
+        company_id=body.company_id,
+        requested_by_user=(
+            current_user.get("preferred_username")
+            or current_user.get("email")
+            or current_user.get("sub")
+        ),
+        task_name="factura_cove_generate",
+        task_group="factura_cove",
+        task_origin="a76/factura_cove/invoices/cove",
+        args=[invoice_id, int(tenant_id), body.company_id],
+    )
+
+    return FacturaCoveResponse(
+        task_id=task.id,
+        status="queued",
+        message="Tarea de validación/generación de COVE encolada.",
+    )
+
+
+@router.get(
+    "/invoices/cove/{task_id}/status",
+    summary="Estado de tarea de COVE para factura",
+)
+def get_cove_status(task_id: str) -> Dict[str, Any]:
+    """
+    Consulta el estado de una tarea Celery de generación de COVE.
+
+    Retorna:
+      - state: 'PROCESSING' | 'SUCCESS' | 'FAILURE'
+      - info: { current: int, status: str }  (cuando state == 'PROCESSING')
+      - result: dict                          (cuando state == 'SUCCESS' o 'FAILURE')
+    """
+    task_result = celery_app.AsyncResult(task_id)
+
+    if task_result.state in ("PENDING", "STARTED"):
+        return {
+            "state": "PROCESSING",
+            "info": {"current": 0, "status": "Iniciando generación de COVE..."},
+        }
+
+    if task_result.state == "PROGRESS":
+        return {
+            "state": "PROCESSING",
+            "info": task_result.info or {"current": 0, "status": "Procesando COVE..."},
+        }
+
+    if task_result.state == "SUCCESS":
+        return {
+            "state": "SUCCESS",
+            "result": task_result.result,
+        }
+
+    error_info = task_result.result
+    if isinstance(error_info, Exception):
+        error_msg = str(error_info)
+    else:
+        error_msg = str(error_info) if error_info else "Error desconocido"
+
+    return {
+        "state": "FAILURE",
+        "result": error_msg,
+    }
+
+
+@router.get(
+    "/invoices/{invoice_id}/cove/eligibility",
+    response_model=CoveEligibilityResponse,
+    summary="Verifica si una factura puede generar COVE",
+)
+def check_cove_eligibility(
+    invoice_id: int,
+    company_id: int = Query(..., description="Company ID"),
+    db: Session = Depends(get_core_db),
+    current_user: Dict[str, Any] = Depends(get_current_user),
+):
+    """
+    Evalúa si la factura tiene todos los datos necesarios (VU, factura, partidas)
+    para poder generar un COVE. No dispara la tarea Celery.
+    """
+    tenant_id = get_tenant_from_token(current_user)
+    if not tenant_id:
+        raise HTTPException(status_code=400, detail="Tenant ID not found in token")
+
+    tenant_id_int = int(tenant_id)
+
+    service = FacturaCoveDomainService(db)
+    eligibility = service.check_eligibility(invoice_id=invoice_id, tenant_id=tenant_id_int, company_id=company_id)
+    return eligibility
+
--- a/backend/api/v1/modules/a76/factura_cove/schemas.py
+++ b/backend/api/v1/modules/a76/factura_cove/schemas.py
@@ -0,0 +1,135 @@
+from __future__ import annotations
+
+from datetime import datetime
+from decimal import Decimal
+from typing import List, Optional, Dict, Any
+
+from pydantic import BaseModel, EmailStr, Field
+
+
+class ConfiguracionVU(BaseModel):
+    """
+    Configuración de Ventanilla Única / VUCEM para generación de COVE.
+
+    Nota: estos campos se pueden poblar desde CustomsBrokerVU (web_service_user,
+    web_service_access_key, fiel_access_key, query_tax_id, etc.) o desde el
+    propio request de la API pública, según el flujo que se implemente.
+    """
+
+    rfc_usuario_vu: str = Field(..., max_length=30)
+    # Clave encriptada/token del webservice; puede ser larga (base64)
+    clave_webservice: str = Field(..., max_length=512)
+    archivo_cer_base64: str
+    archivo_key_base64: str
+    clave_fiel: str = Field(..., max_length=100)
+
+
+class PersonaCove(BaseModel):
+    tipo_identificador: str = Field(..., max_length=10)
+    identificacion: str = Field(..., max_length=30)
+    apellido_paterno: Optional[str] = Field(None, max_length=80)
+    apellido_materno: Optional[str] = Field(None, max_length=80)
+    nombre: Optional[str] = Field(None, max_length=80)
+    calle: Optional[str] = Field(None, max_length=120)
+    numero_exterior: Optional[str] = Field(None, max_length=20)
+    numero_interior: Optional[str] = Field(None, max_length=20)
+    colonia: Optional[str] = Field(None, max_length=120)
+    localidad: Optional[str] = Field(None, max_length=120)
+    municipio: Optional[str] = Field(None, max_length=120)
+    entidad_federativa: Optional[str] = Field(None, max_length=120)
+    pais: str = Field(..., max_length=3, description="País en formato ISO o catálogo VU")
+    codigo_postal: Optional[str] = Field(None, max_length=15)
+
+
+class DescripcionEspecifica(BaseModel):
+    marca: Optional[str] = Field(None, max_length=80)
+    modelo: Optional[str] = Field(None, max_length=80)
+    submodelo: Optional[str] = Field(None, max_length=80)
+    numero_serie: Optional[str] = Field(None, max_length=80)
+
+
+class MercanciaCove(BaseModel):
+    descripcion_generica: str = Field(..., max_length=500)
+    clave_unidad_medida: str = Field(..., max_length=10)
+    tipo_moneda: str = Field(..., max_length=5)
+    cantidad: Decimal = Field(..., gt=0)
+    valor_unitario: Decimal = Field(..., ge=0)
+    valor_total: Decimal = Field(..., ge=0)
+    valor_dolares: Optional[Decimal] = Field(None, ge=0)
+    descripcion_especifica: List[DescripcionEspecifica] = Field(default_factory=list)
+
+
+class FacturaCoveRequest(BaseModel):
+    """
+    Payload completo para generación de COVE.
+
+    Este modelo replica el contrato del servicio externo de COVE que se
+    mostró en la documentación compartida por el usuario.
+    """
+
+    configuracion_vu: ConfiguracionVU
+    rfc_consulta: str = Field(..., max_length=30)
+    tipo_figura: str = Field(..., max_length=10)
+    numero_factura: str = Field(..., max_length=50)
+    tipo_operacion: str = Field(..., max_length=10)
+    patente_aduanal: str = Field(..., max_length=10)
+    fecha_expedicion: datetime
+    observaciones: Optional[str] = Field(None, max_length=500)
+    correo_electronico: Optional[EmailStr] = None
+    tiene_subdivision: bool = False
+    certificado_origen: bool = False
+    numero_exportador_autorizado: Optional[str] = Field(None, max_length=50)
+    emisor: PersonaCove
+    destinatario: PersonaCove
+    mercancias: List[MercanciaCove] = Field(default_factory=list, min_length=1)
+
+
+class FacturaCoveResponse(BaseModel):
+    """
+    Respuesta base del endpoint público de generación de COVE.
+    Para el flujo asíncrono interno, solo usamos task_id/status/message.
+    """
+
+    task_id: Optional[str] = None
+    status: str
+    message: Optional[str] = None
+
+
+class GenerateCoveFromInvoiceRequest(BaseModel):
+    """
+    Request minimalista desde la vista de facturas.
+
+    Solo necesita el company_id porque el invoice_id viene en la URL y el
+    tenant_id se resuelve desde el token.
+    """
+
+    company_id: int
+    force_regen: Optional[bool] = False
+
+
+class GenerateCoveResult(BaseModel):
+    """
+    Resultado estándar que produce la tarea Celery factura_cove_generate.
+    """
+
+    status: str
+    message: Optional[str] = None
+    invoice_id: Optional[int] = None
+    cove_number: Optional[str] = None
+    vucem_operation_num: Optional[str] = None
+    # ID de tarea devuelto por el servicio externo de COVE (si aplica)
+    external_task_id: Optional[str] = None
+    # Respuesta cruda devuelta por el servicio externo (POST generar-factura-cove)
+    external_response: Optional[Dict[str, Any]] = None
+    errors: Optional[list[dict]] = None
+
+
+class CoveEligibilityIssue(BaseModel):
+    field: str
+    message: str
+
+
+class CoveEligibilityResponse(BaseModel):
+    can_generate: bool
+    reasons: list[CoveEligibilityIssue] = Field(default_factory=list)
+
--- a/backend/api/v1/modules/a76/factura_cove/service.py
+++ b/backend/api/v1/modules/a76/factura_cove/service.py
@@ -0,0 +1,621 @@
+from __future__ import annotations
+
+import base64
+from dataclasses import dataclass
+from decimal import Decimal
+from typing import List, Tuple
+
+from sqlalchemy.orm import Session
+
+from core.database import CoreSessionLocal
+from core.exceptions import ValidationException, ErrorCollector
+from core.storage_s3 import get_object_bytes, object_exists
+
+from api.v1.modules.a76.invoices.models import InvoiceHeader
+from api.v1.modules.a76.customs_brokers import models as cb_models
+from api.v1.modules.a76.clients_and_providers.models import ClientProvider
+from api.v1.modules.a76.general_catalogs.company.models import Company
+from api.v1.modules.a76.items.models import LineItem
+
+from .schemas import (
+    ConfiguracionVU,
+    CoveEligibilityIssue,
+    CoveEligibilityResponse,
+    FacturaCoveRequest,
+    MercanciaCove,
+    PersonaCove,
+)
+
+
+@dataclass
+class InvoiceContext:
+    invoice: InvoiceHeader
+    broker: cb_models.CustomsBroker | None
+    vu: cb_models.CustomsBrokerVU | None
+
+
+class FacturaCoveDomainService:
+    """
+    Servicio de dominio para validar y construir el payload de COVE a partir de una factura.
+
+    NOTA IMPORTANTE:
+    ----------------
+    Este servicio prepara la estructura de datos y realiza validaciones de negocio,
+    pero **no** realiza todavía la llamada HTTP al webservice de COVE. Eso se puede
+    implementar posteriormente en un servicio dedicado (p. ej. CoveExternalService).
+    """
+
+    def __init__(self, db: Session):
+        self.db = db
+
+    def _load_context(self, invoice_id: int, tenant_id: int, company_id: int) -> InvoiceContext:
+        invoice: InvoiceHeader | None = self.db.get(InvoiceHeader, invoice_id)
+        if not invoice:
+            raise ValidationException(
+                "Factura no encontrada",
+                errors=[{"field": "invoice_id", "message": f"Factura {invoice_id} no encontrada"}],
+            )
+
+        if invoice.company_id != company_id or invoice.tenant_id != tenant_id:
+            raise ValidationException(
+                "Factura no pertenece a la compañía/tenant actual",
+                errors=[
+                    {
+                        "field": "invoice_id",
+                        "message": "La factura no pertenece a la compañía o tenant actuales",
+                    }
+                ],
+            )
+
+        compliance = invoice.compliance_mx
+        broker = None
+        vu = None
+
+        if compliance and compliance.customs_broker_id:
+            broker = self.db.get(cb_models.CustomsBroker, compliance.customs_broker_id)
+            if broker:
+                vu = broker.vu
+
+        return InvoiceContext(invoice=invoice, broker=broker, vu=vu)
+
+    def _build_configuracion_vu(self, ctx: InvoiceContext, errors: ErrorCollector) -> ConfiguracionVU | None:
+        """
+        Construye la sección configuracion_vu usando CustomsBrokerVU + S3.
+
+        Lee los archivos .cer y .key desde almacenamiento de objetos, los
+        convierte a base64 y construye un ConfiguracionVU listo para enviar
+        al API externo de COVE.
+        """
+        vu = ctx.vu
+
+        if not (vu.web_service_user and vu.web_service_access_key and vu.fiel_access_key):
+            errors.add_error(
+                field="vu",
+                message="Faltan credenciales de web service o clave FIEL en VU",
+                solution=[
+                    "Captura usuario y clave de web service (web_service_user, web_service_access_key) "
+                    "y la clave FIEL en la configuración VU del agente aduanal."
+                ],
+                code="MISSING_VU_CREDENTIALS",
+            )
+
+        if not (vu.certificate_path and vu.key_path):
+            errors.add_error(
+                field="vu",
+                message="No hay rutas de certificado o llave en VU",
+                solution=[
+                    "Sube el certificado (.cer) y la llave (.key) del VU desde el módulo de agentes aduanales."
+                ],
+                code="MISSING_VU_CERT_KEY",
+            )
+            return None
+
+        # Convertir archivos .cer y .key de S3 a base64
+        cer_b64 = None
+        key_b64 = None
+
+        try:
+            if not object_exists(vu.certificate_path):
+                errors.add_error(
+                    field="vu.certificate_path",
+                    message="El certificado VU no existe en el almacenamiento de objetos",
+                    solution=["Vuelve a subir el certificado VU en el agente aduanal."],
+                    code="VU_CERT_NOT_FOUND",
+                )
+            else:
+                cer_bytes = get_object_bytes(vu.certificate_path)
+                cer_b64 = base64.b64encode(cer_bytes).decode("ascii")
+
+            if not object_exists(vu.key_path):
+                errors.add_error(
+                    field="vu.key_path",
+                    message="La llave VU no existe en el almacenamiento de objetos",
+                    solution=["Vuelve a subir la llave VU en el agente aduanal."],
+                    code="VU_KEY_NOT_FOUND",
+                )
+            else:
+                key_bytes = get_object_bytes(vu.key_path)
+                key_b64 = base64.b64encode(key_bytes).decode("ascii")
+        except Exception as exc:  # pragma: no cover - errores de IO externos
+            errors.add_error(
+                field="vu",
+                message="Error leyendo certificados VU desde almacenamiento de objetos.",
+                solution=["Verifica la configuración de MinIO/S3 y las rutas de certificados/llaves en la configuración VU."],
+                code="VU_STORAGE_ERROR",
+            )
+
+        if errors.has_errors():
+            return None
+
+        rfc_usuario_vu = vu.query_tax_id or ""
+
+        return ConfiguracionVU(
+            rfc_usuario_vu=rfc_usuario_vu,
+            clave_webservice=vu.web_service_access_key or "",
+            archivo_cer_base64=cer_b64 or "",
+            archivo_key_base64=key_b64 or "",
+            clave_fiel=vu.fiel_access_key or "",
+        )
+
+    def _clientprovider_to_persona(self, cp: ClientProvider) -> PersonaCove:
+        """
+        Construye una PersonaCove a partir de un ClientProvider + su dirección.
+        No expone IDs internos; solo valores normalizados.
+        """
+        addr = cp.address
+
+        tipo_nat = (cp.type_nat_foreign or "").strip().upper()
+        tipo_identificador = "0" if tipo_nat == "E" else "1"
+        identificacion = (cp.rfc or "").strip().upper()
+
+        # País: normalizar a código de 3 caracteres (ISO o catálogo VU).
+        raw_country = (addr.country or "") if addr and getattr(addr, "country", None) else ""
+        country_code = raw_country.strip().upper()[:3] if raw_country else ""
+
+        return PersonaCove(
+            tipo_identificador=tipo_identificador,
+            identificacion=identificacion,
+            apellido_paterno="",
+            apellido_materno="",
+            nombre=(cp.name or cp.short_name or "").strip() or None,
+            calle=(addr.streets or "").strip() if addr and addr.streets else None,
+            numero_exterior=(addr.exterior_number or "").strip()
+            if addr and addr.exterior_number
+            else None,
+            # El API de COVE exige texto (no null) para numero_interior y municipio.
+            # Si no hay valor, enviamos cadena vacía.
+            numero_interior=(addr.interior_number or "").strip()
+            if addr
+            else "",
+            colonia=(addr.neighborhood or "").strip()
+            if addr and addr.neighborhood
+            else None,
+            localidad=(addr.city or "").strip() if addr and addr.city else None,
+            municipio=(addr.municipality or "").strip()
+            if addr
+            else "",
+            entidad_federativa=(addr.state or "").strip()
+            if addr and addr.state
+            else None,
+            pais=country_code,
+            codigo_postal=(addr.postal_code or "").strip()
+            if addr and addr.postal_code
+            else None,
+        )
+
+    def _company_to_persona(self, company: Company) -> PersonaCove:
+        """
+        Construye una PersonaCove a partir de Company + su dirección principal.
+        """
+        # Tomar dirección 'main' si existe; si no, la primera.
+        addr = None
+        for a in company.addresses or []:
+            if getattr(a, "address_type", None) == "main":
+                addr = a
+                break
+        if addr is None and company.addresses:
+            addr = company.addresses[0]
+
+        # País: normalizar a código de 3 caracteres (ISO o catálogo VU).
+        raw_country = (addr.country or "") if addr and getattr(addr, "country", None) else ""
+        country_code = raw_country.strip().upper()[:3] if raw_country else ""
+
+        return PersonaCove(
+            tipo_identificador="1",  # Empresa mexicana por defecto
+            identificacion=(company.rfc or "").strip().upper(),
+            apellido_paterno="",
+            apellido_materno="",
+            nombre=(company.name or "").strip() or None,
+            calle=(addr.street or "").strip() if addr and addr.street else None,
+            numero_exterior=(addr.exterior_number or "").strip()
+            if addr and addr.exterior_number
+            else None,
+            numero_interior=(addr.interior_number or "").strip()
+            if addr
+            else "",
+            colonia=(addr.neighborhood or "").strip()
+            if addr and addr.neighborhood
+            else None,
+            localidad=(addr.city or "").strip() if addr and addr.city else None,
+            municipio=(addr.municipality or "").strip()
+            if addr
+            else "",
+            entidad_federativa=(addr.state or "").strip()
+            if addr and addr.state
+            else None,
+            pais=country_code,
+            codigo_postal=(addr.postal_code or "").strip()
+            if addr and addr.postal_code
+            else None,
+        )
+
+    def _build_personas(
+        self, ctx: InvoiceContext, errors: ErrorCollector
+    ) -> Tuple[PersonaCove | None, PersonaCove | None]:
+        """
+        Construye emisor (exportador) y destinatario (importador) a partir de:
+        - InvoiceComplianceMx.provider_id / sold_to_id / shipped_to_id
+        - Catálogo de clientes/proveedores
+        - Company (datos de la propia empresa) como último recurso
+        """
+        compliance = ctx.invoice.compliance_mx
+        tenant_id = getattr(ctx.invoice, "tenant_id", None)
+        company_id = getattr(ctx.invoice, "company_id", None)
+
+        emisor_persona: PersonaCove | None = None
+        destinatario_persona: PersonaCove | None = None
+
+        # --- Emisor: proveedor/exportador ---
+        if compliance and compliance.provider_id:
+            provider = (
+                self.db.query(ClientProvider)
+                .filter(
+                    ClientProvider.id == compliance.provider_id,
+                    ClientProvider.tenant_id == tenant_id,
+                    ClientProvider.company_id == company_id,
+                )
+                .first()
+            )
+            if provider:
+                emisor_persona = self._clientprovider_to_persona(provider)
+            else:
+                errors.add_error(
+                    field="emisor",
+                    message="No se encontró el proveedor/exportador asociado a la factura",
+                    solution=[
+                        "Verifica que el proveedor/exportador exista en el catálogo y que el invoice_compliance_mx.provider_id sea válido."
+                    ],
+                    code="EMISOR_PROVIDER_NOT_FOUND",
+                )
+        else:
+            errors.add_error(
+                field="emisor",
+                message="La factura no tiene proveedor/exportador configurado en cumplimiento (provider_id)",
+                solution=[
+                    "Configura el proveedor/exportador (provider_id) en los datos de cumplimiento de la factura."
+                ],
+                code="EMISOR_PROVIDER_MISSING",
+            )
+
+        # --- Destinatario: importador mexicano ---
+        dest_client: ClientProvider | None = None
+        if compliance and compliance.sold_to_id:
+            dest_client = (
+                self.db.query(ClientProvider)
+                .filter(
+                    ClientProvider.id == compliance.sold_to_id,
+                    ClientProvider.tenant_id == tenant_id,
+                    ClientProvider.company_id == company_id,
+                )
+                .first()
+            )
+        elif compliance and compliance.shipped_to_id:
+            dest_client = (
+                self.db.query(ClientProvider)
+                .filter(
+                    ClientProvider.id == compliance.shipped_to_id,
+                    ClientProvider.tenant_id == tenant_id,
+                    ClientProvider.company_id == company_id,
+                )
+                .first()
+            )
+
+        if dest_client:
+            destinatario_persona = self._clientprovider_to_persona(dest_client)
+        else:
+            # Fallback: usar la empresa de A76 como importador/destinatario
+            if company_id is not None:
+                company = (
+                    self.db.query(Company)
+                    .filter(Company.id == company_id, Company.tenant_id == tenant_id)
+                    .first()
+                )
+            else:
+                company = None
+
+            if company:
+                destinatario_persona = self._company_to_persona(company)
+            else:
+                errors.add_error(
+                    field="destinatario",
+                    message="No se pudo determinar el destinatario (cliente/importador) para COVE",
+                    solution=[
+                        "Configura sold_to_id o shipped_to_id en los datos de cumplimiento de la factura, "
+                        "o asegura que la compañía tenga datos de dirección configurados."
+                    ],
+                    code="DESTINATARIO_NOT_FOUND",
+                )
+
+        return emisor_persona, destinatario_persona
+
+    def _build_mercancias(self, ctx: InvoiceContext, errors: ErrorCollector) -> list[MercanciaCove]:
+        """
+        Construye la lista de mercancías COVE a partir de las partidas (LineItem)
+        asociadas a la factura.
+        """
+        # Obtener todas las partidas de la factura
+        lines: list[LineItem] = (
+            self.db.query(LineItem)
+            .filter(LineItem.invoice_id == ctx.invoice.id)
+            .all()
+        )
+
+        if not lines:
+            errors.add_error(
+                field="mercancias",
+                message="La factura no tiene partidas (LineItem) asociadas",
+                solution=[
+                    "Verifica que la factura tenga partidas capturadas antes de generar COVE."
+                ],
+                code="NO_LINE_ITEMS_FOR_COVE",
+            )
+            return []
+
+        mercancias: list[MercanciaCove] = []
+
+        # Determinar moneda base: usamos la moneda de la factura tal como
+        # la maneja el módulo de invoices. En financials se normaliza:
+        # - currency: 'foreign' | 'local' | 'manual'
+        # - currency_type: código de catálogo (ej. 'USD', 'MXN'), upper.
+        fin = getattr(ctx.invoice, "financials", None)
+        raw_currency_type = getattr(fin, "currency_type", None)
+        invoice_currency = (raw_currency_type or "").strip().upper() or "USD"
+
+        for line in lines:
+            qty_model = line.quantity
+            fin_model = line.financial
+            desc_model = line.description
+
+            if not qty_model or qty_model.quantity is None or qty_model.quantity <= 0:
+                # Saltar partidas sin cantidad válida
+                continue
+
+            cantidad = Decimal(str(qty_model.quantity))
+
+            # Descripción genérica: priorizar descripción de parte / inglés / español
+            descripcion = ""
+            if desc_model:
+                descripcion = (
+                    desc_model.part_description
+                    or desc_model.description_english
+                    or desc_model.description_spanish
+                    or ""
+                ).strip()
+            if not descripcion:
+                descripcion = (line.line_concept or "").strip()
+            if not descripcion:
+                descripcion = "SIN DESCRIPCION"
+
+            # Clave unidad de medida: usar OMA/customs si están disponibles
+            clave_unidad = ""
+            uom = line.unit_of_measure_info
+            if uom:
+                if uom.oma_unit and uom.oma_unit.code:
+                    clave_unidad = uom.oma_unit.code
+                elif uom.customs_unit and uom.customs_unit.code:
+                    clave_unidad = uom.customs_unit.code
+                elif uom.code:
+                    clave_unidad = uom.code
+
+            if not clave_unidad:
+                errors.add_error(
+                    field="mercancias",
+                    message="No se pudo determinar la unidad de medida para una partida de la factura",
+                    solution=[
+                        "Asegúrate de que la partida tenga una unidad de medida configurada en el catálogo "
+                        "y que esté ligada a una clave OMA/aduana válida."
+                    ],
+                    code="MERCANCIA_UOM_MISSING",
+                )
+                continue
+
+            # Moneda y valores: usamos la moneda de la factura (3 caracteres)
+            tipo_moneda = invoice_currency
+
+            valor_total = Decimal("0")
+            valor_dolares = Decimal("0")
+            valor_unitario = Decimal("0")
+
+            if fin_model:
+                if tipo_moneda == "USD":
+                    base_total = (
+                        fin_model.value_total_usd
+                        or fin_model.value_usd
+                        or Decimal("0")
+                    )
+                else:
+                    # Para otras monedas, usamos el total en MXN/MC como respaldo
+                    base_total = (
+                        fin_model.value_total_mxn
+                        or fin_model.value_mxn
+                        or fin_model.value_total_mc
+                        or fin_model.value_mc
+                        or Decimal("0")
+                    )
+
+                valor_total = Decimal(str(base_total or 0))
+                if cantidad > 0:
+                    valor_unitario = (valor_total / cantidad).quantize(
+                        Decimal("0.000001")
+                    )
+                else:
+                    valor_unitario = Decimal("0")
+
+                # Valor en dólares: si ya existe, lo usamos; si no, convertimos suponiendo que los totales ya están en USD
+                if fin_model.value_total_usd:
+                    valor_dolares = Decimal(str(fin_model.value_total_usd))
+                elif tipo_moneda == "USD":
+                    valor_dolares = valor_total
+                else:
+                    valor_dolares = Decimal("0")
+            else:
+                errors.add_error(
+                    field="mercancias",
+                    message="La partida de la factura no tiene información financiera asociada",
+                    solution=[
+                        "Verifica que las partidas tengan datos financieros (LineFinancial) antes de generar COVE."
+                    ],
+                    code="MERCANCIA_FINANCIAL_MISSING",
+                )
+                continue
+
+            mercancia = MercanciaCove(
+                descripcion_generica=descripcion[:500],
+                clave_unidad_medida=clave_unidad,
+                tipo_moneda=tipo_moneda,
+                cantidad=cantidad,
+                valor_unitario=valor_unitario,
+                valor_total=valor_total,
+                valor_dolares=valor_dolares,
+                descripcion_especifica=[],
+            )
+            mercancias.append(mercancia)
+
+        if not mercancias and not errors.has_errors():
+            errors.add_error(
+                field="mercancias",
+                message="No se generó ninguna mercancía COVE a partir de las partidas de la factura",
+                solution=[
+                    "Verifica que las partidas tengan cantidad y datos financieros válidos antes de generar COVE."
+                ],
+                code="NO_MERCANCIAS_GENERATED",
+            )
+
+        return mercancias
+
+    def build_factura_cove_request(
+        self, invoice_id: int, tenant_id: int, company_id: int
+    ) -> FacturaCoveRequest:
+        """
+        Construye el FacturaCoveRequest completo a partir de una factura,
+        validando prerrequisitos de VU, factura y mapeos.
+        """
+        ctx = self._load_context(invoice_id, tenant_id, company_id)
+        errors = ErrorCollector()
+
+        # Validaciones básicas de factura
+        if not ctx.invoice.invoice_number:
+            errors.add_error(
+                field="invoice.invoice_number",
+                message="La factura no tiene número de factura",
+                solution=["Captura el número de factura antes de generar COVE."],
+                code="MISSING_INVOICE_NUMBER",
+            )
+
+        # Construir configuración VU (puede agregar errores)
+        configuracion_vu = self._build_configuracion_vu(ctx, errors)
+
+        # Personas y mercancías (por ahora placeholders con errores explícitos)
+        emisor, destinatario = self._build_personas(ctx, errors)
+        mercancias = self._build_mercancias(ctx, errors)
+
+        if errors.has_errors():
+            # Levantamos ValidationException con todos los errores
+            raise ValidationException("No se puede generar COVE desde la factura", errors=errors.get_errors())
+
+        # Campos genéricos que se pueden poblar de forma segura
+        raw_tipo_operacion = (ctx.invoice.operation_type or "").strip()
+        # Normalizar tipo_operacion a MAYÚSCULAS (ej. IMP, EXP) con longitud acotada
+        tipo_operacion = raw_tipo_operacion.upper()[:10]
+        numero_factura = (ctx.invoice.invoice_number or "").strip()[:50]
+        fecha_expedicion = ctx.invoice.invoice_date or ctx.invoice.emission_date or ctx.invoice.capture_date
+
+        if not fecha_expedicion:
+            raise ValidationException(
+                "Falta fecha de expedición de factura",
+                errors=[
+                    {
+                        "field": "invoice.invoice_date",
+                        "message": "La factura no tiene fecha de expedición/emisión/captura",
+                        "solution": ["Captura la fecha de la factura antes de generar COVE."],
+                    }
+                ],
+            )
+
+        # Patente aduanal en mayúsculas y acotada a 10 caracteres
+        raw_patente = (ctx.broker.license if ctx.broker else "") or ""
+        patente_aduanal = raw_patente.strip().upper()[:10]
+
+        return FacturaCoveRequest(
+            configuracion_vu=configuracion_vu,
+            rfc_consulta=(
+                (ctx.vu.query_tax_id or "").strip().upper() if ctx.vu and ctx.vu.query_tax_id else ""
+            ),
+            tipo_figura=(
+                (ctx.vu.vu_figure_type or "").strip().upper() if ctx.vu and ctx.vu.vu_figure_type else ""
+            ),
+            numero_factura=numero_factura,
+            tipo_operacion=tipo_operacion,
+            patente_aduanal=patente_aduanal,
+            fecha_expedicion=fecha_expedicion,
+            observaciones=ctx.invoice.vu_observations or None,
+            # Correo hardcodeado temporalmente para pruebas de COVE
+            correo_electronico="hreyes@aduanasoft.com.mx",
+            tiene_subdivision=bool(ctx.invoice.logistics and ctx.invoice.logistics.is_subdivision),
+            certificado_origen=False,
+            numero_exportador_autorizado=None,
+            emisor=emisor,  # type: ignore[arg-type]
+            destinatario=destinatario,  # type: ignore[arg-type]
+            mercancias=mercancias,
+        )
+
+    def check_eligibility(self, invoice_id: int, tenant_id: int, company_id: int) -> CoveEligibilityResponse:
+        """
+        Versión "ligera" para frontend: evalúa si la factura puede generar COVE
+        e informa por qué no, sin disparar la tarea Celery.
+        """
+        db = self.db or CoreSessionLocal()
+        errors = ErrorCollector()
+
+        try:
+            ctx = self._load_context(invoice_id, tenant_id, company_id)
+            # Reutilizamos solo las validaciones, sin necesidad de devolver el request completo
+            if not ctx.invoice.invoice_number:
+                errors.add_error(
+                    field="invoice.invoice_number",
+                    message="La factura no tiene número de factura",
+                    solution=["Captura el número de factura antes de generar COVE."],
+                    code="MISSING_INVOICE_NUMBER",
+                )
+
+            self._build_configuracion_vu(ctx, errors)
+            self._build_personas(ctx, errors)
+            self._build_mercancias(ctx, errors)
+        except ValidationException as exc:
+            # Errores de load_context (factura no existe, compañía distinta, etc.)
+            return CoveEligibilityResponse(
+                can_generate=False,
+                reasons=[CoveEligibilityIssue(field=e.get("field", ""), message=e.get("message", "")) for e in exc.errors],
+            )
+
+        if not errors.has_errors():
+            return CoveEligibilityResponse(can_generate=True, reasons=[])
+
+        return CoveEligibilityResponse(
+            can_generate=False,
+            reasons=[
+                CoveEligibilityIssue(field=e.get("field", ""), message=e.get("message", ""))
+                for e in errors.get_errors()
+            ],
+        )
+
--- a/backend/api/v1/modules/a76/factura_cove/tasks.py
+++ b/backend/api/v1/modules/a76/factura_cove/tasks.py
@@ -0,0 +1,215 @@
+from __future__ import annotations
+
+import logging
+import time
+from typing import Any, Dict
+
+from celery import Task
+
+from core.celery_app import celery_app
+from core.database import CoreSessionLocal
+from core.exceptions import ValidationException
+
+from .service import FacturaCoveDomainService
+from .schemas import GenerateCoveResult
+from .external_service import CoveExternalService, CoveExternalResult
+
+logger = logging.getLogger(__name__)
+
+
+def _progress(task: Task, current: int, status: str) -> None:
+    task.update_state(state="PROGRESS", meta={"current": current, "status": status})
+
+
+def _poll_external_status(
+    task: Task, external: CoveExternalService, external_task_id: str, timeout_seconds: int = 300
+) -> CoveExternalResult:
+    """
+    Realiza polling al endpoint externo de status de COVE hasta obtener un estado final
+    o agotar el timeout.
+    """
+    start = time.time()
+    last_payload: Dict[str, Any] = {}
+
+    while True:
+        if time.time() - start > timeout_seconds:
+            logger.error("Timeout consultando estado de COVE para external_task_id=%s", external_task_id)
+            return CoveExternalResult(
+                status="error",
+                message="Timeout consultando estado de COVE en Ventanilla Única.",
+                cove_number=None,
+                vucem_operation_num=None,
+                raw_response={"last_status": last_payload, "external_task_id": external_task_id},
+            )
+
+        try:
+            status_payload = external.get_status(external_task_id)
+        except Exception as exc:  # pragma: no cover - errores HTTP inesperados
+            logger.exception("Error consultando estado externo de COVE")
+            return CoveExternalResult(
+                status="error",
+                message=f"Error consultando estado de COVE en Ventanilla Única: {exc}",
+                cove_number=None,
+                vucem_operation_num=None,
+                raw_response={"last_status": last_payload, "external_task_id": external_task_id},
+            )
+
+        last_payload = status_payload or {}
+        state = str(last_payload.get("state") or "").upper()
+        progress = last_payload.get("progress") or {}
+        # En el ejemplo: progress.progress (float 0-100), progress.current_step (texto), numero_operacion
+        try:
+            percent = float(progress.get("progress", 0.0))
+        except (TypeError, ValueError):
+            percent = 0.0
+        current_step = progress.get("current_step") or "Consultando estado de COVE en Ventanilla Única..."
+        numero_operacion = progress.get("numero_operacion") or last_payload.get("numero_operacion")
+        if numero_operacion:
+            current_step = f"{current_step} (Operación: {numero_operacion})"
+
+        # Actualizar progreso para que el frontend lo vea en el diálogo
+        _progress(task, int(percent), str(current_step))
+
+        # Estados intermedios: seguimos pollendo
+        if state in {"PENDING", "STARTED", "PROGRESS"} or not state:
+            time.sleep(5)
+            continue
+
+        # Estado final: SUCCESS / FAILURE u otros
+        result_payload = last_payload.get("result") or {}
+        error_text = last_payload.get("error")
+
+        if state in {"SUCCESS", "COMPLETED"}:
+            cove_number = result_payload.get("cove_number") or result_payload.get("cove")
+            vucem_operation_num = result_payload.get("vucem_operation_num") or result_payload.get(
+                "numero_operacion"
+            )
+            message = result_payload.get("message") or last_payload.get("message") or "COVE generado correctamente."
+
+            return CoveExternalResult(
+                status="success",
+                message=message,
+                cove_number=cove_number,
+                vucem_operation_num=vucem_operation_num,
+                raw_response={**last_payload, "external_task_id": external_task_id},
+            )
+
+        # Cualquier otro estado lo tratamos como error
+        message = error_text or result_payload.get("message") or last_payload.get("message") or state
+
+        return CoveExternalResult(
+            status="error",
+            message=str(message),
+            cove_number=None,
+            vucem_operation_num=None,
+            raw_response={**last_payload, "external_task_id": external_task_id},
+        )
+
+
+@celery_app.task(bind=True, name="factura_cove_generate")
+def factura_cove_generate(self: Task, invoice_id: int, tenant_id: int, company_id: int) -> dict:
+    """
+    Tarea Celery para preparar (y en el futuro generar) un COVE a partir de una factura.
+
+    Actualmente:
+      - Valida prerrequisitos de factura y configuración VU.
+      - Construye el payload FacturaCoveRequest (sin llamar aún al webservice externo).
+      - Devuelve un resultado estándar indicando éxito o errores de validación.
+
+    En el futuro se puede extender para:
+      - Invocar al servicio externo de COVE.
+      - Persistir número de COVE / operación VUCEM en la factura.
+    """
+    db = CoreSessionLocal()
+
+    try:
+        _progress(self, 5, "Validando factura para COVE...")
+        service = FacturaCoveDomainService(db)
+
+        # Esta llamada valida todo y construye el payload; si algo falla, lanza ValidationException
+        request_payload = service.build_factura_cove_request(
+            invoice_id=invoice_id, tenant_id=tenant_id, company_id=company_id
+        )
+
+        _progress(self, 80, "Enviando solicitud al servicio COVE...")
+
+        # Integración externa que encola la generación de COVE en Ventanilla Única
+        external = CoveExternalService()
+        external_result = external.generate_cove(request_payload)
+
+        # Si el servicio externo devolvió un error inmediato (por ejemplo 422),
+        # devolvemos ese resultado tal cual sin hacer polling adicional.
+        if external_result.status in {"error", "validation_error"}:
+            result = GenerateCoveResult(
+                status=external_result.status,
+                message=external_result.message,
+                invoice_id=invoice_id,
+                cove_number=external_result.cove_number,
+                vucem_operation_num=external_result.vucem_operation_num,
+                external_task_id=(
+                    external_result.raw_response.get("task_id") if external_result.raw_response else None
+                ),
+                external_response=external_result.raw_response,
+                errors=None,
+            )
+            return result.model_dump()
+
+        external_task_id = (
+            external_result.raw_response.get("task_id") if external_result.raw_response else None
+        )
+
+        # Si la factura quedó encolada en VU y tenemos un task_id externo, hacemos polling
+        # al endpoint de status para acompañar el progreso completo hasta obtener COVE.
+        if external_task_id and external_result.status in {"external_queued", "queued", "success"}:
+            _progress(
+                self,
+                85,
+                "Factura enviada a Ventanilla Única, consultando estado de COVE...",
+            )
+            final_external = _poll_external_status(self, external, external_task_id)
+        else:
+            # Fallback: usamos el resultado tal cual devolvió el endpoint de generación
+            final_external = external_result
+
+        _progress(self, 100, "Proceso de COVE finalizado.")
+
+        result = GenerateCoveResult(
+            status=final_external.status,
+            message=final_external.message,
+            invoice_id=invoice_id,
+            cove_number=final_external.cove_number,
+            vucem_operation_num=final_external.vucem_operation_num,
+            external_task_id=(
+                final_external.raw_response.get("external_task_id")
+                or final_external.raw_response.get("task_id")
+                if final_external.raw_response
+                else None
+            ),
+            external_response=final_external.raw_response,
+            errors=None,
+        )
+        return result.model_dump()
+
+    except ValidationException as exc:
+        db.rollback()
+        logger.info("Validation error in factura_cove_generate: %s", exc.message)
+        result = GenerateCoveResult(
+            status="validation_error",
+            message=exc.message,
+            invoice_id=invoice_id,
+            errors=exc.errors,
+        )
+        return result.model_dump()
+    except Exception as exc:  # pragma: no cover - errores inesperados de runtime
+        db.rollback()
+        logger.exception("Unexpected error in factura_cove_generate")
+        result = GenerateCoveResult(
+            status="error",
+            message=str(exc),
+            invoice_id=invoice_id,
+            errors=None,
+        )
+        return result.model_dump()
+    finally:
+        db.close()
+
--- a/backend/api/v1/modules/a76/general_catalogs/company/dto.py
+++ b/backend/api/v1/modules/a76/general_catalogs/company/dto.py
@@ -57,7 +57,7 @@ class CompanyCreateDTO(BaseModel):
    )

    # Configuration
-    logo: Optional[str] = Field(None, max_length=255, description="Company logo")
+    logo: Optional[str] = Field(None, max_length=512, description="Company logo (path local o clave S3)")
    has_express_line: Optional[bool] = Field(None, description="Has express line")
    order_format_type: Optional[str] = Field(
        None, max_length=19, description="Order format type"
--- a/backend/api/v1/modules/a76/general_catalogs/company/models.py
+++ b/backend/api/v1/modules/a76/general_catalogs/company/models.py
@@ -60,7 +60,7 @@ class Company(Base, TimestampMixin):
    position: Mapped[Optional[str]] = mapped_column(String(30))

    # Configuración básica
-    logo: Mapped[Optional[str]] = mapped_column(String(255))
+    logo: Mapped[Optional[str]] = mapped_column(String(512))
    has_express_line: Mapped[Optional[bool]] = mapped_column(Boolean, default=False, server_default="false")
    order_format_type: Mapped[Optional[str]] = mapped_column(String(19))
    is_service_company: Mapped[Optional[bool]] = mapped_column(Boolean, default=False, server_default="false")
--- a/backend/api/v1/modules/a76/general_catalogs/company/routes.py
+++ b/backend/api/v1/modules/a76/general_catalogs/company/routes.py
@@ -2,20 +2,23 @@
 Rutas para gestión de empresa
 """

+import logging
+import mimetypes
 import os
 import shutil
 from typing import List, Optional
-import os
-import shutil
 from pathlib import Path
 from datetime import datetime

 from fastapi import APIRouter, Depends, HTTPException, status, File, UploadFile
-from fastapi.responses import FileResponse
+from fastapi.responses import FileResponse, Response
 from sqlalchemy.orm import Session

+from core.config import settings
 from core.database import get_core_db
-from core.security import get_current_user, validate_access_to_resource
+from core.s3_keys import company_certificate_key, company_logo_key
+from core.storage_s3 import delete_object_if_exists, get_object_bytes, put_object_bytes
+from core.security import get_current_user, get_tenant_from_token, validate_access_to_resource
 from .....common.tenant_crud_routes import TenantCRUDRoutes
 from .dto import CompanyCreateDTO, CompanyResponseDTO, CompanyUpdateDTO
 from .models import Company
@@ -26,6 +29,44 @@ UPLOAD_DIR = "uploads/companies"
 ALLOWED_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp"}
 MAX_FILE_SIZE = 5 * 1024 * 1024  # 5MB

+logger = logging.getLogger(__name__)
+
+
+def _resolve_tenant_id_int(current_user: dict) -> int:
+    """Misma lógica que validate_access_to_resource: entero estable para BD y claves S3."""
+    tid = get_tenant_from_token(current_user)
+    if tid is not None:
+        return int(tid)
+    raw = current_user.get("tenant_id")
+    if isinstance(raw, list) and raw:
+        raw = raw[0]
+    if raw is None:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Tenant ID not found in user data",
+        )
+    try:
+        return int(raw)
+    except (TypeError, ValueError):
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Invalid tenant ID in token",
+        )
+
+
+def _is_s3_object_key(ref: Optional[str]) -> bool:
+    return bool(ref and ref.startswith("tenants/"))
+
+
+def _remove_stored_file(ref: str) -> None:
+    if _is_s3_object_key(ref):
+        delete_object_if_exists(ref)
+    elif ref and os.path.isfile(ref):
+        try:
+            os.remove(ref)
+        except OSError:
+            pass
+
 # Main router that includes base CRUD
 router = APIRouter(prefix="/company")

@@ -219,14 +260,20 @@ async def get_company_logo_image(
    if not company or not company.logo:
        raise HTTPException(status_code=404, detail="Logo not found")

+    if _is_s3_object_key(company.logo):
+        try:
+            data = get_object_bytes(company.logo)
+        except Exception:
+            raise HTTPException(status_code=404, detail="Logo file not found on server")
+        media = mimetypes.guess_type(company.logo)[0] or "image/jpeg"
+        return Response(content=data, media_type=media)
+
    file_path = Path(company.logo)
    if not file_path.exists():
-        # Fallback for old paths or moved files
-        # Check if it exists in the 'standard' location even if DB thinks otherwise
        standard_path = Path(f"app_data/logos/{company_id}") / file_path.name
        if standard_path.exists():
            return FileResponse(standard_path)
-            
+
        raise HTTPException(status_code=404, detail="Logo file not found on server")

    return FileResponse(file_path)
@@ -272,12 +319,7 @@ async def upload_company_logo(
    current_user: dict = Depends(get_current_user),
 ):
    """Upload a logo for a company"""
-    tenant_id = current_user.get("tenant_id")
-    if not tenant_id:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail="Tenant ID not found in user data",
-        )
+    tenant_id = _resolve_tenant_id_int(current_user)

    # Validar que la empresa existe
    company = CompanyService.get_by_id(db, company_id, tenant_id, 0)
@@ -303,35 +345,34 @@ async def upload_company_logo(
            detail=f"File too large. Maximum size: {MAX_FILE_SIZE / 1024 / 1024}MB",
        )

-    # Crear directorio si no existe
-    os.makedirs(UPLOAD_DIR, exist_ok=True)
-
-    # Eliminar logo anterior si existe
    if company.logo:
-        old_logo_path = company.logo
-        if os.path.exists(old_logo_path):
-            try:
-                os.remove(old_logo_path)
-            except Exception:
-                pass  # No es crítico si falla
+        _remove_stored_file(company.logo)

-    # Generar nombre único
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
-    filename = f"company_{company_id}_{timestamp}{file_ext}"
-    file_path = os.path.join(UPLOAD_DIR, filename)
+    filename = f"logo_{company_id}_{timestamp}{file_ext}"

-    # Guardar archivo
    try:
-        await file.seek(0)
-        with open(file_path, "wb") as buffer:
-            shutil.copyfileobj(file.file, buffer)
+        if settings.use_s3_object_storage:
+            ct = mimetypes.guess_type(filename)[0] or "image/jpeg"
+            key = company_logo_key(tenant_id, company_id, filename)
+            put_object_bytes(key, content, content_type=ct)
+            logger.info(
+                "Company logo stored in S3 key=%s bytes=%s",
+                key,
+                len(content),
+            )
+            file_path = key
+        else:
+            os.makedirs(UPLOAD_DIR, exist_ok=True)
+            file_path = os.path.join(UPLOAD_DIR, filename)
+            with open(file_path, "wb") as f:
+                f.write(content)
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Error saving file: {str(e)}",
        )

-    # Actualizar la empresa con la ruta del logo
    update_data = CompanyUpdateDTO(logo=file_path)
    service = CompanyService(db)
    updated_company = service.update(db, company_id, tenant_id, 0, update_data)
@@ -359,12 +400,7 @@ async def upload_company_certificate(
    Upload a certificate for a company
    certificate_type: fiel_cer, fiel_key, cfdi_cert_cer, cfdi_cert_key, cancel_cer, cancel_key
    """
-    tenant_id = current_user.get("tenant_id")
-    if not tenant_id:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail="Tenant ID not found in user data",
-        )
+    tenant_id = _resolve_tenant_id_int(current_user)

    # Validar que la empresa existe
    service = CompanyService(db)
@@ -410,27 +446,38 @@ async def upload_company_certificate(
            detail=f"File too large. Maximum size: {MAX_FILE_SIZE / 1024 / 1024}MB",
        )

-    # Crear directorio si no existe
-    certs_dir = os.path.join(UPLOAD_DIR, str(company_id), "certificates")
-    os.makedirs(certs_dir, exist_ok=True)
-
-    # Generar nombre único
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
-    filename = f"{certificate_type}_{timestamp}{file_ext}"
-    file_path = os.path.join(certs_dir, filename)

-    # Guardar archivo
    try:
-        await file.seek(0)
-        with open(file_path, "wb") as buffer:
-            shutil.copyfileobj(file.file, buffer)
+        if settings.use_s3_object_storage:
+            key = company_certificate_key(
+                tenant_id, company_id, certificate_type, timestamp, file_ext
+            )
+            ct = (
+                "application/x-x509-ca-cert"
+                if file_ext == ".cer"
+                else "application/pkcs8"
+            )
+            put_object_bytes(key, content, content_type=ct)
+            logger.info(
+                "Company certificate stored in S3 key=%s bytes=%s",
+                key,
+                len(content),
+            )
+            file_path = key
+        else:
+            certs_dir = os.path.join(UPLOAD_DIR, str(company_id), "certificates")
+            os.makedirs(certs_dir, exist_ok=True)
+            filename = f"{certificate_type}_{timestamp}{file_ext}"
+            file_path = os.path.join(certs_dir, filename)
+            with open(file_path, "wb") as f:
+                f.write(content)
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Error saving file: {str(e)}",
        )

-    # Actualizar la base de datos
    service.upload_certificate(company_id, certificate_type, file_path, tenant_id)

    return {
--- a/backend/api/v1/modules/a76/general_catalogs/company/service.py
+++ b/backend/api/v1/modules/a76/general_catalogs/company/service.py
@@ -10,6 +10,10 @@ from fastapi import HTTPException
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session

+from core.config import settings
+from core.s3_keys import tenant_company_prefix
+from core.storage_s3 import delete_objects_with_prefix
+
 from .dto import CompanyCreateDTO, CompanyResponseDTO, CompanyUpdateDTO
 from .models import Company
 from ...audit_log.services.service import AuditService
@@ -594,9 +598,24 @@ class CompanyService:
        # ----------------------

        try:
+            # 1) Borrado lógico en base de datos
            company.deleted_at = datetime.utcnow()
-            
            db.flush()
+
+            # 2) Limpieza de objetos S3/MinIO asociados a la compañía
+            if settings.use_s3_object_storage:
+                try:
+                    prefix = tenant_company_prefix(tenant_id, company_id)
+                    delete_objects_with_prefix(prefix)
+                except Exception as e:
+                    # No bloquear la eliminación lógica si falla la limpieza de objetos
+                    logger.error(
+                        "Error deleting S3 objects for company %s (tenant %s): %s",
+                        company_id,
+                        tenant_id,
+                        e,
+                    )
+
            db.commit()

            # --- Audit Log ---
@@ -678,6 +697,11 @@ class CompanyService:
        
        try:
            if target_cert:
+                old_path = getattr(target_cert, field_to_update, None)
+                if old_path and str(old_path).startswith("tenants/"):
+                    from core.storage_s3 import delete_object_if_exists
+
+                    delete_object_if_exists(str(old_path))
                # Si existe, actualizamos
                setattr(target_cert, field_to_update, file_path)
            else:
--- a/backend/api/v1/modules/a76/layouts_csv/boms/routes.py
+++ b/backend/api/v1/modules/a76/layouts_csv/boms/routes.py
@@ -2,7 +2,6 @@
 Rutas de importación CSV para BOMs.
 Flujo: upload → scan → status (polling) → commit.
 """
-import base64
 import json
 import logging
 import os
@@ -14,7 +13,6 @@ from typing import 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 api.v1.modules.core.tasks_tracking import track_and_dispatch

@@ -22,10 +20,11 @@ from .schemas import ImportJobResponse
 from .tasks import (
    scan_file,
    insert_valid_rows,
-    BOM_IMPORT_FILE_PREFIX,
+    JOB_TYPE,
    BOM_IMPORT_META_PREFIX,
    BOM_IMPORT_REDIS_TTL,
 )
+from ..common import storage as common_storage
 from ..common.error_csv import download_scan_errors_csv_stream
 from ..common.track_commit_dispatch import dispatch_tracked_layouts_csv_commit
 from ..common.responses import normalize_commit_status_payload
@@ -67,31 +66,23 @@ async def upload_import_file(
    }

    try:
-        r = _get_redis()
-        r.set(
-            f"{BOM_IMPORT_FILE_PREFIX}{job_id}",
-            base64.b64encode(contents),
-            ex=BOM_IMPORT_REDIS_TTL,
-        )
-        r.set(
-            f"{BOM_IMPORT_META_PREFIX}{job_id}",
-            json.dumps(meta_data).encode("utf-8"),
-            ex=BOM_IMPORT_REDIS_TTL,
+        common_storage.store_import_file(
+            JOB_TYPE,
+            job_id,
+            contents,
+            meta_data,
+            tenant_id=int(tenant_id),
+            company_id=company_id,
+            ttl=BOM_IMPORT_REDIS_TTL,
+            log_label="BOMs import",
        )
+    except common_storage.ImportStoreError as e:
+        logger.error(f"BOMs import: store error: {e}")
+        raise HTTPException(status_code=500, detail="No se pudo encolar el archivo.")
    except Exception as e:
        logger.error(f"BOMs import: Redis store error: {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)
-        with open(os.path.join(upload_dir, f"bom_{job_id}.csv"), "wb") as f:
-            f.write(contents)
-        with open(os.path.join(upload_dir, f"bom_{job_id}.meta.json"), "w") as f:
-            json.dump(meta_data, f)
-    except Exception as e:
-        logger.warning(f"BOMs import: local file save failed: {e}")
-
    track_and_dispatch(
        db=db,
        task=scan_file,
--- a/backend/api/v1/modules/a76/layouts_csv/cambio_regimen_regularizacion/routes.py
+++ b/backend/api/v1/modules/a76/layouts_csv/cambio_regimen_regularizacion/routes.py
@@ -2,7 +2,6 @@
 Rutas de importación CSV para Cambio de régimen y Regularizació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
@@ -14,7 +13,6 @@ 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 api.v1.modules.core.tasks_tracking import track_and_dispatch

@@ -65,7 +63,6 @@ async def upload_import_file(
    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,
@@ -76,25 +73,23 @@ async def upload_import_file(
    }

    try:
-        r = _get_redis()
-        r.set(file_key, base64.b64encode(contents), ex=CRREG_IMPORT_REDIS_TTL)
-        r.set(meta_key, json.dumps(meta_data).encode("utf-8"), ex=CRREG_IMPORT_REDIS_TTL)
+        common_storage.store_import_file(
+            JOB_TYPE,
+            job_id,
+            contents,
+            meta_data,
+            tenant_id=int(tenant_id),
+            company_id=company_id,
+            ttl=CRREG_IMPORT_REDIS_TTL,
+            log_label="Cambio régimen/Regularización import",
+        )
+    except common_storage.ImportStoreError as e:
+        logger.error("Cambio régimen/Regularización import: store error: %s", e)
+        raise HTTPException(status_code=500, detail="No se pudo encolar el archivo.")
    except Exception as e:
        logger.error("Cambio régimen/Regularizació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("Cambio régimen/Regularización import: local file save failed: %s", e)
-
    track_and_dispatch(
        db=db,
        task=scan_file,
--- a/backend/api/v1/modules/a76/layouts_csv/classes/routes.py
+++ b/backend/api/v1/modules/a76/layouts_csv/classes/routes.py
@@ -2,7 +2,6 @@
 Rutas de importación CSV para Clases de Materiales.
 Flujo: upload → scan → status (polling) → commit.
 """
-import base64
 import json
 import logging
 import os
@@ -14,7 +13,6 @@ from typing import 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 api.v1.modules.core.tasks_tracking import track_and_dispatch

@@ -24,10 +22,11 @@ from .schemas import ImportJobResponse
 from .tasks import (
    scan_file,
    insert_valid_rows,
-    CLS_IMPORT_FILE_PREFIX,
+    JOB_TYPE,
    CLS_IMPORT_META_PREFIX,
    CLS_IMPORT_REDIS_TTL,
 )
+from ..common import storage as common_storage
 from ..common.error_csv import download_scan_errors_csv_stream

 router = APIRouter()
@@ -71,31 +70,23 @@ async def upload_import_file(
    }

    try:
-        r = _get_redis()
-        r.set(
-            f"{CLS_IMPORT_FILE_PREFIX}{job_id}",
-            base64.b64encode(contents),
-            ex=CLS_IMPORT_REDIS_TTL,
-        )
-        r.set(
-            f"{CLS_IMPORT_META_PREFIX}{job_id}",
-            json.dumps(meta_data).encode("utf-8"),
-            ex=CLS_IMPORT_REDIS_TTL,
+        common_storage.store_import_file(
+            JOB_TYPE,
+            job_id,
+            contents,
+            meta_data,
+            tenant_id=int(tenant_id),
+            company_id=company_id,
+            ttl=CLS_IMPORT_REDIS_TTL,
+            log_label="Classes import",
        )
+    except common_storage.ImportStoreError as e:
+        logger.error(f"Classes import: store error: {e}")
+        raise HTTPException(status_code=500, detail="No se pudo encolar el archivo.")
    except Exception as e:
        logger.error(f"Classes import: Redis store error: {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)
-        with open(os.path.join(upload_dir, f"cls_{job_id}.csv"), "wb") as f:
-            f.write(contents)
-        with open(os.path.join(upload_dir, f"cls_{job_id}.meta.json"), "w") as f:
-            json.dump(meta_data, f)
-    except Exception as e:
-        logger.warning(f"Classes import: local file save failed: {e}")
-
    track_and_dispatch(
        db=db,
        task=scan_file,
--- a/backend/api/v1/modules/a76/layouts_csv/clients_and_providers/routes.py
+++ b/backend/api/v1/modules/a76/layouts_csv/clients_and_providers/routes.py
@@ -2,7 +2,6 @@
 Rutas de importación CSV para Clientes y Proveedores.
 Mismo flujo que a76.imports: upload → scan → status (polling) → commit.
 """
-import base64
 import json
 import logging
 import os
@@ -14,7 +13,6 @@ from typing import 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 api.v1.modules.core.tasks_tracking import track_and_dispatch

@@ -24,10 +22,11 @@ from .schemas import ImportJobResponse
 from .tasks import (
    scan_file,
    insert_valid_rows,
-    CP_IMPORT_FILE_PREFIX,
+    JOB_TYPE,
    CP_IMPORT_META_PREFIX,
    CP_IMPORT_REDIS_TTL,
 )
+from ..common import storage as common_storage
 from ..common.error_csv import download_scan_errors_csv_stream

 router = APIRouter()
@@ -70,31 +69,23 @@ async def upload_import_file(
    }

    try:
-        r = _get_redis()
-        r.set(
-            f"{CP_IMPORT_FILE_PREFIX}{job_id}",
-            base64.b64encode(contents),
-            ex=CP_IMPORT_REDIS_TTL,
-        )
-        r.set(
-            f"{CP_IMPORT_META_PREFIX}{job_id}",
-            json.dumps(meta_data).encode("utf-8"),
-            ex=CP_IMPORT_REDIS_TTL,
+        common_storage.store_import_file(
+            JOB_TYPE,
+            job_id,
+            contents,
+            meta_data,
+            tenant_id=int(tenant_id),
+            company_id=company_id,
+            ttl=CP_IMPORT_REDIS_TTL,
+            log_label="CP import",
        )
+    except common_storage.ImportStoreError as e:
+        logger.error(f"CP import: store error: {e}")
+        raise HTTPException(status_code=500, detail="No se pudo encolar el archivo.")
    except Exception as e:
        logger.error(f"CP import: Redis store error: {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)
-        with open(os.path.join(upload_dir, f"cp_{job_id}.csv"), "wb") as f:
-            f.write(contents)
-        with open(os.path.join(upload_dir, f"cp_{job_id}.meta.json"), "w") as f:
-            json.dump(meta_data, f)
-    except Exception as e:
-        logger.warning(f"CP import: local file save failed: {e}")
-
    track_and_dispatch(
        db=db,
        task=scan_file,
--- a/backend/api/v1/modules/a76/layouts_csv/common/storage.py
+++ b/backend/api/v1/modules/a76/layouts_csv/common/storage.py
@@ -15,6 +15,12 @@ logger = logging.getLogger(__name__)
 IMPORT_REDIS_TTL = 3600


+class ImportStoreError(Exception):
+    """Fallo al guardar CSV en Redis/MinIO."""
+
+    pass
+
+
 def _get_redis():
    import redis
    url = os.getenv("VALKEY_URL", os.getenv("REDIS_URL", "redis://valkey:6379/0"))
@@ -60,23 +66,103 @@ def error_path_for_job(job_type: str, job_id: str) -> str:
    return os.path.join(error_dir(), f"{job_type}_{job_id}.jsonl")


+def store_import_file(
+    job_type: str,
+    job_id: str,
+    raw_bytes: bytes,
+    meta_dict: dict,
+    tenant_id,
+    company_id: int,
+    ttl: int = IMPORT_REDIS_TTL,
+    log_label: str = "",
+) -> None:
+    """
+    Guarda CSV y meta en Redis. Si CSV_IMPORT_STORAGE=minio, sube el CSV a S3 y en Redis
+    guarda JSON {"v":2,"s3_key":...}; si no, base64 en Redis (comportamiento anterior).
+    En modo redis opcionalmente escribe copia local bajo layouts/imports/temp (debug).
+    """
+    from core.config import settings
+
+    file_key, meta_key, _ = storage_keys(job_type, job_id)
+    r = _get_redis()
+    meta_bytes = json.dumps(meta_dict).encode("utf-8")
+
+    if settings.CSV_IMPORT_STORAGE == "minio":
+        from core.storage_s3 import put_csv_object, s3_key_for_csv_import
+
+        key = s3_key_for_csv_import(tenant_id, company_id, job_type, job_id)
+        try:
+            put_csv_object(key, raw_bytes)
+        except Exception as e:
+            logger.exception("%s MinIO put failed: %s", log_label or job_type, e)
+            raise ImportStoreError(str(e)) from e
+        payload = json.dumps({"v": 2, "s3_key": key}).encode("utf-8")
+        r.set(file_key, payload, ex=ttl)
+    else:
+        r.set(file_key, base64.b64encode(raw_bytes), ex=ttl)
+
+    r.set(meta_key, meta_bytes, ex=ttl)
+
+    if settings.CSV_IMPORT_STORAGE != "minio":
+        try:
+            path = file_path_for_job(job_type, job_id)
+            os.makedirs(upload_dir(), exist_ok=True)
+            with open(path, "wb") as f:
+                f.write(raw_bytes)
+            meta_path = path.replace(".csv", ".meta.json")
+            with open(meta_path, "w", encoding="utf-8") as f:
+                json.dump(meta_dict, f)
+        except Exception as e:
+            logger.warning("%s local file save failed: %s", log_label or job_type, e)
+
+
 def ensure_file_from_redis(job_type: str, job_id: str, log_prefix: str = "") -> Optional[str]:
    """
-    Descarga contenido del CSV desde Redis y lo escribe en disco.
+    Obtiene el CSV desde Redis (referencia MinIO o base64 legacy) y lo escribe en disco.
    Devuelve la ruta del archivo o None si no hay datos o falla.
    """
+    from core.config import settings
+
    file_key, _, _ = storage_keys(job_type, job_id)
    r = _get_redis()
    data = r.get(file_key)
    if not data:
        return None
+
+    path = file_path_for_job(job_type, job_id)
+    os.makedirs(upload_dir(), exist_ok=True)
+
+    if data.startswith(b"{"):
+        try:
+            obj = json.loads(data.decode("utf-8"))
+            if isinstance(obj, dict) and obj.get("v") == 2 and obj.get("s3_key"):
+                if settings.CSV_IMPORT_STORAGE != "minio":
+                    logger.warning(
+                        "%s Redis has MinIO ref but CSV_IMPORT_STORAGE=%s",
+                        log_prefix or job_type,
+                        settings.CSV_IMPORT_STORAGE,
+                    )
+                from core.storage_s3 import get_object_bytes
+
+                try:
+                    raw = get_object_bytes(obj["s3_key"])
+                except Exception as e:
+                    logger.warning("%s MinIO get failed: %s", log_prefix or job_type, e)
+                    return None
+                with open(path, "wb") as f:
+                    f.write(raw)
+                return path
+        except json.JSONDecodeError:
+            pass
+        except Exception as e:
+            logger.warning("%s failed to parse MinIO ref from Redis: %s", log_prefix or job_type, e)
+            return None
+
    try:
        raw = base64.b64decode(data)
    except Exception as e:
        logger.warning("%s failed to decode file from Redis: %s", log_prefix or job_type, e)
        return None
-    path = file_path_for_job(job_type, job_id)
-    os.makedirs(upload_dir(), exist_ok=True)
    with open(path, "wb") as f:
        f.write(raw)
    return path
@@ -152,7 +238,31 @@ def cleanup_import_job(
    error_path: Optional[str] = None,
    meta_path: Optional[str] = None,
 ) -> None:
-    """Elimina archivos locales y claves Redis del job."""
+    """Elimina archivos locales, objeto MinIO si aplica, y claves Redis del job."""
+    from core.config import settings
+
+    if settings.CSV_IMPORT_STORAGE == "minio":
+        from core.storage_s3 import delete_object_if_exists
+        from core.s3_keys import legacy_csv_import_key
+
+        file_key, _, _ = storage_keys(job_type, job_id)
+        try:
+            r = _get_redis()
+            data = r.get(file_key)
+            deleted = False
+            if data and data.startswith(b"{"):
+                try:
+                    obj = json.loads(data.decode("utf-8"))
+                    if isinstance(obj, dict) and obj.get("v") == 2 and obj.get("s3_key"):
+                        delete_object_if_exists(obj["s3_key"])
+                        deleted = True
+                except Exception as e:
+                    logger.warning("Cleanup: parse Redis file ref: %s", e)
+            if not deleted:
+                delete_object_if_exists(legacy_csv_import_key(job_type, job_id))
+        except Exception as e:
+            logger.warning("Cleanup: MinIO delete failed: %s", e)
+
    if file_path and os.path.exists(file_path):
        try:
            os.remove(file_path)
--- a/backend/api/v1/modules/a76/layouts_csv/customs_brokers/routes.py
+++ b/backend/api/v1/modules/a76/layouts_csv/customs_brokers/routes.py
@@ -2,7 +2,6 @@
 Rutas de importación CSV para Agentes Aduanales.
 Mismo flujo que a76.imports: upload → scan → status (polling) → commit.
 """
-import base64
 import json
 import logging
 import os
@@ -14,7 +13,6 @@ from typing import 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 api.v1.modules.core.tasks_tracking import track_and_dispatch

@@ -24,10 +22,11 @@ from .schemas import ImportJobResponse
 from .tasks import (
    scan_file,
    insert_valid_rows,
-    CB_IMPORT_FILE_PREFIX,
+    JOB_TYPE,
    CB_IMPORT_META_PREFIX,
    CB_IMPORT_REDIS_TTL,
 )
+from ..common import storage as common_storage
 from ..common.error_csv import download_scan_errors_csv_stream

 router = APIRouter()
@@ -70,31 +69,23 @@ async def upload_import_file(
    }

    try:
-        r = _get_redis()
-        r.set(
-            f"{CB_IMPORT_FILE_PREFIX}{job_id}",
-            base64.b64encode(contents),
-            ex=CB_IMPORT_REDIS_TTL,
-        )
-        r.set(
-            f"{CB_IMPORT_META_PREFIX}{job_id}",
-            json.dumps(meta_data).encode("utf-8"),
-            ex=CB_IMPORT_REDIS_TTL,
+        common_storage.store_import_file(
+            JOB_TYPE,
+            job_id,
+            contents,
+            meta_data,
+            tenant_id=int(tenant_id),
+            company_id=company_id,
+            ttl=CB_IMPORT_REDIS_TTL,
+            log_label="CB import",
        )
+    except common_storage.ImportStoreError as e:
+        logger.error(f"CB import: store error: {e}")
+        raise HTTPException(status_code=500, detail="No se pudo encolar el archivo.")
    except Exception as e:
        logger.error(f"CB import: Redis store error: {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)
-        with open(os.path.join(upload_dir, f"cb_{job_id}.csv"), "wb") as f:
-            f.write(contents)
-        with open(os.path.join(upload_dir, f"cb_{job_id}.meta.json"), "w") as f:
-            json.dump(meta_data, f)
-    except Exception as e:
-        logger.warning(f"CB import: local file save failed: {e}")
-
    track_and_dispatch(
        db=db,
        task=scan_file,
--- a/backend/api/v1/modules/a76/layouts_csv/drivers/routes.py
+++ b/backend/api/v1/modules/a76/layouts_csv/drivers/routes.py
@@ -2,7 +2,6 @@
 Rutas de importacion CSV para Conductores.
 Flujo: upload -> scan -> status (polling) -> commit.
 """
-import base64
 import json
 import logging
 import os
@@ -15,18 +14,17 @@ from typing import 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
 from .tasks import (
    run_scan_sync,
    run_commit_sync,
-    DRV_IMPORT_FILE_PREFIX,
-    DRV_IMPORT_META_PREFIX,
+    JOB_TYPE,
    DRV_IMPORT_STATUS_PREFIX,
    DRV_IMPORT_REDIS_TTL,
 )
+from ..common import storage as common_storage
 from ..common.error_csv import download_scan_errors_csv_stream
 from ..common.responses import normalize_commit_status_payload

@@ -67,31 +65,23 @@ async def upload_import_file(
    }

    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,
+        common_storage.store_import_file(
+            JOB_TYPE,
+            job_id,
+            contents,
+            meta_data,
+            tenant_id=int(tenant_id),
+            company_id=company_id,
+            ttl=DRV_IMPORT_REDIS_TTL,
+            log_label="Drivers import",
        )
+    except common_storage.ImportStoreError as e:
+        logger.error(f"Drivers import: store error: {e}")
+        raise HTTPException(status_code=500, detail="No se pudo encolar el archivo.")
    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 = layout_path("imports", "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}")
-
    def run_scan_background():
        try:
            run_scan_sync(job_id)
--- a/backend/api/v1/modules/a76/layouts_csv/exchange_rate/routes.py
+++ b/backend/api/v1/modules/a76/layouts_csv/exchange_rate/routes.py
@@ -2,7 +2,6 @@
 Rutas de importación CSV para Tipos de Cambio.
 Mismo flujo que customs_brokers/imports: upload → scan → status (polling) → commit.
 """
-import base64
 import json
 import logging
 import os
@@ -14,7 +13,6 @@ from typing import Dict, Any, Optional

 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 api.v1.modules.core.tasks_tracking import track_and_dispatch

@@ -24,10 +22,11 @@ from .schemas import ImportJobResponse
 from .tasks import (
    scan_file,
    insert_valid_rows,
-    ER_IMPORT_FILE_PREFIX,
+    JOB_TYPE,
    ER_IMPORT_META_PREFIX,
    ER_IMPORT_REDIS_TTL,
 )
+from ..common import storage as common_storage
 from ..common.error_csv import download_scan_errors_csv_stream

 router = APIRouter()
@@ -81,31 +80,23 @@ async def upload_import_file(
    }

    try:
-        r = _get_redis()
-        r.set(
-            f"{ER_IMPORT_FILE_PREFIX}{job_id}",
-            base64.b64encode(contents),
-            ex=ER_IMPORT_REDIS_TTL,
-        )
-        r.set(
-            f"{ER_IMPORT_META_PREFIX}{job_id}",
-            json.dumps(meta_data).encode("utf-8"),
-            ex=ER_IMPORT_REDIS_TTL,
+        common_storage.store_import_file(
+            JOB_TYPE,
+            job_id,
+            contents,
+            meta_data,
+            tenant_id=int(tenant_id),
+            company_id=company_id,
+            ttl=ER_IMPORT_REDIS_TTL,
+            log_label="ER import",
        )
+    except common_storage.ImportStoreError as e:
+        logger.error(f"ER import: store error: {e}")
+        raise HTTPException(status_code=500, detail="No se pudo encolar el archivo.")
    except Exception as e:
        logger.error(f"ER import: Redis store error: {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)
-        with open(os.path.join(upload_dir, f"er_{job_id}.csv"), "wb") as f:
-            f.write(contents)
-        with open(os.path.join(upload_dir, f"er_{job_id}.meta.json"), "w") as f:
-            json.dump(meta_data, f)
-    except Exception as e:
-        logger.warning(f"ER import: local file save failed: {e}")
-
    track_and_dispatch(
        db=db,
        task=scan_file,
--- a/backend/api/v1/modules/a76/layouts_csv/exportacion/routes.py
+++ b/backend/api/v1/modules/a76/layouts_csv/exportacion/routes.py
@@ -2,7 +2,6 @@
 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
@@ -17,7 +16,6 @@ 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 api.v1.modules.core.tasks_tracking import track_and_dispatch

@@ -61,7 +59,6 @@ async def upload_import_file(
    job_id = str(uuid4())
    contents = await file.read()

-    file_key, meta_key, _ = common_storage.storage_keys(JOB_TYPE, job_id)
    default_template = (
        "exp_def_header" if model_target == "invoice_header"
        else "exp_def_series" if model_target == "invoice_series"
@@ -83,25 +80,23 @@ async def upload_import_file(
    }

    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)
+        common_storage.store_import_file(
+            JOB_TYPE,
+            job_id,
+            contents,
+            meta_data,
+            tenant_id=int(tenant_id),
+            company_id=company_id,
+            ttl=EXP_IMPORT_REDIS_TTL,
+            log_label="Exportación import",
+        )
+    except common_storage.ImportStoreError as e:
+        logger.error("Exportación import: store error: %s", e)
+        raise HTTPException(status_code=500, detail="No se pudo encolar el archivo.")
    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)
-
    track_and_dispatch(
        db=db,
        task=scan_file,
--- a/backend/api/v1/modules/a76/layouts_csv/facturas/routes.py
+++ b/backend/api/v1/modules/a76/layouts_csv/facturas/routes.py
@@ -1,6 +1,5 @@
 from datetime import datetime
 from uuid import uuid4
-import base64
 import os
 import json
 import logging
@@ -14,14 +13,13 @@ from typing import Optional, Literal, Dict, Any
 from core.celery_app import celery_app
 from core.config import settings
 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 api.v1.modules.core.tasks_tracking import track_and_dispatch

 from .tasks import (
    scan_file,
    insert_valid_rows,
-    IMPORT_FILE_KEY_PREFIX,
+    JOB_TYPE,
    IMPORT_META_KEY_PREFIX,
    IMPORT_REDIS_TTL,
 )
@@ -83,37 +81,25 @@ async def upload_import_file(
        "template_id": template_id,
    }

-    # Store file and meta in Redis so the Celery worker can read them (no shared filesystem needed)
    try:
-        redis_client = _get_redis()
-        redis_client.set(
-            f"{IMPORT_FILE_KEY_PREFIX}{job_id}",
-            base64.b64encode(contents),
-            ex=IMPORT_REDIS_TTL,
-        )
-        redis_client.set(
-            f"{IMPORT_META_KEY_PREFIX}{job_id}",
-            json.dumps(meta_data).encode("utf-8"),
-            ex=IMPORT_REDIS_TTL,
+        common_storage.store_import_file(
+            JOB_TYPE,
+            job_id,
+            contents,
+            meta_data,
+            tenant_id=int(tenant_id),
+            company_id=company_id,
+            ttl=IMPORT_REDIS_TTL,
+            log_label="Facturas import",
        )
+    except common_storage.ImportStoreError as e:
+        logger.error(f"Import store error: {e}")
+        raise HTTPException(status_code=500, detail="Failed to queue file for processing.")
    except Exception as e:
        logger.error(f"Redis store error: {e}")
        raise HTTPException(status_code=500, detail="Failed to queue file for processing.")

-    # Optional: also write to local disk (e.g. for same-machine worker or debugging)
-    try:
-        upload_dir = layout_path("imports", "temp")
-        os.makedirs(upload_dir, exist_ok=True)
-        file_path = os.path.join(upload_dir, f"{job_id}.csv")
-        meta_path = os.path.join(upload_dir, f"{job_id}.meta.json")
-        with open(file_path, "wb") as f:
-            f.write(contents)
-        with open(meta_path, "w") as f:
-            json.dump(meta_data, f)
-    except Exception as e:
-        logger.warning(f"Local file save failed (worker will use Redis): {e}")
-
-    # Trigger Celery Task (Async). Worker loads file from Redis.
+    # Trigger Celery Task (Async). Worker loads file from Redis / MinIO.
    track_and_dispatch(
        db=db,
        task=scan_file,
--- a/backend/api/v1/modules/a76/layouts_csv/parts/routes.py
+++ b/backend/api/v1/modules/a76/layouts_csv/parts/routes.py
@@ -2,7 +2,6 @@
 Rutas de importación CSV para Números de Parte.
 Flujo: upload → scan → status (polling) → commit.
 """
-import base64
 import json
 import logging
 import os
@@ -14,7 +13,6 @@ from typing import 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 api.v1.modules.core.tasks_tracking import track_and_dispatch

@@ -22,10 +20,11 @@ from .schemas import ImportJobResponse
 from .tasks import (
    scan_file,
    insert_valid_rows,
-    PART_IMPORT_FILE_PREFIX,
+    JOB_TYPE,
    PART_IMPORT_META_PREFIX,
    PART_IMPORT_REDIS_TTL,
 )
+from ..common import storage as common_storage
 from ..common.error_csv import download_scan_errors_csv_stream
 from ..common.track_commit_dispatch import dispatch_tracked_layouts_csv_commit
 from ..common.responses import normalize_commit_status_payload
@@ -71,31 +70,23 @@ async def upload_import_file(
    }

    try:
-        r = _get_redis()
-        r.set(
-            f"{PART_IMPORT_FILE_PREFIX}{job_id}",
-            base64.b64encode(contents),
-            ex=PART_IMPORT_REDIS_TTL,
-        )
-        r.set(
-            f"{PART_IMPORT_META_PREFIX}{job_id}",
-            json.dumps(meta_data).encode("utf-8"),
-            ex=PART_IMPORT_REDIS_TTL,
+        common_storage.store_import_file(
+            JOB_TYPE,
+            job_id,
+            contents,
+            meta_data,
+            tenant_id=int(tenant_id),
+            company_id=company_id,
+            ttl=PART_IMPORT_REDIS_TTL,
+            log_label="Parts import",
        )
+    except common_storage.ImportStoreError as e:
+        logger.error(f"Parts import: store error: {e}")
+        raise HTTPException(status_code=500, detail="No se pudo encolar el archivo.")
    except Exception as e:
        logger.error(f"Parts import: Redis store error: {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)
-        with open(os.path.join(upload_dir, f"part_{job_id}.csv"), "wb") as f:
-            f.write(contents)
-        with open(os.path.join(upload_dir, f"part_{job_id}.meta.json"), "w") as f:
-            json.dump(meta_data, f)
-    except Exception as e:
-        logger.warning(f"Parts import: local file save failed: {e}")
-
    track_and_dispatch(
        db=db,
        task=scan_file,
--- a/backend/api/v1/modules/a76/layouts_csv/pedmientos/routes.py
+++ b/backend/api/v1/modules/a76/layouts_csv/pedmientos/routes.py
@@ -2,7 +2,6 @@
 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
@@ -14,7 +13,6 @@ from typing import Dict, Any, Optional

 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 api.v1.modules.core.tasks_tracking import track_and_dispatch

@@ -64,7 +62,6 @@ async def upload_import_file(
    job_id = str(uuid4())
    contents = await file.read()

-    file_key, meta_key, _ = common_storage.storage_keys(PED_JOB_TYPE, job_id)
    meta_data: Dict[str, Any] = {
        "tenant_id": tenant_id,
        "company_id": company_id,
@@ -76,33 +73,23 @@ async def upload_import_file(
        meta_data["dateFormat"] = dateFormat

    try:
-        r = _get_redis()
-        r.set(
-            file_key,
-            base64.b64encode(contents),
-            ex=PED_IMPORT_REDIS_TTL,
-        )
-        r.set(
-            meta_key,
-            json.dumps(meta_data).encode("utf-8"),
-            ex=PED_IMPORT_REDIS_TTL,
+        common_storage.store_import_file(
+            PED_JOB_TYPE,
+            job_id,
+            contents,
+            meta_data,
+            tenant_id=int(tenant_id),
+            company_id=company_id,
+            ttl=PED_IMPORT_REDIS_TTL,
+            log_label="Pedimentos import",
        )
+    except common_storage.ImportStoreError as e:
+        logger.error(f"Pedimentos import: store error: {e}")
+        raise HTTPException(status_code=500, detail="No se pudo encolar el archivo.")
    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 = layout_path("imports", "temp")
-        os.makedirs(upload_dir, exist_ok=True)
-        csv_path = common_storage.file_path_for_job(PED_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(f"Pedimentos import: local file save failed: {e}")
-
    track_and_dispatch(
        db=db,
        task=scan_file,
--- a/backend/api/v1/modules/a76/layouts_csv/trailers/routes.py
+++ b/backend/api/v1/modules/a76/layouts_csv/trailers/routes.py
@@ -2,7 +2,6 @@
 Rutas de importación CSV para Trailers y Cajas.
 Flujo: upload -> scan -> status (polling) -> commit.
 """
-import base64
 import json
 import logging
 import os
@@ -15,7 +14,6 @@ from typing import 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 api.v1.modules.core.tasks_tracking import track_and_dispatch

@@ -24,11 +22,11 @@ from .tasks import (
    scan_file,
    run_scan_sync,
    run_commit_sync,
-    TRL_IMPORT_FILE_PREFIX,
-    TRL_IMPORT_META_PREFIX,
+    JOB_TYPE,
    TRL_IMPORT_STATUS_PREFIX,
    TRL_IMPORT_REDIS_TTL,
 )
+from ..common import storage as common_storage
 from ..common.error_csv import download_scan_errors_csv_stream
 from ..common.responses import normalize_commit_status_payload

@@ -71,31 +69,23 @@ async def upload_import_file(
    }

    try:
-        r = _get_redis()
-        r.set(
-            f"{TRL_IMPORT_FILE_PREFIX}{job_id}",
-            base64.b64encode(contents),
-            ex=TRL_IMPORT_REDIS_TTL,
-        )
-        r.set(
-            f"{TRL_IMPORT_META_PREFIX}{job_id}",
-            json.dumps(meta_data).encode("utf-8"),
-            ex=TRL_IMPORT_REDIS_TTL,
+        common_storage.store_import_file(
+            JOB_TYPE,
+            job_id,
+            contents,
+            meta_data,
+            tenant_id=int(tenant_id),
+            company_id=company_id,
+            ttl=TRL_IMPORT_REDIS_TTL,
+            log_label="Trailers import",
        )
+    except common_storage.ImportStoreError as e:
+        logger.error(f"Trailers import: store error: {e}")
+        raise HTTPException(status_code=500, detail="No se pudo encolar el archivo.")
    except Exception as e:
        logger.error(f"Trailers import: Redis store error: {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)
-        with open(os.path.join(upload_dir, f"trl_{job_id}.csv"), "wb") as f:
-            f.write(contents)
-        with open(os.path.join(upload_dir, f"trl_{job_id}.meta.json"), "w") as f:
-            json.dump(meta_data, f)
-    except Exception as e:
-        logger.warning(f"Trailers import: local file save failed: {e}")
-
    track_and_dispatch(
        db=db,
        task=scan_file,
--- a/backend/api/v1/modules/a76/layouts_csv/transportistas/routes.py
+++ b/backend/api/v1/modules/a76/layouts_csv/transportistas/routes.py
@@ -2,7 +2,6 @@
 Rutas de importación CSV para Transportistas.
 Flujo: upload → scan → status (polling) → commit.
 """
-import base64
 import json
 import logging
 import os
@@ -15,7 +14,6 @@ from typing import 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 api.v1.modules.core.tasks_tracking import track_and_dispatch

@@ -24,11 +22,11 @@ from .tasks import (
    scan_file,
    run_scan_sync,
    run_commit_sync,
-    TRP_IMPORT_FILE_PREFIX,
-    TRP_IMPORT_META_PREFIX,
+    JOB_TYPE,
    TRP_IMPORT_STATUS_PREFIX,
    TRP_IMPORT_REDIS_TTL,
 )
+from ..common import storage as common_storage
 from ..common.error_csv import download_scan_errors_csv_stream
 from ..common.responses import normalize_commit_status_payload

@@ -71,31 +69,23 @@ async def upload_import_file(
    }

    try:
-        r = _get_redis()
-        r.set(
-            f"{TRP_IMPORT_FILE_PREFIX}{job_id}",
-            base64.b64encode(contents),
-            ex=TRP_IMPORT_REDIS_TTL,
-        )
-        r.set(
-            f"{TRP_IMPORT_META_PREFIX}{job_id}",
-            json.dumps(meta_data).encode("utf-8"),
-            ex=TRP_IMPORT_REDIS_TTL,
+        common_storage.store_import_file(
+            JOB_TYPE,
+            job_id,
+            contents,
+            meta_data,
+            tenant_id=int(tenant_id),
+            company_id=company_id,
+            ttl=TRP_IMPORT_REDIS_TTL,
+            log_label="Transportistas import",
        )
+    except common_storage.ImportStoreError as e:
+        logger.error("Transportistas import: store error: %s", e)
+        raise HTTPException(status_code=500, detail="No se pudo encolar el archivo.")
    except Exception as e:
        logger.error("Transportistas 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)
-        with open(os.path.join(upload_dir, f"trp_{job_id}.csv"), "wb") as f:
-            f.write(contents)
-        with open(os.path.join(upload_dir, f"trp_{job_id}.meta.json"), "w") as f:
-            json.dump(meta_data, f)
-    except Exception as e:
-        logger.warning("Transportistas import: local file save failed: %s", e)
-
    track_and_dispatch(
        db=db,
        task=scan_file,
--- a/backend/api/v1/modules/a76/layouts_csv/us_tariff_fractions/routes.py
+++ b/backend/api/v1/modules/a76/layouts_csv/us_tariff_fractions/routes.py
@@ -2,7 +2,6 @@
 Rutas de importación CSV para Fracción Americana (US Tariff Fractions).
 Mismo flujo que exchange_rate/imports: upload → scan → status (polling) → commit.
 """
-import base64
 import json
 import logging
 import os
@@ -14,7 +13,6 @@ from typing import 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 api.v1.modules.core.tasks_tracking import track_and_dispatch

@@ -24,10 +22,11 @@ from .schemas import ImportJobResponse
 from .tasks import (
    scan_file,
    insert_valid_rows,
-    FA_IMPORT_FILE_PREFIX,
+    JOB_TYPE,
    FA_IMPORT_META_PREFIX,
    FA_IMPORT_REDIS_TTL,
 )
+from ..common import storage as common_storage
 from ..common.error_csv import download_scan_errors_csv_stream

 router = APIRouter()
@@ -73,31 +72,23 @@ async def upload_import_file(
    }

    try:
-        r = _get_redis()
-        r.set(
-            f"{FA_IMPORT_FILE_PREFIX}{job_id}",
-            base64.b64encode(contents),
-            ex=FA_IMPORT_REDIS_TTL,
-        )
-        r.set(
-            f"{FA_IMPORT_META_PREFIX}{job_id}",
-            json.dumps(meta_data).encode("utf-8"),
-            ex=FA_IMPORT_REDIS_TTL,
+        common_storage.store_import_file(
+            JOB_TYPE,
+            job_id,
+            contents,
+            meta_data,
+            tenant_id=int(tenant_id),
+            company_id=company_id,
+            ttl=FA_IMPORT_REDIS_TTL,
+            log_label="FA import",
        )
+    except common_storage.ImportStoreError as e:
+        logger.error(f"FA import: store error: {e}")
+        raise HTTPException(status_code=500, detail="No se pudo encolar el archivo.")
    except Exception as e:
        logger.error(f"FA import: Redis store error: {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)
-        with open(os.path.join(upload_dir, f"fa_{job_id}.csv"), "wb") as f:
-            f.write(contents)
-        with open(os.path.join(upload_dir, f"fa_{job_id}.meta.json"), "w") as f:
-            json.dump(meta_data, f)
-    except Exception as e:
-        logger.warning(f"FA import: local file save failed: {e}")
-
    track_and_dispatch(
        db=db,
        task=scan_file,
--- a/backend/api/v1/modules/a76/layouts_csv/vehicles/routes.py
+++ b/backend/api/v1/modules/a76/layouts_csv/vehicles/routes.py
@@ -2,7 +2,6 @@
 Rutas de importación CSV para Vehículos (Transportes).
 Flujo: upload → scan → status (polling) → commit.
 """
-import base64
 import json
 import logging
 import os
@@ -15,7 +14,6 @@ from typing import 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 api.v1.modules.core.tasks_tracking import track_and_dispatch

@@ -24,11 +22,11 @@ from .tasks import (
    scan_file,
    run_scan_sync,
    run_commit_sync,
-    VEHL_IMPORT_FILE_PREFIX,
-    VEHL_IMPORT_META_PREFIX,
+    JOB_TYPE,
    VEHL_IMPORT_STATUS_PREFIX,
    VEHL_IMPORT_REDIS_TTL,
 )
+from ..common import storage as common_storage
 from ..common.error_csv import download_scan_errors_csv_stream
 from ..common.responses import normalize_commit_status_payload

@@ -71,31 +69,23 @@ async def upload_import_file(
    }

    try:
-        r = _get_redis()
-        r.set(
-            f"{VEHL_IMPORT_FILE_PREFIX}{job_id}",
-            base64.b64encode(contents),
-            ex=VEHL_IMPORT_REDIS_TTL,
-        )
-        r.set(
-            f"{VEHL_IMPORT_META_PREFIX}{job_id}",
-            json.dumps(meta_data).encode("utf-8"),
-            ex=VEHL_IMPORT_REDIS_TTL,
+        common_storage.store_import_file(
+            JOB_TYPE,
+            job_id,
+            contents,
+            meta_data,
+            tenant_id=int(tenant_id),
+            company_id=company_id,
+            ttl=VEHL_IMPORT_REDIS_TTL,
+            log_label="Vehicles import",
        )
+    except common_storage.ImportStoreError as e:
+        logger.error(f"Vehicles import: store error: {e}")
+        raise HTTPException(status_code=500, detail="No se pudo encolar el archivo.")
    except Exception as e:
        logger.error(f"Vehicles import: Redis store error: {e}")
        raise HTTPException(status_code=500, detail="No se pudo encolar el archivo.")

-    try:
-        upload_dir = layout_path("imports", "temp")
-        os.makedirs(upload_dir, exist_ok=True)
-        with open(os.path.join(upload_dir, f"veh_{job_id}.csv"), "wb") as f:
-            f.write(contents)
-        with open(os.path.join(upload_dir, f"veh_{job_id}.meta.json"), "w") as f:
-            json.dump(meta_data, f)
-    except Exception as e:
-        logger.warning(f"Vehicles import: local file save failed: {e}")
-
    track_and_dispatch(
        db=db,
        task=scan_file,
--- a/backend/api/v1/modules/a76/router.py
+++ b/backend/api/v1/modules/a76/router.py
@@ -34,6 +34,7 @@ from .transportation.trailers.routes import router as trailers_router
 from .transportation.transporters.routes import router as transporters_router
 from .transportation.vehicles.routes import router as vehicles_router
 from api.v1.modules.public.reference_data.material_types.routes import router as material_types_router
+from .factura_cove.routes import router as factura_cove_router

 # --- NUEVO IMPORT PARA REPORTES DE FACTURAS ---
 from .reports.importacion.facturas.routes import router as invoices_reports_router
@@ -82,6 +83,7 @@ router.include_router(doc_types_dig_router, prefix="/a76", tags=["a76 / document
 router.include_router(drivers_router, prefix="/a76", tags=["a76 / drivers"])
 router.include_router(transporters_router, prefix="/a76", tags=["a76 / transporters"])
 router.include_router(vehicles_router, prefix="/a76/transportation", tags=["a76 / vehicles"])
+router.include_router(factura_cove_router, prefix="/a76/factura-cove", tags=["a76 / factura_cove"])

 # Registrar router de tipos de material públicos
 router.include_router(