Merge branch 'development' of https://git.aduanasoft.com/ADUANASOFT/anexo76 into feature/tareas_TODO_configuracion

backend/api/v1/modules/a76/factura_cove/external_service.py

@@ -0,0 +1,192 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Dict
+import logging
+
+import httpx
+
+from core.config import settings
+from .schemas import FacturaCoveRequest
+
+
+logger = logging.getLogger(__name__)
+
+
+@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")
+        configuracion_vu = json_payload.get("configuracion_vu") or {}
+        logger.info(
+            "Sending COVE payload: invoice=%s rfc_vu=%s clave_fiel_len=%s clave_ws_len=%s cer_len=%s key_len=%s",
+            json_payload.get("numero_factura"),
+            configuracion_vu.get("rfc_usuario_vu"),
+            len(configuracion_vu.get("clave_fiel") or ""),
+            len(configuracion_vu.get("clave_webservice") or ""),
+            len(configuracion_vu.get("archivo_cer_base64") or ""),
+            len(configuracion_vu.get("archivo_key_base64") or ""),
+        )
+
+        # 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
+

backend/api/v1/modules/a76/factura_cove/routes.py

@@ -0,0 +1,150 @@
+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("email")
+            or current_user.get("preferred_username")
+            or current_user.get("username")
+            or "system"
+        ),
+        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, body.recipient_email],
+    )
+
+    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
+

backend/api/v1/modules/a76/factura_cove/schemas.py

@@ -0,0 +1,136 @@
+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
+    recipient_email: Optional[EmailStr] = None
+
+
+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)
+

backend/api/v1/modules/a76/factura_cove/service.py

@@ -0,0 +1,785 @@
+from __future__ import annotations
+
+import base64
+from dataclasses import dataclass
+from decimal import Decimal
+from typing import List, Tuple
+
+from cryptography.hazmat.primitives import padding
+from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
+from sqlalchemy.orm import Session
+
+from core.config import settings
+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 _get_company(self, ctx: InvoiceContext) -> Company | None:
+        company_id = getattr(ctx.invoice, "company_id", None)
+        if not company_id:
+            return None
+        return self.db.get(Company, company_id)
+
+    def _get_company_fiel_certificate(self, company: Company | None):
+        if not company:
+            return None
+
+        for certificate in company.digital_certificates or []:
+            if (certificate.certificate_type or "").strip().lower() == "fiel":
+                return certificate
+
+        return None
+
+    def _encrypt_fiel(self, raw_fiel: str) -> str:
+        """
+        Cifra la clave FIEL con el mismo esquema del sistema legado PHP:
+        AES-256-CBC + PKCS7 + base64.
+        """
+        normalized_fiel = (raw_fiel or "").strip()
+        if not normalized_fiel:
+            return ""
+
+        encryption_key = (settings.COVE_FIEL_HASH_KEY or "").encode("utf-8")
+        encryption_iv = (settings.COVE_FIEL_HASH_IV or "").encode("utf-8")
+        if not encryption_key or not encryption_iv:
+            return ""
+
+        key_bytes = encryption_key[:32].ljust(32, b"\0")
+        iv_bytes = encryption_iv[:16].ljust(16, b"\0")
+
+        padder = padding.PKCS7(algorithms.AES.block_size).padder()
+        padded_data = padder.update(normalized_fiel.encode("utf-8")) + padder.finalize()
+
+        cipher = Cipher(algorithms.AES(key_bytes), modes.CBC(iv_bytes))
+        encryptor = cipher.encryptor()
+        encrypted = encryptor.update(padded_data) + encryptor.finalize()
+        return base64.b64encode(encrypted).decode("ascii")
+
+    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
+        company = self._get_company(ctx)
+        company_vu = company.ventanilla_unica if company else None
+        company_fiel_certificate = self._get_company_fiel_certificate(company)
+
+        if not vu and not company_vu and not company_fiel_certificate:
+            errors.add_error(
+                field="vu",
+                message="La factura no tiene configuración VU asociada ni configuración VU/certificado FIEL en la empresa",
+                solution=[
+                    "Configura los datos VU del agente aduanal o de la empresa y sube certificado (.cer) y llave (.key) antes de generar COVE."
+                ],
+                code="MISSING_VU_CONFIGURATION",
+            )
+            return None
+
+        # Determinar usuario efectivo de WebService:
+        # - Preferimos el usuario configurado en VU (web_service_user)
+        # - Si no existe, usamos el de DODA-PITA (doda_web_service_user)
+        # - Si no existe, usamos la configuración VU de la empresa
+        effective_ws_user = (
+            (
+                vu.web_service_user
+                or vu.doda_web_service_user
+                or getattr(company_vu, "webservice_user", None)
+                or ""
+            ).strip()
+            if (vu or company_vu)
+            else ""
+        )
+
+        # Determinar clave FIEL efectiva desde la configuración persistida.
+        # Se envía cifrada con el mismo esquema AES-256-CBC del sistema legado.
+        clave_fiel_value = ""
+
+        if vu and getattr(vu, "fiel_access_key", None):
+            clave_fiel_value = self._encrypt_fiel(vu.fiel_access_key or "")
+        elif company_fiel_certificate:
+            # Fallback en base de datos: certificado FIEL de la empresa
+            company_fiel_secret = (
+                getattr(company_fiel_certificate, "access_key", None)
+                or getattr(company_fiel_certificate, "password", None)
+                or ""
+            )
+            clave_fiel_value = self._encrypt_fiel(str(company_fiel_secret))
+
+        # Validación básica de credenciales VU: para COVE necesitamos al menos
+        # un usuario de web service (VU o DODA) y una clave FIEL no vacía.
+        if not effective_ws_user:
+            errors.add_error(
+                field="vu",
+                message="Faltan credenciales de web service o clave FIEL en VU",
+                solution=[
+                    "Captura usuario y clave de web service en la pestaña VU o DODA del agente, "
+                    "o completa la configuración VU de la empresa y su certificado FIEL."
+                ],
+                code="MISSING_VU_CREDENTIALS",
+            )
+
+        if not clave_fiel_value:
+            errors.add_error(
+                field="vu.clave_fiel",
+                message="La clave FIEL para COVE no está configurada ni en VU ni en la empresa.",
+                solution=[
+                    "Captura la clave FIEL en la configuración VU del agente aduanal o en el certificado FIEL de la empresa."
+                ],
+                code="MISSING_FIEL_PASSWORD",
+            )
+
+        certificate_path = (
+            (getattr(vu, "certificate_path", None) or "").strip() if vu else ""
+        ) or (
+            (getattr(company_fiel_certificate, "cer_file_path", None) or "").strip()
+            if company_fiel_certificate
+            else ""
+        )
+        key_path = (
+            (getattr(vu, "key_path", None) or "").strip() if vu else ""
+        ) or (
+            (getattr(company_fiel_certificate, "key_file_path", None) or "").strip()
+            if company_fiel_certificate
+            else ""
+        )
+
+        if not (certificate_path and 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) en la configuración VU del agente aduanal o en los certificados digitales de la empresa."
+                ],
+                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(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 en la configuración VU del agente aduanal o en los certificados digitales de la empresa."],
+                    code="VU_CERT_NOT_FOUND",
+                )
+            else:
+                cer_bytes = get_object_bytes(certificate_path)
+                cer_b64 = base64.b64encode(cer_bytes).decode("ascii")
+
+            if not object_exists(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 en la configuración VU del agente aduanal o en los certificados digitales de la empresa."],
+                    code="VU_KEY_NOT_FOUND",
+                )
+            else:
+                key_bytes = get_object_bytes(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 VU o en los certificados digitales de la empresa."],
+                code="VU_STORAGE_ERROR",
+            )
+
+        if errors.has_errors():
+            return None
+
+        rfc_usuario_vu = (
+            (getattr(vu, "query_tax_id", None) or "").strip() if vu else ""
+        ) or (
+            (getattr(company_vu, "query_rfc", None) or "").strip() if company_vu else ""
+        )
+
+        # Clave/token del webservice: usar el valor de VU si existe, o una
+        # clave fija de pruebas mientras se termina la configuración real.
+        hardcoded_ws_key = (
+            "RZGd+CB4R6PfSrstOyN8Is9FXL9AK9NPFisyGGaEWa0vVHoVOl8v2SBcHBoGbt3T/4uHTGcsFQO3b7EonWVfugQjBooywbz74K+jM68j8/Y="
+        )
+
+        return ConfiguracionVU(
+            rfc_usuario_vu=rfc_usuario_vu,
+            clave_webservice=(
+                (getattr(vu, "web_service_access_key", None) or "").strip()
+                or (getattr(company_vu, "webservice_password", None) or "").strip()
+                or hardcoded_ws_key
+            ),
+            archivo_cer_base64=cer_b64 or "",
+            archivo_key_base64=key_b64 or "",
+            clave_fiel=clave_fiel_value,
+        )
+
+    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: normalizar a EXACTAMENTE 2 decimales (ej. 12.23)
+            cantidad = Decimal(str(qty_model.quantity)).quantize(Decimal("0.01"))
+
+            # 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: normalizar a EXACTAMENTE 2 decimales
+                valor_total = Decimal(str(base_total or 0)).quantize(Decimal("0.01"))
+                if cantidad > 0:
+                    # Valor unitario también a EXACTAMENTE 2 decimales
+                    valor_unitario = (valor_total / cantidad).quantize(Decimal("0.01"))
+                else:
+                    valor_unitario = Decimal("0")
+
+                # Valor en dólares: si ya existe, lo usamos; si no, asumimos 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")
+
+                # Normalizar valor en dólares a EXACTAMENTE 2 decimales
+                valor_dolares = valor_dolares.quantize(Decimal("0.01"))
+            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,
+        recipient_email: str | None = None,
+    ) -> 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().lower()
+        # Mapear tipo_operacion al código esperado por el API de COVE
+        # Ejemplos:
+        # - IMP / importación -> "TOCE.IMP"
+        # - EXP / exportación -> "TOCE.EXP"
+        if raw_tipo_operacion in {"imp", "import", "importacion", "importación"}:
+            tipo_operacion = "TOCE.IMP"
+        elif raw_tipo_operacion in {"exp", "export", "exportacion", "exportación"}:
+            tipo_operacion = "TOCE.EXP"
+        else:
+            # Fallback seguro: usar valor por defecto de importación
+            tipo_operacion = "TOCE.IMP"
+        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]
+
+        # Normalizar tipo_figura desde VU: el API externo espera un código corto
+        # (en el ejemplo: "5" para agente aduanal). Hacemos un mapeo simple
+        # desde el texto configurado en la UI.
+        raw_figura = (
+            (ctx.vu.vu_figure_type or "").strip().upper()
+            if ctx.vu and ctx.vu.vu_figure_type
+            else ""
+        )
+        if "AGENTE" in raw_figura:
+            tipo_figura = "5"
+        elif "APODERADO" in raw_figura:
+            tipo_figura = "6"
+        elif "MANDATARIO" in raw_figura:
+            tipo_figura = "7"
+        else:
+            # Fallback: recortar a máximo 10 caracteres para cumplir el esquema
+            tipo_figura = raw_figura[:10]
+
+        correo_destino = (recipient_email or (ctx.vu.vu_email if ctx.vu else None) or "").strip() or None
+
+        return FacturaCoveRequest(
+            configuracion_vu=configuracion_vu,
+            # El RFC de consulta NO debe ser igual al RFC del que registra el comprobante.
+            # Usamos como RFC de consulta el RFC del agente aduanal (customs broker),
+            # y dejamos que configuracion_vu.rfc_usuario_vu represente al contribuyente.
+            rfc_consulta=(
+                (ctx.broker.tax_id or "").strip().upper() if ctx.broker and ctx.broker.tax_id else ""
+            ),
+            tipo_figura=tipo_figura,
+            numero_factura=numero_factura,
+            tipo_operacion=tipo_operacion,
+            patente_aduanal=patente_aduanal,
+            fecha_expedicion=fecha_expedicion,
+            observaciones=ctx.invoice.vu_observations or None,
+            correo_electronico=correo_destino,
+            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.
+        """
+        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()
+            ],
+        )
+

backend/api/v1/modules/a76/factura_cove/tasks.py

@@ -0,0 +1,276 @@
+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 api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceComplianceMx
+
+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 _save_cove_result(
+    db: "Session", invoice_id: int, final_external: CoveExternalResult
+) -> None:
+    """
+    Persiste en la factura el número de COVE y el número de operación VUCEM
+    cuando el servicio externo reporta SUCCESS.
+    """
+    if final_external.status != "success":
+        return
+
+    if not final_external.cove_number and not final_external.vucem_operation_num:
+        return
+
+    invoice = db.get(InvoiceHeader, invoice_id)
+    if not invoice:
+        logger.error("No se encontró la factura %s para guardar COVE", invoice_id)
+        return
+
+    compliance = invoice.compliance_mx
+    if not compliance:
+        # Creamos un registro mínimo de compliance ligado a la factura.
+        compliance = InvoiceComplianceMx(
+            invoice_id=invoice.id,
+            tenant_id=invoice.tenant_id,
+            company_id=invoice.company_id,
+        )
+        db.add(compliance)
+
+    # Idempotencia básica: solo sobrescribir si está vacío o coincide.
+    if final_external.cove_number:
+        current_cove = compliance.edocument or ""
+        new_cove = final_external.cove_number or ""
+        if not current_cove or current_cove == new_cove:
+            compliance.edocument = new_cove
+
+    if final_external.vucem_operation_num:
+        current_op = compliance.vucem_operation_num or ""
+        new_op = final_external.vucem_operation_num or ""
+        if not current_op or current_op == new_op:
+            compliance.vucem_operation_num = new_op
+
+    db.commit()
+
+
+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,
+    recipient_email: str | None = None,
+) -> 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,
+            recipient_email=recipient_email,
+        )
+
+        _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
+
+        # Intentar persistir COVE / número de operación en la factura cuando sea éxito.
+        try:
+            _save_cove_result(db, invoice_id, final_external)
+        except Exception:
+            # No fallamos la tarea por errores de persistencia; solo los registramos.
+            logger.exception("Error guardando COVE en la factura %s", invoice_id)
+
+        _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()
+

backend/api/v1/modules/a76/general_catalogs/company/routes.py

@@ -432,11 +432,18 @@ async def upload_company_certificate(
             detail=f"File type not allowed. Allowed: {', '.join(allowed_exts)}",
         )
         
-    # Validar correspondencia extensión vs tipo (simple check)
-    if "cer" in certificate_type and file_ext != ".cer":
-         raise HTTPException(status_code=400, detail="For this certificate type, file must be .cer")
-    if "key" in certificate_type and file_ext != ".key":
-         raise HTTPException(status_code=400, detail="For this certificate type, file must be .key")
+    # Validar correspondencia extensión vs tipo (simple check, respetando sufijo)
+    ctype = (certificate_type or "").lower()
+    if ctype.endswith("_cer") and file_ext != ".cer":
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="For this certificate type, file must be .cer",
+        )
+    if ctype.endswith("_key") and file_ext != ".key":
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="For this certificate type, file must be .key",
+        )
 
     # Validar tamaño
     content = await file.read()

backend/api/v1/modules/a76/pedmientos/services/pedimentos.py

@@ -3,9 +3,10 @@ Service layer for Pedimentos CRUD operations
 """
 
 import logging
+import re
 from typing import Any, Dict, List, Optional
 
-from sqlalchemy import desc
+from sqlalchemy import desc, func
 from sqlalchemy.orm import Session, joinedload
 from sqlalchemy.orm import selectinload
 from sqlalchemy.exc import IntegrityError
@@ -95,10 +96,21 @@ class PedimentosService:
             if filters.get("status"):
                 query = query.filter(Pedimentos.status == filters["status"])
             if filters.get("client_id"):
-                query = query.filter(
-                    Pedimentos.client_id == filters["client_id"])
+                query = query.filter(Pedimentos.client_id == filters["client_id"])
             if filters.get("year"):
                 query = query.filter(Pedimentos.year == filters["year"])
+            if filters.get("pedimento"):
+                raw_value = str(filters["pedimento"])
+                # Normalizar: dejar solo dígitos (ignorar guiones, espacios, etc.)
+                normalized = re.sub(r"\D", "", raw_value)
+                if normalized:
+                    ped_key_expr = func.concat(
+                        func.coalesce(Pedimentos.year, ""),
+                        func.coalesce(func.substr(Pedimentos.customs_office, 1, 2), ""),
+                        func.coalesce(Pedimentos.license, ""),
+                        func.coalesce(Pedimentos.pedimento_number, ""),
+                    )
+                    query = query.filter(ped_key_expr.ilike(f"%{normalized}%"))
 
         total = query.count()
 

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
@@ -83,6 +84,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(

backend/api/v1/modules/public/reference_data/incoterms/routes.py

@@ -15,11 +15,22 @@ router = APIRouter(prefix="/incoterms")
 async def list_incoterms(
     page: int = Query(1, ge=1, description="Número de página"),
     page_size: int = Query(50, ge=1, le=100, description="Tamaño de página"),
+    code: str = Query(None, description="Filtrar por clave"),
+    description: str = Query(None, description="Filtrar por descripción"),
     db: Session = Depends(get_core_db),
     current_user: dict = Depends(get_current_user),
 ):
     skip = (page - 1) * page_size
     query = db.query(Incoterm)
+    
+    if code:
+        query = query.filter(Incoterm.code.ilike(f"%{code}%"))
+    if description:
+        query = query.filter(
+            (Incoterm.description_es.ilike(f"%{description}%")) |
+            (Incoterm.description_en.ilike(f"%{description}%"))
+        )
+        
     items = query.offset(skip).limit(page_size).all()
     total = query.count()
     return {