docs(changelog): rastreo del stack legacy de descarga (T2025-09-004)

Documenta que el signal procesamiento.py dispara el stack viejo (api_v1 ->
peticiones.py) en paralelo al auto-proceso v2, causando docs sin ligar y
duplicados. Fix del modulo viejo en microservice. Sin cambio de codigo backend.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

CHANGELOG.md

@@ -0,0 +1,57 @@
+# Changelog — EFC Backend
+
+Historial de cambios por ticket (más reciente arriba). Cada entrada: fecha, ticket,
+tipo, repos afectados, qué se hizo y por qué. Reglas del flujo en `../CLAUDE.md`.
+
+## T2025-09-004 — Pertenencia documento→entidad (matching de partidas + FK polimórfica)
+- **Fecha:** 2026-06-24
+- **Tipo:** feature
+- **Repos:** backend + microservice (el fix del stack legacy de descarga está en
+  `microservice/CHANGELOG.md`; frontend sin cambios)
+- **Branch:** `feature/T2025-09-004` · **PR:** (pendiente de aceptación manual)
+- **Qué se hizo:**
+  - Matching documento→partida con frontera `(_|.|$)` en `core/partida_docs.py` (cubre los
+    3 formatos de nombre sin confundir la partida 1 con 11/100).
+  - FK reales `document.partida` / `cove` / `edocument` (nullable, `CASCADE`); los acuses
+    cuelgan de su cove/edoc padre; los documentos nativos (PC, remesa, subidas generales)
+    quedan sin FK.
+  - Resolución central en `Document.save()` vía `core/document_links.py`: liga la FK por
+    `document_type` + nombre de archivo en toda ruta de creación (incluida la ingesta del
+    microservicio); set explícito de la FK en `create_vu_record`.
+  - Comando `backfill_document_links` para poblar la FK en filas existentes (idempotente).
+  - Comando `backfill_document_links_legacy` para documentos LEGADOS de cove/edoc con
+    nomenclatura vieja (otro `pedimento_app`/prefijo, p.ej. `vu_EDC_0201_800_..._{numero}`
+    en un pedimento `25-80-...`): liga por el `numero_cove`/`numero_edocument` único
+    presente en el nombre, sin exigir app ni prefijo. Solo cove/edoc (llaves únicas);
+    partida y nativos quedan fuera. Correr DESPUÉS del backfill estricto.
+  - Comando `dedup_documents` para limpiar documentos duplicados legados (misma entidad +
+    mismo tipo): conserva el más reciente con archivo válido en storage, borra el resto
+    (archivo MinIO si no lo referencia otra fila + fila + ajuste de cuota), `--dry-run`,
+    conteo previo, idempotente. Los duplicados eran **pre-fix**: la descarga ya reemplaza
+    en vez de re-crear desde jun-2026 (microservicio, `post_or_update_document`), verificado
+    con 0 duplicados creados después del fix. Solo aplica a docs ligados a entidad.
+  - Lectura, descarga y borrado SIEMPRE por la FK (id), nunca por nombre. El nombre solo
+    ESTABLECE la FK (en `Document.save()` para altas y en el backfill para filas viejas).
+  - **Rastreo del stack legacy (docs sin ligar/duplicados con datos de hoy):** el signal
+    `api/customs/signals/procesamiento.py` (post_save de Pedimento/Cove/EDocument, p.ej. al
+    cargar un datastage) dispara el stack VIEJO de descarga (`tasks/microservice.py` → api_v1
+    → `utils/peticiones.py` del microservicio), que armaba el nombre con campos crudos (no
+    `pedimento_app`, prefijo `vu_EDC`) y posteaba sin reemplazar → docs sin FK + partidas
+    duplicadas. El auto-proceso (Celery beat `microservice_v2.process_all_organizations`) ya usa
+    el stack v2 correcto. Fix del módulo viejo en `microservice/CHANGELOG.md`. El signal/api_v1
+    se dejan activos (decisión: arreglar `peticiones.py`, no retirarlos); unificar a un solo
+    stack queda como follow-up.
+- **Por qué:** retirar el matching frágil por nombre de archivo (`icontains`/prefijo, que
+  confundía entidades y se rompía con formatos nuevos) y tener la pertenencia
+  documento→entidad como dato real, consultable e íntegro.
+- **Migraciones:** `0004_document_subentidad_fk` (campos, metadata-only),
+  `0005_document_subentidad_idx` (índices con `CREATE INDEX CONCURRENTLY IF NOT EXISTS`,
+  `atomic=False`, idempotente vía `SeparateDatabaseAndState`),
+  `0006_analyze_document` (`ANALYZE document`: refresca estadísticas del planner — sin esto,
+  el prefetch hacía seq scan sobre ~5M filas y los endpoints tardaban ~9s).
+  La tabla `document` tiene ~5M filas: cada índice tarda minutos y NO debe interrumpirse.
+  Recuperación si se corta: índices válidos → `migrate --fake record 0005`; alguno INVALID →
+  `DROP INDEX IF EXISTS "<nombre>";` y reintentar `migrate record`.
+- **Despliegue (orden obligatorio):** aplicar migraciones (0004-0006) → **correr el backfill
+  completo** → recién entonces la lectura/descarga/borrado por FK es correcta. Como NO hay
+  fallback por nombre, un documento sin backfillear queda invisible hasta ligar su FK.

api/customs/management/commands/backfill_document_links.py

@@ -0,0 +1,94 @@
+"""
+Backfill de la FK de sub-entidad en documentos existentes (T2025-09-004).
+
+Liga cada Document a su Partida / Cove / EDocument por nombre de archivo, usando el
+mismo resolver que `Document.save()` (core.document_links). Solo toca documentos aún
+no ligados (idempotente: re-ejecutar converge). Los documentos nativos del pedimento
+(PC, remesa, subidas generales) y los que no matchean ninguna entidad se dejan sin FK.
+
+Uso:
+  python manage.py backfill_document_links [--pedimento UUID] [--organizacion UUID]
+                                           [--offset N] [--limit N] [--dry-run]
+"""
+
+from django.core.management.base import BaseCommand
+from django.db import transaction
+
+from api.customs.models import Pedimento
+from api.record.models import Document
+from core.document_links import SECCION_CAMPO, match_entidad, seccion_de_tipo
+
+# related_name del FK a Pedimento en cada entidad (EDocument usa 'documentos').
+_RELACION = {'partida': 'partidas', 'cove': 'coves', 'edocument': 'documentos'}
+
+
+class Command(BaseCommand):
+    help = "Liga documentos existentes a su sub-entidad (partida/cove/edocument) por nombre de archivo."
+
+    def add_arguments(self, parser):
+        parser.add_argument('--pedimento', type=str, default=None, help='UUID de un solo pedimento')
+        parser.add_argument('--organizacion', type=str, default=None, help='UUID de organización')
+        parser.add_argument('--offset', type=int, default=0)
+        parser.add_argument('--limit', type=int, default=None)
+        parser.add_argument('--dry-run', action='store_true', help='Reporta sin escribir')
+
+    def handle(self, *args, **opts):
+        dry_run = opts['dry_run']
+        peds = Pedimento.objects.all().order_by('created_at', 'id')
+        if opts['pedimento']:
+            peds = peds.filter(id=opts['pedimento'])
+        if opts['organizacion']:
+            peds = peds.filter(organizacion_id=opts['organizacion'])
+
+        offset = opts['offset'] or 0
+        peds = peds[offset:offset + opts['limit']] if opts['limit'] else peds[offset:]
+
+        if dry_run:
+            self.stdout.write(self.style.WARNING("== DRY-RUN: no se escribe nada =="))
+
+        stats = {'pedimentos': 0, 'partida': 0, 'cove': 0, 'edocument': 0, 'sin_match': 0}
+        for ped in peds.iterator():
+            self._procesar_pedimento(ped, dry_run, stats)
+
+        self.stdout.write("")
+        self.stdout.write(self.style.SUCCESS(
+            f"Pedimentos: {stats['pedimentos']} | ligados → partida={stats['partida']} "
+            f"cove={stats['cove']} edocument={stats['edocument']} | "
+            f"tipados sin entidad que matchee={stats['sin_match']}"
+        ))
+
+    def _procesar_pedimento(self, ped, dry_run, stats):
+        # Solo documentos aún NO ligados (idempotente).
+        docs = list(Document.objects.filter(
+            pedimento=ped, partida__isnull=True, cove__isnull=True, edocument__isnull=True,
+        ))
+        if not docs:
+            return
+        stats['pedimentos'] += 1
+
+        # Precargar las entidades del pedimento una sola vez; el match es en memoria.
+        entidades = {sec: list(getattr(ped, rel).all()) for sec, rel in _RELACION.items()}
+        app = ped.pedimento_app
+
+        lote = []
+        for doc in docs:
+            seccion = seccion_de_tipo(doc.document_type_id)
+            if not seccion or not doc.archivo:
+                continue  # nativo de pedimento o sin archivo
+            inst = match_entidad(doc.archivo.name, seccion, app, entidades[seccion])
+            if inst is None:
+                stats['sin_match'] += 1
+                continue
+            setattr(doc, SECCION_CAMPO[seccion], inst)
+            lote.append(doc)
+            stats[seccion] += 1
+
+        # SELECT + COUNT previo antes de escribir (estándar de la organización).
+        self.stdout.write(
+            f"  {ped.pedimento_app}: {len(lote)} de {len(docs)} doc(s) sin ligar serán ligados"
+        )
+        if lote and not dry_run:
+            with transaction.atomic():
+                Document.objects.bulk_update(
+                    lote, ['partida', 'cove', 'edocument'], batch_size=1000
+                )

api/customs/management/commands/backfill_document_links_legacy.py

@@ -0,0 +1,94 @@
+"""
+Backfill LEGADO de la FK de cove/edocument por número único (T2025-09-004).
+
+Para documentos viejos cuyo nombre quedó con una nomenclatura olvidada —otro
+`pedimento_app` y/u otro prefijo (p.ej. `vu_EDC_0201_800_..._04382515ZIFF5_hex.pdf`
+en un pedimento cuyo `pedimento_app` es `25-80-3452-5000586`)— el matcher estricto
+de `backfill_document_links` no liga porque arma `vu_ed_{pedimento_app}_{numero}`.
+Pero el **número de cove/edoc** (único y largo) sí está en el nombre y la entidad
+existe.
+
+Este comando, SOLO para cove y edocument (llaves únicas; partida queda fuera por ser
+enteros cortos con colisiones, y los nativos no tienen entidad), liga la FK buscando
+el `numero_cove`/`numero_edocument` como token en el nombre, sin exigir app ni prefijo.
+
+Correr DESPUÉS de `backfill_document_links` (solo toca lo que quedó sin ligar).
+
+Uso:
+  python manage.py backfill_document_links_legacy [--pedimento UUID] [--organizacion UUID] [--dry-run]
+"""
+
+from django.core.management.base import BaseCommand
+from django.db import transaction
+
+from api.customs.models import Cove, EDocument
+from api.record.models import Document
+from core.document_links import (
+    COVE_TYPES, EDOCUMENT_TYPES, SECCION_CAMPO, SECCION_LLAVE,
+    numero_en_nombre, seccion_de_tipo,
+)
+
+
+class Command(BaseCommand):
+    help = "Liga documentos legados de cove/edoc (FK nula) por su número único en el nombre, ignorando app/prefijo viejos. Correr después de backfill_document_links."
+
+    def add_arguments(self, parser):
+        parser.add_argument('--pedimento', type=str, default=None)
+        parser.add_argument('--organizacion', type=str, default=None)
+        parser.add_argument('--dry-run', action='store_true', help='Reporta sin escribir')
+
+    def handle(self, *args, **opts):
+        dry_run = opts['dry_run']
+        tipos = sorted(COVE_TYPES | EDOCUMENT_TYPES)
+        base = Document.objects.filter(
+            partida__isnull=True, cove__isnull=True, edocument__isnull=True,
+            document_type_id__in=tipos,
+        )
+        if opts['pedimento']:
+            base = base.filter(pedimento_id=opts['pedimento'])
+        if opts['organizacion']:
+            base = base.filter(pedimento__organizacion_id=opts['organizacion'])
+
+        if dry_run:
+            self.stdout.write(self.style.WARNING("== DRY-RUN: no se escribe nada =="))
+
+        # order_by() limpia el ordering por defecto del modelo (created_at), que si
+        # no se quita se cuela en el DISTINCT y duplica los pedimento_id.
+        ped_ids = list(base.order_by().values_list('pedimento_id', flat=True).distinct())
+        stats = {'pedimentos': 0, 'cove': 0, 'edocument': 0, 'ambiguos': 0, 'sin_match': 0}
+
+        for ped_id in ped_ids:
+            docs = list(base.filter(pedimento_id=ped_id))
+            entidades = {
+                'cove': list(Cove.objects.filter(pedimento_id=ped_id)),
+                'edocument': list(EDocument.objects.filter(pedimento_id=ped_id)),
+            }
+            lote = []
+            for doc in docs:
+                seccion = seccion_de_tipo(doc.document_type_id)
+                llave = SECCION_LLAVE[seccion]
+                matches = [
+                    e for e in entidades[seccion]
+                    if numero_en_nombre(doc.archivo.name, getattr(e, llave))
+                ]
+                if len(matches) == 1:
+                    setattr(doc, SECCION_CAMPO[seccion], matches[0])
+                    lote.append(doc)
+                    stats[seccion] += 1
+                elif len(matches) > 1:
+                    stats['ambiguos'] += 1  # número ambiguo: se deja sin ligar
+                else:
+                    stats['sin_match'] += 1  # ni por número aparece la entidad
+
+            if lote:
+                stats['pedimentos'] += 1
+                self.stdout.write("  %s: %d doc(s) ligados por número" % (str(ped_id)[:8], len(lote)))
+                if not dry_run:
+                    with transaction.atomic():
+                        Document.objects.bulk_update(lote, ['cove', 'edocument'], batch_size=500)
+
+        self.stdout.write("")
+        self.stdout.write(self.style.SUCCESS(
+            "Pedimentos: %d | ligados → cove=%d edocument=%d | ambiguos: %d | sin match ni por número: %d"
+            % (stats['pedimentos'], stats['cove'], stats['edocument'], stats['ambiguos'], stats['sin_match'])
+        ))

api/customs/management/commands/dedup_documents.py

@@ -0,0 +1,138 @@
+"""
+Limpieza de documentos duplicados legados (T2025-09-004).
+
+Un mismo documento (misma sub-entidad + mismo document_type) quedó con varias
+filas porque, antes del fix de reemplazo (microservicio, jun-2026), cada descarga
+re-creaba en vez de reemplazar. La creación ya está corregida; esto solo limpia lo
+viejo.
+
+Estrategia: encuentra los grupos duplicados con UNA agregación global por FK
+(no itera pedimento por pedimento — hay ~110k pedimentos y casi ninguno tiene
+duplicados). Por cada grupo (FK partida/cove/edocument + document_type, con >1
+fila): conserva el MÁS RECIENTE cuyo archivo exista en storage (si ninguno existe,
+conserva el más reciente y NO borra el grupo entero) y elimina el resto —archivo
+en MinIO (si no lo referencia otra fila) + fila + ajuste de cuota vía Document.delete().
+
+Solo toca documentos ligados a una entidad (partida/cove/edoc). NO toca documentos
+nativos del pedimento ni subidas sin FK.
+
+Uso:
+  python manage.py dedup_documents [--pedimento UUID] [--organizacion UUID]
+                                   [--offset N] [--limit N] [--dry-run]
+"""
+
+from django.core.management.base import BaseCommand
+from django.db import transaction
+from django.db.models import Count
+
+from api.record.models import Document
+from api.utils.storage_service import storage_service
+
+# Campos FK de sub-entidad sobre los que se detectan duplicados.
+_CAMPOS_FK = ('partida_id', 'cove_id', 'edocument_id')
+
+
+class Command(BaseCommand):
+    help = "Elimina documentos duplicados legados (misma entidad + mismo tipo), conservando el más reciente con archivo válido."
+
+    def add_arguments(self, parser):
+        parser.add_argument('--pedimento', type=str, default=None)
+        parser.add_argument('--organizacion', type=str, default=None)
+        parser.add_argument('--offset', type=int, default=0, help='Saltar los primeros N grupos')
+        parser.add_argument('--limit', type=int, default=None, help='Procesar máximo N grupos')
+        parser.add_argument('--dry-run', action='store_true', help='Reporta sin borrar')
+
+    def handle(self, *args, **opts):
+        dry_run = opts['dry_run']
+        base = Document.objects.all()
+        if opts['pedimento']:
+            base = base.filter(pedimento_id=opts['pedimento'])
+        if opts['organizacion']:
+            base = base.filter(pedimento__organizacion_id=opts['organizacion'])
+
+        offset = opts['offset'] or 0
+        limit = opts['limit']
+        if dry_run:
+            self.stdout.write(self.style.WARNING("== DRY-RUN: no se borra nada =="))
+
+        stats = {'grupos': 0, 'eliminados': 0, 'bytes': 0, 'sin_archivo_valido': 0}
+        visto = 0  # índice global de grupos (para offset/limit)
+        for campo in _CAMPOS_FK:
+            # Una sola agregación: todos los grupos duplicados de esta FK.
+            grupos = (
+                base.filter(**{campo + '__isnull': False})
+                .values('pedimento_id', campo, 'document_type_id')
+                .annotate(n=Count('id'))
+                .filter(n__gt=1)
+                .order_by()  # limpia el ordering por defecto del modelo (rompe el GROUP BY)
+            )
+            for g in grupos.iterator():
+                if visto < offset:
+                    visto += 1
+                    continue
+                if limit is not None and stats['grupos'] >= limit:
+                    break
+                visto += 1
+                self._dedup_grupo(campo, g, dry_run, stats)
+            if limit is not None and stats['grupos'] >= limit:
+                break
+
+        mb = stats['bytes'] / (1024 * 1024)
+        self.stdout.write("")
+        self.stdout.write(self.style.SUCCESS(
+            "Grupos con duplicados: %d | filas eliminadas: %d | espacio liberado: %.1f MB | grupos sin archivo válido (se conservó el más reciente): %d"
+            % (stats['grupos'], stats['eliminados'], mb, stats['sin_archivo_valido'])
+        ))
+
+    def _dedup_grupo(self, campo, g, dry_run, stats):
+        docs = list(
+            Document.objects.filter(
+                pedimento_id=g['pedimento_id'],
+                document_type_id=g['document_type_id'],
+                **{campo: g[campo]},
+            ).select_related('pedimento').order_by('-created_at')
+        )
+        if len(docs) < 2:
+            return
+        conservado = self._elegir_conservado(docs, dry_run, stats)
+        a_borrar = [d for d in docs if d.id != conservado.id]
+        if not a_borrar:
+            return
+        stats['grupos'] += 1
+        # SELECT + COUNT previo (estándar de la organización): reportar antes de borrar.
+        self.stdout.write(
+            "  ped=%s %s=%s type=%s: conservar %s, eliminar %d (%s)"
+            % (str(g['pedimento_id'])[:8], campo, g[campo], g['document_type_id'],
+               str(conservado.id)[:8], len(a_borrar), ', '.join(str(d.id)[:8] for d in a_borrar))
+        )
+        if not dry_run:
+            for d in a_borrar:
+                self._borrar(d, stats)
+
+    def _elegir_conservado(self, docs_desc, dry_run, stats):
+        """docs_desc viene ordenado por -created_at. En dry-run conserva el más
+        reciente sin tocar storage; en ejecución real, el más reciente cuyo
+        archivo exista en MinIO (fallback: el más reciente, para no borrar todo)."""
+        if dry_run:
+            return docs_desc[0]
+        for d in docs_desc:
+            try:
+                if d.archivo and storage_service.file_exists(d.archivo.name):
+                    return d
+            except Exception:
+                continue
+        stats['sin_archivo_valido'] += 1
+        return docs_desc[0]
+
+    def _borrar(self, doc, stats):
+        with transaction.atomic():
+            # Borrar el archivo en MinIO solo si ninguna OTRA fila lo referencia.
+            nombre = doc.archivo.name if doc.archivo else None
+            if nombre and not Document.objects.filter(archivo=nombre).exclude(id=doc.id).exists():
+                try:
+                    storage_service.delete_file(nombre)
+                except Exception as e:
+                    self.stdout.write(self.style.WARNING(f"      no se pudo borrar de storage {nombre}: {e}"))
+            stats['bytes'] += doc.size or 0
+            doc.delete()  # ajusta la cuota de almacenamiento (UsoAlmacenamiento)
+            stats['eliminados'] += 1

api/customs/management/commands/fix_partidas_error.py

@@ -46,7 +46,6 @@ Uso:
 """
 import io
 import posixpath
-import re
 
 from django.core.management.base import BaseCommand, CommandError
 from django.db import transaction
@@ -56,6 +55,7 @@ from django.db.models.functions import Length
 from api.customs.models import Partida, Pedimento
 from api.record.models import Document
 from api.utils.minio_client import minio_client
+from core.partida_docs import es_doc_de_partida
 
 _PT_REQUEST = 17
 _PT_ERROR = 18
@@ -384,24 +384,13 @@ class Command(BaseCommand):
 
     def _docs_de_partida(self, docs, pedimento_app, numero_partida):
         """
-        Naming actual : vu_PT_{pedimento_app}_{numero} seguido de "_" o "."
-                        (cubre éxito canónico, sufijos de unicidad del storage,
-                        REQUEST y ERROR; "_" evita confundir partida 1 con 11)
-        Naming legacy : vu_PT_..._{numero}.xml (número de partida al final)
+        Asigna documentos a una partida por nombre de archivo. La regla (frontera
+        _/./fin de cadena + formato legacy) vive en core.partida_docs como fuente
+        única, compartida con el serializer y los handlers de borrado/descarga.
         """
-        prefijo = f"vu_pt_{pedimento_app}_{numero_partida}".lower()
-        legacy_re = re.compile(
-            rf"^vu_pt_.+_{re.escape(str(numero_partida))}\.xml$", re.IGNORECASE
-        )
         asignados = {}
         for doc in docs:
-            base = posixpath.basename(doc.archivo.name or "").lower()
-            es_actual = (
-                base.startswith(prefijo)
-                and len(base) > len(prefijo)
-                and base[len(prefijo)] in "_."
-            )
-            if es_actual or legacy_re.match(base):
+            if es_doc_de_partida(doc.archivo.name, pedimento_app, numero_partida):
                 asignados[doc.id] = doc
         return list(asignados.values())
 

api/customs/serializers.py

@@ -11,9 +11,11 @@ from api.customs.models import (
 )
 from django.db import models
 from django.db.models import Q
-from api.record.models import Document  # Asegúrate de importar el modelo Documento
 from api.record.serializers import DocumentSerializer
 from api.vucem.serializers import VucemSerializer
+import logging
+
+logger = logging.getLogger(__name__)
 
 class PedimentoSerializer(serializers.ModelSerializer):
     documentos_count = serializers.SerializerMethodField()
@@ -48,31 +50,18 @@ class PartidaSerializer(serializers.ModelSerializer):
     documentos = serializers.SerializerMethodField()
 
     def get_documentos(self, obj):
-        if not obj or not getattr(obj, 'pedimento', None):
+        if not obj:
             return []
-        if not obj or not getattr(obj, 'numero_partida', None):
-            return []
-
         try:
-            pedimento_app = str(obj.pedimento.pedimento_app).strip()
-            numero = str(obj.numero_partida).strip()
-            # Incluir pedimento_app en el patrón para evitar falsos positivos
-            # entre partidas con números cortos (1 matchearía 10, 100, etc.)
-            patron = f"vu_PT_{pedimento_app}_{numero}_"
-
-            # 17 = REQUEST partida, 18 = ERROR partida
-            qs = Document.objects.filter(
-                pedimento=obj.pedimento,
-                archivo__icontains=patron,
-            ).exclude(document_type_id__in=[17, 18])
-
-            if hasattr(obj, 'organizacion') and obj.organizacion:
-                qs = qs.filter(organizacion=obj.organizacion)
-
-            serializer = DocumentSerializer(qs, many=True, context=self.context)
-            return serializer.data
-
-        except Exception:
+            # Documentos de respuesta de la partida (tipo 1) vía la FK real
+            # document.partida. 'documentos_vu' lo precarga el ViewSet con prefetch;
+            # si no está, se consulta directo (retrieve u otros callers).
+            docs = getattr(obj, 'documentos_vu', None)
+            if docs is None:
+                docs = list(obj.documents.filter(document_type_id=1).select_related('pedimento', 'fuente'))
+            return DocumentSerializer(docs, many=True, context=self.context).data
+        except Exception as e:
+            logger.warning("get_documentos partida %s: %s", getattr(obj, 'id', '?'), e)
             return []
     class Meta:
         model = Partida
@@ -163,43 +152,18 @@ class EDocumentSerializer(serializers.ModelSerializer):
     documentos = serializers.SerializerMethodField()
 
     def get_documentos(self, obj):
-        """
-        Busca documentos en la tabla `document` que coincidan con el
-        `numero_edocument` dentro del nombre del archivo (`archivo`). Se
-        filtra por organización para evitar devolver documentos de otras orgs.
-        Devuelve la serialización completa de los documentos encontrados:
-        1. Empiecen con 'vu_EDOCUMENT' en el nombre del archivo
-        2. Terminen con el numero_edocument + .xml
-        3. Pertenezcan a la misma organización
-        """
-        if not obj or not getattr(obj, 'numero_edocument', None):
+        """Documentos del e-documento (incluye acuse y errores; excluye solo los
+        REQUEST 21/25) vía la FK real document.edocument. 'documentos_vu' lo
+        precarga el ViewSet con prefetch; si no está, se consulta directo."""
+        if not obj:
             return []
-
-        if not obj or not getattr(obj, 'pedimento', None):
-            return []
-
-        # if not obj or not getattr(obj, 'pedimento_id', None):
-        #     return []
-
         try:
-            numero = str(obj.numero_edocument).strip()
-            # id_pedimento = str(obj.pedimento_id).strip()
-
-            # excluir solo request (21, 25); errores (22, 26) se incluyen para detección en frontend
-            qs = Document.objects.filter(
-                pedimento=obj.pedimento,
-                archivo__icontains=numero,
-            ).exclude(document_type_id__in=[21, 25])
-
-            # Filtro por organización si aplica
-            if hasattr(obj, 'organizacion') and obj.organizacion:
-                qs = qs.filter(organizacion=obj.organizacion)
-            
-            serializer = DocumentSerializer(qs, many=True, context=self.context)
-            return serializer.data
-        
-        except Exception:
-            # En caso de cualquier error (por ejemplo, importaciones circulares), devolver lista vacía
+            docs = getattr(obj, 'documentos_vu', None)
+            if docs is None:
+                docs = list(obj.documents.exclude(document_type_id__in=[21, 25]).select_related('pedimento', 'fuente'))
+            return DocumentSerializer(docs, many=True, context=self.context).data
+        except Exception as e:
+            logger.warning("get_documentos edocument %s: %s", getattr(obj, 'id', '?'), e)
             return []
 
     class Meta:
@@ -255,39 +219,18 @@ class CoveSerializer(serializers.ModelSerializer):
         return attrs
 
     def get_documentos(self, obj):
-        """
-        Busca documentos en la tabla `document` que coincidan con el
-        `numero_cove` dentro del nombre del archivo (`archivo`). Se
-        filtra por organización para evitar devolver documentos de otras orgs.
-        Devuelve la serialización completa de los documentos encontrados:
-        1. Empiecen con 'vu_COVE' en el nombre del archivo
-        2. Terminen con el numero_cove + .xml
-        3. Pertenezcan a la misma organización
-        """
-        if not obj or not getattr(obj, 'numero_cove', None):
+        """Documentos del cove (incluye acuse cove y errores; excluye solo los
+        REQUEST 19/23) vía la FK real document.cove. 'documentos_vu' lo precarga
+        el ViewSet con prefetch; si no está, se consulta directo."""
+        if not obj:
             return []
-
-        if not obj or not getattr(obj, 'pedimento', None):
-            return []
-
         try:
-            numero = str(obj.numero_cove).strip()
-
-            # Excluir solo request (19, 23); errores (20, 24) se incluyen para detección en frontend
-            qs = Document.objects.filter(
-                pedimento=obj.pedimento,
-                archivo__icontains=numero,
-            ).exclude(document_type_id__in=[19, 23])
-
-            # Filtro por organización si aplica
-            if hasattr(obj, 'organizacion') and obj.organizacion:
-                qs = qs.filter(organizacion=obj.organizacion)
-
-            serializer = DocumentSerializer(qs, many=True, context=self.context)
-            return serializer.data
-
-        except Exception:
-            # En caso de cualquier error (por ejemplo, importaciones circulares), devolver lista vacía
+            docs = getattr(obj, 'documentos_vu', None)
+            if docs is None:
+                docs = list(obj.documents.exclude(document_type_id__in=[19, 23]).select_related('pedimento', 'fuente'))
+            return DocumentSerializer(docs, many=True, context=self.context).data
+        except Exception as e:
+            logger.warning("get_documentos cove %s: %s", getattr(obj, 'id', '?'), e)
             return []
 
 class ImportadorSerializer(serializers.ModelSerializer):

api/customs/tests.py

@@ -236,7 +236,11 @@ class BulkCreateDocumentReplaceTests(APITestCase):
 from io import StringIO
 from types import SimpleNamespace
 from django.core.management import call_command
-from django.test import TestCase
+from django.test import TestCase, SimpleTestCase
+from core.partida_docs import es_doc_de_partida, patron_regex_partida
+from api.customs.serializers import PartidaSerializer, CoveSerializer, EDocumentSerializer
+from api.customs.models import Partida
+from api.record.models import Document, DocumentType
 
 
 XML_RESPUESTA_VALIDA = (
@@ -496,3 +500,347 @@ class FixPartidasErrorCommandTests(TestCase):
 
         self.assertEqual(ids_p1, {1, 2})
         self.assertEqual(ids_p11, {3, 4})
+
+
+class PartidaDocsHelperTests(SimpleTestCase):
+    """Matching documento→partida (core.partida_docs), sin BD."""
+
+    APP = "24-01-3420-1234567"
+
+    def test_es_doc_de_partida_cubre_los_tres_formatos(self):
+        for nombre in (
+            f"documents/vu_PT_{self.APP}_1",             # #1 sin extensión
+            f"documents/vu_PT_{self.APP}_1.xml",         # #2
+            f"documents/vu_PT_{self.APP}_1_a1b2c3.xml",  # #3 sufijo hex del storage
+            f"documents/vu_PT_{self.APP}_1_REQUEST.xml", # REQUEST (coincide por nombre)
+        ):
+            self.assertTrue(es_doc_de_partida(nombre, self.APP, 1), nombre)
+
+    def test_es_doc_de_partida_no_confunde_1_con_11_ni_100(self):
+        for n in (11, 12, 100):
+            self.assertFalse(es_doc_de_partida(f"vu_PT_{self.APP}_{n}.xml", self.APP, 1))
+            self.assertFalse(es_doc_de_partida(f"vu_PT_{self.APP}_{n}_x.xml", self.APP, 1))
+        # a la inversa: la 11 sí coincide con la 11, no con la 1
+        self.assertTrue(es_doc_de_partida(f"vu_PT_{self.APP}_11.xml", self.APP, 11))
+        self.assertFalse(es_doc_de_partida(f"vu_PT_{self.APP}_1.xml", self.APP, 11))
+
+    def test_es_doc_de_partida_case_insensitive_y_con_ruta(self):
+        self.assertTrue(es_doc_de_partida(f"ORG/X/VU_PT_{self.APP}_1.XML", self.APP, 1))
+
+    def test_es_doc_de_partida_vacio_o_none(self):
+        self.assertFalse(es_doc_de_partida("", self.APP, 1))
+        self.assertFalse(es_doc_de_partida(None, self.APP, 1))
+
+    def test_es_doc_de_partida_legacy_segun_flag(self):
+        legacy = "org/x/vu_PT_010Imp_034_3420_1234567_1.xml"  # número de partida al final
+        self.assertTrue(es_doc_de_partida(legacy, self.APP, 1, incluir_legacy=True))
+        self.assertFalse(es_doc_de_partida(legacy, self.APP, 1, incluir_legacy=False))
+        # el formato legacy tampoco confunde la 1 con la 11
+        legacy11 = "org/x/vu_PT_010Imp_034_3420_1234567_11.xml"
+        self.assertFalse(es_doc_de_partida(legacy11, self.APP, 1, incluir_legacy=True))
+
+    def test_patron_regex_partida_semantica(self):
+        import re
+        rx = re.compile(patron_regex_partida(self.APP, 1), re.IGNORECASE)
+        self.assertTrue(rx.search(f"documents/vu_PT_{self.APP}_1"))
+        self.assertTrue(rx.search(f"documents/vu_PT_{self.APP}_1.xml"))
+        self.assertTrue(rx.search(f"documents/vu_PT_{self.APP}_1_a1b2.xml"))
+        self.assertFalse(rx.search(f"documents/vu_PT_{self.APP}_11.xml"))
+        self.assertFalse(rx.search(f"documents/vu_PT_{self.APP}_100.xml"))
+        # pedimento_app se trata como literal: otro pedimento no coincide
+        self.assertFalse(rx.search("documents/vu_PT_99-99-9999-9999999_1.xml"))
+        # legacy solo cuando se pide
+        rxl = re.compile(patron_regex_partida(self.APP, 1, incluir_legacy=True), re.IGNORECASE)
+        self.assertTrue(rxl.search("vu_PT_010Imp_034_3420_1234567_1.xml"))
+
+
+class PartidaDocumentosSerializerTests(TestCase):
+    """get_documentos (PartidaSerializer) y el prefetch de PartidaViewSet asignan
+    los documentos correctos a cada partida por nombre de archivo."""
+
+    APP = "24-01-3420-1234567"
+
+    def setUp(self):
+        from api.organization.models import Organizacion
+        from api.licence.models import Licencia
+        from .models import Pedimento
+
+        self.licencia = Licencia.objects.create(nombre="LicPartDocs", almacenamiento=100)
+        self.org = Organizacion.objects.create(
+            nombre="OrgPartDocs", licencia=self.licencia, is_active=True, is_verified=True
+        )
+        self.pedimento = Pedimento.objects.create(
+            organizacion=self.org, pedimento="1234567", pedimento_app=self.APP,
+            aduana="034", patente="3420", numero_operacion="12345678",
+        )
+        self.p1 = Partida.objects.create(
+            pedimento=self.pedimento, organizacion=self.org, numero_partida=1, descargado=True
+        )
+        self.p11 = Partida.objects.create(
+            pedimento=self.pedimento, organizacion=self.org, numero_partida=11, descargado=True
+        )
+        self.type_resp = DocumentType.objects.get_or_create(id=1, defaults={"nombre": "XML"})[0]
+        self.type_req = DocumentType.objects.get_or_create(id=17, defaults={"nombre": "PT Request"})[0]
+
+    def _doc(self, filename, doc_type=None):
+        return Document.objects.create(
+            organizacion=self.org, pedimento=self.pedimento,
+            document_type=doc_type or self.type_resp,
+            archivo=f"documents/{filename}", size=100, extension="xml",
+        )
+
+    def _blob(self, data):
+        return " ".join(d["archivo"] for d in data["documentos"])
+
+    def test_get_documentos_tres_formatos_sin_confundir(self):
+        self._doc(f"vu_PT_{self.APP}_1")                          # #1 sin extensión
+        self._doc(f"vu_PT_{self.APP}_1.xml")                      # #2
+        self._doc(f"vu_PT_{self.APP}_1_a1b2c3.xml")              # #3 hex
+        self._doc(f"vu_PT_{self.APP}_1_REQUEST.xml", self.type_req)  # tipo 17: no debe salir
+        self._doc(f"vu_PT_{self.APP}_11.xml")                     # partida 11
+        self._doc(f"vu_PT_{self.APP}_11_a1b2c3.xml")             # partida 11
+
+        data = PartidaSerializer(self.p1).data  # sin contexto -> camino fallback
+        blob = self._blob(data)
+        # los 3 documentos tipo-1 de la partida 1; el REQUEST (17) excluido
+        self.assertEqual(len(data["documentos"]), 3)
+        self.assertIn(f"vu_PT_{self.APP}_1.xml", blob)
+        self.assertIn(f"vu_PT_{self.APP}_1_a1b2c3.xml", blob)
+        self.assertNotIn("_11", blob)       # no arrastra la partida 11
+        self.assertNotIn("REQUEST", blob)   # no incluye el REQUEST tipo 17
+
+    def test_get_documentos_incluye_legacy(self):
+        self._doc("vu_PT_010Imp_034_3420_1234567_1.xml")  # legacy tipo 1, número al final
+        data = PartidaSerializer(self.p1).data
+        self.assertEqual(len(data["documentos"]), 1)
+
+    def test_get_documentos_via_fk(self):
+        # save() liga la FK al crear; get_documentos lee por la FK real.
+        d1 = self._doc(f"vu_PT_{self.APP}_1.xml")
+        self._doc(f"vu_PT_{self.APP}_11.xml")
+        d1.refresh_from_db()
+        self.assertEqual(d1.partida_id, self.p1.id)  # FK ligada en save()
+        data = PartidaSerializer(self.p1).data
+        self.assertEqual(len(data["documentos"]), 1)
+        self.assertIn(f"vu_PT_{self.APP}_1.xml", self._blob(data))
+        self.assertNotIn("_11", self._blob(data))
+
+    def test_patron_regex_partida_en_bd(self):
+        d1 = self._doc(f"vu_PT_{self.APP}_1")
+        d2 = self._doc(f"vu_PT_{self.APP}_1.xml")
+        d3 = self._doc(f"vu_PT_{self.APP}_1_a1b2c3.xml")
+        d_otro = self._doc(f"vu_PT_{self.APP}_11.xml")
+        ids = set(
+            Document.objects.filter(
+                pedimento=self.pedimento,
+                archivo__iregex=patron_regex_partida(self.APP, 1),
+            ).values_list("id", flat=True)
+        )
+        self.assertEqual(ids, {d1.id, d2.id, d3.id})
+        self.assertNotIn(d_otro.id, ids)
+
+    def test_prefetch_documentos_vu_evita_n_plus_1(self):
+        from django.db.models import Prefetch
+        self._doc(f"vu_PT_{self.APP}_1.xml")
+        self._doc(f"vu_PT_{self.APP}_11.xml")
+        prefetch = Prefetch(
+            'documents',
+            queryset=Document.objects.filter(document_type_id=1).select_related('pedimento'),
+            to_attr='documentos_vu',
+        )
+        partidas = list(
+            Partida.objects.filter(pedimento=self.pedimento)
+            .order_by('numero_partida').prefetch_related(prefetch)
+        )
+        # Serializar la lista no dispara consultas extra: todo viene del prefetch.
+        with self.assertNumQueries(0):
+            data = PartidaSerializer(partidas, many=True).data
+        por_num = {d['numero_partida']: d['documentos'] for d in data}
+        self.assertEqual(len(por_num[1]), 1)
+        self.assertEqual(len(por_num[11]), 1)
+
+
+class DocumentLinksHelperTests(SimpleTestCase):
+    """Resolver tipo→sección y matcher por frontera (core.document_links), sin BD."""
+
+    # pedimento_app con guiones bajos (caso real): no se puede extraer la llave
+    # partiendo por '_'; por eso se itera la entidad con su número exacto.
+    APP = "0101_230_1703_3004804"
+
+    def test_seccion_de_tipo(self):
+        from core.document_links import seccion_de_tipo
+        self.assertEqual(seccion_de_tipo(1), 'partida')
+        self.assertEqual(seccion_de_tipo(8), 'cove')
+        self.assertEqual(seccion_de_tipo(7), 'cove')        # acuse cove
+        self.assertEqual(seccion_de_tipo(5), 'edocument')
+        self.assertEqual(seccion_de_tipo(4), 'edocument')   # acuse edoc
+        self.assertIsNone(seccion_de_tipo(2))               # PC nativo
+        self.assertIsNone(seccion_de_tipo(3))               # remesa nativo
+        self.assertIsNone(seccion_de_tipo(None))
+
+    def test_coincide_cove(self):
+        from core.document_links import coincide
+        self.assertTrue(coincide(f"documents/vu_COVE_{self.APP}_654001.xml", 'cove', self.APP, "654001"))
+        self.assertTrue(coincide(f"documents/vu_AC_COVE_{self.APP}_654001.pdf", 'cove', self.APP, "654001"))
+        self.assertTrue(coincide(f"documents/vu_COVE_{self.APP}_654001_REQUEST.xml", 'cove', self.APP, "654001"))
+        # colisión de prefijo: 654001 no debe matchear 6540012
+        self.assertFalse(coincide(f"documents/vu_COVE_{self.APP}_6540012.xml", 'cove', self.APP, "654001"))
+
+    def test_coincide_edocument(self):
+        from core.document_links import coincide
+        self.assertTrue(coincide(f"documents/vu_ED_{self.APP}_EDOC001.pdf", 'edocument', self.APP, "EDOC001"))
+        self.assertTrue(coincide(f"documents/vu_AC_{self.APP}_EDOC001.pdf", 'edocument', self.APP, "EDOC001"))
+        self.assertFalse(coincide(f"documents/vu_ED_{self.APP}_EDOC0011.pdf", 'edocument', self.APP, "EDOC001"))
+
+
+class DocumentFKResolutionTests(TestCase):
+    """save()-resolución de FK por sección, lectura cove/edoc por FK y backfill."""
+
+    APP = "0101_230_1703_3004804"  # con guiones bajos
+
+    def setUp(self):
+        from api.organization.models import Organizacion
+        from api.licence.models import Licencia
+        from .models import Pedimento, Cove, EDocument
+
+        self.licencia = Licencia.objects.create(nombre="LicFK", almacenamiento=100)
+        self.org = Organizacion.objects.create(
+            nombre="OrgFK", licencia=self.licencia, is_active=True, is_verified=True
+        )
+        self.pedimento = Pedimento.objects.create(
+            organizacion=self.org, pedimento="1234567", pedimento_app=self.APP,
+            aduana="034", patente="3420", numero_operacion="12345678",
+        )
+        self.partida = Partida.objects.create(
+            pedimento=self.pedimento, organizacion=self.org, numero_partida=3, descargado=True
+        )
+        self.cove = Cove.objects.create(
+            pedimento=self.pedimento, organizacion=self.org, numero_cove="654001"
+        )
+        self.edoc = EDocument.objects.create(
+            pedimento=self.pedimento, organizacion=self.org, numero_edocument="EDOC001"
+        )
+        for tid, nombre in [(1, "PT"), (8, "COVE"), (7, "AC_COVE"), (5, "ED"),
+                            (4, "AC_ED"), (2, "PC"), (19, "COVE_REQ")]:
+            DocumentType.objects.get_or_create(id=tid, defaults={"nombre": nombre})
+
+    def _doc(self, filename, type_id):
+        return Document.objects.create(
+            organizacion=self.org, pedimento=self.pedimento, document_type_id=type_id,
+            archivo=f"documents/{filename}", size=100, extension="xml",
+        )
+
+    def test_save_liga_fk_por_seccion(self):
+        d_pt = self._doc(f"vu_PT_{self.APP}_3.xml", 1)
+        d_cv = self._doc(f"vu_COVE_{self.APP}_654001.xml", 8)
+        d_accv = self._doc(f"vu_AC_COVE_{self.APP}_654001.pdf", 7)
+        d_ed = self._doc(f"vu_ED_{self.APP}_EDOC001.pdf", 5)
+        d_aced = self._doc(f"vu_AC_{self.APP}_EDOC001.pdf", 4)
+        d_pc = self._doc(f"vu_PC_{self.APP}.xml", 2)  # nativo: sin FK
+        for d in (d_pt, d_cv, d_accv, d_ed, d_aced, d_pc):
+            d.refresh_from_db()
+        self.assertEqual(d_pt.partida_id, self.partida.id)
+        self.assertEqual(d_cv.cove_id, self.cove.id)
+        self.assertEqual(d_accv.cove_id, self.cove.id)        # acuse cove → cove padre
+        self.assertEqual(d_ed.edocument_id, self.edoc.id)
+        self.assertEqual(d_aced.edocument_id, self.edoc.id)   # acuse edoc → edoc padre
+        self.assertIsNone(d_pc.partida_id)
+        self.assertIsNone(d_pc.cove_id)
+        self.assertIsNone(d_pc.edocument_id)
+
+    def test_lectura_cove_edoc_por_fk(self):
+        self._doc(f"vu_COVE_{self.APP}_654001.xml", 8)
+        self._doc(f"vu_AC_COVE_{self.APP}_654001.pdf", 7)
+        self._doc(f"vu_COVE_{self.APP}_654001_REQUEST.xml", 19)  # request: excluido en lectura
+        self._doc(f"vu_ED_{self.APP}_EDOC001.pdf", 5)
+        cove_data = CoveSerializer(self.cove).data
+        edoc_data = EDocumentSerializer(self.edoc).data
+        self.assertEqual(len(cove_data["documentos"]), 2)  # cove + acuse, sin request
+        self.assertEqual(len(edoc_data["documentos"]), 1)
+
+    def test_backfill_liga_filas_existentes(self):
+        d_pt = self._doc(f"vu_PT_{self.APP}_3.xml", 1)
+        d_cv = self._doc(f"vu_COVE_{self.APP}_654001.xml", 8)
+        # Simular filas viejas sin ligar (save() las ligó; las desligamos en BD).
+        Document.objects.filter(id__in=[d_pt.id, d_cv.id]).update(
+            partida=None, cove=None, edocument=None
+        )
+        # dry-run no escribe
+        call_command("backfill_document_links", pedimento=str(self.pedimento.id),
+                     dry_run=True, stdout=StringIO())
+        d_pt.refresh_from_db()
+        self.assertIsNone(d_pt.partida_id)
+        # ejecución real liga
+        call_command("backfill_document_links", pedimento=str(self.pedimento.id), stdout=StringIO())
+        d_pt.refresh_from_db()
+        d_cv.refresh_from_db()
+        self.assertEqual(d_pt.partida_id, self.partida.id)
+        self.assertEqual(d_cv.cove_id, self.cove.id)
+        # idempotente: re-ejecutar no rompe ni cambia
+        call_command("backfill_document_links", pedimento=str(self.pedimento.id), stdout=StringIO())
+        d_pt.refresh_from_db()
+        self.assertEqual(d_pt.partida_id, self.partida.id)
+
+    def _tres_copias_edoc(self):
+        """3 copias del mismo edoc (type 5) con created_at d1<d2<d3 y archivos distintos."""
+        from django.utils import timezone
+        import datetime
+        d1 = self._doc(f"vu_ED_{self.APP}_EDOC001_aaa.pdf", 5)
+        d2 = self._doc(f"vu_ED_{self.APP}_EDOC001_bbb.pdf", 5)
+        d3 = self._doc(f"vu_ED_{self.APP}_EDOC001_ccc.pdf", 5)
+        base = timezone.now()
+        Document.objects.filter(id=d1.id).update(created_at=base - datetime.timedelta(days=2))
+        Document.objects.filter(id=d2.id).update(created_at=base - datetime.timedelta(days=1))
+        Document.objects.filter(id=d3.id).update(created_at=base)
+        return d1, d2, d3
+
+    def test_dedup_conserva_mas_reciente_y_es_idempotente(self):
+        from unittest.mock import patch
+        d1, d2, d3 = self._tres_copias_edoc()
+        self.assertEqual(Document.objects.filter(edocument=self.edoc, document_type_id=5).count(), 3)
+        with patch('api.customs.management.commands.dedup_documents.storage_service') as st:
+            st.file_exists.return_value = True
+            call_command('dedup_documents', pedimento=str(self.pedimento.id), stdout=StringIO())
+            self.assertEqual(st.delete_file.call_count, 2)  # borró los 2 viejos de MinIO
+        restantes = list(Document.objects.filter(edocument=self.edoc, document_type_id=5))
+        self.assertEqual(len(restantes), 1)
+        self.assertEqual(restantes[0].id, d3.id)  # conservó el más reciente
+        # idempotente: re-correr no borra nada
+        with patch('api.customs.management.commands.dedup_documents.storage_service') as st2:
+            st2.file_exists.return_value = True
+            call_command('dedup_documents', pedimento=str(self.pedimento.id), stdout=StringIO())
+            self.assertEqual(st2.delete_file.call_count, 0)
+
+    def test_dedup_dry_run_no_borra(self):
+        from unittest.mock import patch
+        self._tres_copias_edoc()
+        with patch('api.customs.management.commands.dedup_documents.storage_service') as st:
+            call_command('dedup_documents', pedimento=str(self.pedimento.id), dry_run=True, stdout=StringIO())
+            self.assertEqual(st.delete_file.call_count, 0)
+        self.assertEqual(Document.objects.filter(edocument=self.edoc, document_type_id=5).count(), 3)
+
+    def test_dedup_conserva_el_que_tiene_archivo_en_storage(self):
+        from unittest.mock import patch
+        d1, d2, d3 = self._tres_copias_edoc()  # d3 el más reciente
+        # En storage solo existe el de d1 (el más viejo); los más nuevos no.
+        with patch('api.customs.management.commands.dedup_documents.storage_service') as st:
+            st.file_exists.side_effect = lambda nombre: nombre.endswith('_aaa.pdf')
+            call_command('dedup_documents', pedimento=str(self.pedimento.id), stdout=StringIO())
+        restantes = list(Document.objects.filter(edocument=self.edoc, document_type_id=5))
+        self.assertEqual(len(restantes), 1)
+        self.assertEqual(restantes[0].id, d1.id)  # conservó el único con archivo válido
+
+    def test_backfill_legacy_liga_por_numero(self):
+        # Doc legado: app y prefijo viejos (vu_EDC, otro pedimento_app), pero el
+        # numero_edocument (EDOC001) SÍ está en el nombre.
+        legado = self._doc("vu_EDC_0201_800_3452_5000586_EDOC001_abc123.pdf", 5)
+        legado.refresh_from_db()
+        self.assertIsNone(legado.edocument_id)  # save() (match estricto) no lo ligó
+        # el backfill estricto tampoco (app/prefijo no coinciden)
+        call_command("backfill_document_links", pedimento=str(self.pedimento.id), stdout=StringIO())
+        legado.refresh_from_db()
+        self.assertIsNone(legado.edocument_id)
+        # el backfill LEGADO sí lo liga por número único
+        call_command("backfill_document_links_legacy", pedimento=str(self.pedimento.id), stdout=StringIO())
+        legado.refresh_from_db()
+        self.assertEqual(legado.edocument_id, self.edoc.id)

api/customs/views.py

@@ -60,6 +60,7 @@ from django.core.files.base import ContentFile
 from django.db import transaction
 from rest_framework.parsers import MultiPartParser, FormParser
 from api.record.models import Document, DocumentType, Fuente
+from django.db.models import Prefetch
 from unicodedata import normalize
 from datetime import datetime
 from django.utils import timezone
@@ -2332,14 +2333,21 @@ class PartidaViewSet(viewsets.ModelViewSet):
 
     def get_queryset(self):
         user = self.request.user
+        # Precarga los documentos de respuesta (tipo 1) de cada partida vía la FK
+        # real document.partida, para que get_documentos no genere N+1.
+        prefetch_docs = Prefetch(
+            'documents',
+            queryset=Document.objects.filter(document_type_id=1).select_related('pedimento', 'fuente'),
+            to_attr='documentos_vu',
+        )
         if is_internal_service_request(self.request):
-            return Partida.objects.all()
+            return Partida.objects.all().prefetch_related(prefetch_docs)
         if not user_has_permission(user, 'partidas.view'):
             return Partida.objects.none()
         org = get_org_context(user)
         if not org:
             return Partida.objects.none()
-        qs = Partida.objects.filter(pedimento__organizacion=org)
+        qs = Partida.objects.filter(pedimento__organizacion=org).prefetch_related(prefetch_docs)
         # Misma precedencia que los mixins de filtrado: superuser y roles
         # operativos ven todo lo de su org; is_importador no los degrada.
         if (
@@ -2596,7 +2604,14 @@ class ViewSetEDocument(LoggingMixin, viewsets.ModelViewSet, OrganizacionFiltrada
     def get_queryset(self):
         if not user_has_permission(self.request.user, 'edocuments.view'):
             return EDocument.objects.none()
-        return self.get_queryset_filtrado_por_organizacion()
+        # Precarga documentos del e-doc (incluye acuse; excluye REQUEST 21/25)
+        # vía la FK document.edocument, para que get_documentos no genere N+1.
+        prefetch_docs = Prefetch(
+            'documents',
+            queryset=Document.objects.exclude(document_type_id__in=[21, 25]).select_related('pedimento', 'fuente'),
+            to_attr='documentos_vu',
+        )
+        return self.get_queryset_filtrado_por_organizacion().prefetch_related(prefetch_docs)
 
     def perform_create(self, serializer):
         if is_internal_service_request(self.request):
@@ -2764,7 +2779,14 @@ class ViewSetCove(viewsets.ModelViewSet, OrganizacionFiltradaMixin):
     def get_queryset(self):
         if not user_has_permission(self.request.user, 'coves.view'):
             return Cove.objects.none()
-        return self.get_queryset_filtrado_por_organizacion()
+        # Precarga documentos del cove (incluye acuse cove; excluye REQUEST 19/23)
+        # vía la FK document.cove, para que get_documentos no genere N+1.
+        prefetch_docs = Prefetch(
+            'documents',
+            queryset=Document.objects.exclude(document_type_id__in=[19, 23]).select_related('pedimento', 'fuente'),
+            to_attr='documentos_vu',
+        )
+        return self.get_queryset_filtrado_por_organizacion().prefetch_related(prefetch_docs)
 
     def perform_create(self, serializer):
         if is_internal_service_request(self.request):

api/record/migrations/0004_document_subentidad_fk.py

@@ -0,0 +1,30 @@
+# Generated by Django 5.2.3 on 2026-06-24 13:37
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('customs', '0020_estados_descarga_t2026_05_027'),
+        ('record', '0003_document_vu'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='document',
+            name='cove',
+            field=models.ForeignKey(blank=True, db_index=False, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='documents', to='customs.cove'),
+        ),
+        migrations.AddField(
+            model_name='document',
+            name='edocument',
+            field=models.ForeignKey(blank=True, db_index=False, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='documents', to='customs.edocument'),
+        ),
+        migrations.AddField(
+            model_name='document',
+            name='partida',
+            field=models.ForeignKey(blank=True, db_index=False, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='documents', to='customs.partida'),
+        ),
+    ]

api/record/migrations/0005_document_subentidad_idx.py

@@ -0,0 +1,55 @@
+# Índices de las FK de sub-entidad en la tabla `document` (grande: ~5M filas en
+# prod) con CREATE INDEX CONCURRENTLY para no bloquear escrituras.
+#
+# CONCURRENTLY no corre dentro de transacción (atomic=False) y NO es transaccional:
+# si el proceso muere a mitad puede dejar un índice a medias. Por eso:
+#   - IF NOT EXISTS  → el reintento es idempotente (no choca con "already exists").
+#   - SeparateDatabaseAndState → el índice se refleja en el estado del modelo
+#     (AddIndex) sin que Django intente recrearlo, manteniendo el estado consistente.
+#
+# Recuperación si un build quedó INVALID (kill DURANTE la construcción, no después):
+#   DROP INDEX IF EXISTS "<nombre>";   y reintentar  python manage.py migrate record
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    atomic = False
+
+    dependencies = [
+        ('record', '0004_document_subentidad_fk'),
+    ]
+
+    operations = [
+        migrations.SeparateDatabaseAndState(
+            database_operations=[migrations.RunSQL(
+                sql='CREATE INDEX CONCURRENTLY IF NOT EXISTS "document_partida_idx" ON "document" ("partida_id");',
+                reverse_sql='DROP INDEX IF EXISTS "document_partida_idx";',
+            )],
+            state_operations=[migrations.AddIndex(
+                model_name='document',
+                index=models.Index(fields=['partida'], name='document_partida_idx'),
+            )],
+        ),
+        migrations.SeparateDatabaseAndState(
+            database_operations=[migrations.RunSQL(
+                sql='CREATE INDEX CONCURRENTLY IF NOT EXISTS "document_cove_idx" ON "document" ("cove_id");',
+                reverse_sql='DROP INDEX IF EXISTS "document_cove_idx";',
+            )],
+            state_operations=[migrations.AddIndex(
+                model_name='document',
+                index=models.Index(fields=['cove'], name='document_cove_idx'),
+            )],
+        ),
+        migrations.SeparateDatabaseAndState(
+            database_operations=[migrations.RunSQL(
+                sql='CREATE INDEX CONCURRENTLY IF NOT EXISTS "document_edocument_idx" ON "document" ("edocument_id");',
+                reverse_sql='DROP INDEX IF EXISTS "document_edocument_idx";',
+            )],
+            state_operations=[migrations.AddIndex(
+                model_name='document',
+                index=models.Index(fields=['edocument'], name='document_edocument_idx'),
+            )],
+        ),
+    ]

api/record/migrations/0006_analyze_document.py

@@ -0,0 +1,22 @@
+# Recalcula estadísticas del planner tras crear los índices de las FK (0005).
+# CREATE INDEX CONCURRENTLY NO corre ANALYZE: sin estadísticas frescas de las
+# columnas nuevas (partida_id/cove_id/edocument_id, casi todas NULL antes del
+# backfill), el planner puede elegir un seq scan sobre la tabla `document` (~5M
+# filas) para las consultas del prefetch en vez de usar el índice → endpoints
+# muy lentos. Este ANALYZE lo previene en cada entorno.
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('record', '0005_document_subentidad_idx'),
+    ]
+
+    operations = [
+        migrations.RunSQL(
+            sql='ANALYZE "document";',
+            reverse_sql=migrations.RunSQL.noop,
+        ),
+    ]

api/record/models.py

@@ -17,6 +17,14 @@ class Document(models.Model):
     fuente = models.ForeignKey('Fuente', on_delete=models.CASCADE, related_name='documents', blank=True, null=True)
     vu = models.BooleanField(default=False)
 
+    # Sub-entidad a la que pertenece el documento (None para docs nativos del
+    # pedimento: PC, remesa, subidas generales). Se puebla por nombre de archivo
+    # en save() vía core.document_links. db_index=False: el índice lo crea una
+    # migración aparte con CREATE INDEX CONCURRENTLY (tabla grande en prod).
+    partida = models.ForeignKey('customs.Partida', on_delete=models.CASCADE, related_name='documents', blank=True, null=True, db_index=False)
+    cove = models.ForeignKey('customs.Cove', on_delete=models.CASCADE, related_name='documents', blank=True, null=True, db_index=False)
+    edocument = models.ForeignKey('customs.EDocument', on_delete=models.CASCADE, related_name='documents', blank=True, null=True, db_index=False)
+
     created_at = models.DateTimeField(auto_now_add=True)
     updated_at = models.DateTimeField(auto_now=True)
 
@@ -30,6 +38,17 @@ class Document(models.Model):
         else:
             self.vu = False
 
+        # Ligar la sub-entidad (partida/cove/edocument) por nombre de archivo si
+        # aún no está ligada. Cubre todas las rutas de creación —incluida la
+        # ingesta del microservicio— sin tocar cada call site. Se ejecuta también
+        # en update porque el patrón común es create() sin archivo y luego
+        # asignar archivo + save(). Fuente única: core.document_links.
+        if self.archivo and not (self.partida_id or self.cove_id or self.edocument_id):
+            from core.document_links import resolver_fk
+            campo, inst = resolver_fk(self)
+            if inst is not None:
+                setattr(self, campo, inst)
+
         # Usar get_or_create en lugar de get para manejar el caso cuando no existe
         uso_almacenamiento, created = UsoAlmacenamiento.objects.get_or_create(
             organizacion=self.organizacion,
@@ -77,6 +96,14 @@ class Document(models.Model):
         verbose_name_plural = "Documents"
         db_table = 'document'
         ordering = ['created_at']
+        # Índices de las FK de sub-entidad. Se crean con CREATE INDEX CONCURRENTLY
+        # en una migración aparte (atomic=False); por eso los campos usan
+        # db_index=False y el índice se declara aquí.
+        indexes = [
+            models.Index(fields=['partida'], name='document_partida_idx'),
+            models.Index(fields=['cove'], name='document_cove_idx'),
+            models.Index(fields=['edocument'], name='document_edocument_idx'),
+        ]
 
 class DocumentType(models.Model):
     nombre = models.CharField(max_length=100, unique=True)

api/record/views.py

@@ -33,6 +33,7 @@ from core.permissions import (
     user_has_permission,
     IsInternalService,
 )
+from core.document_links import ids_documentos_entidad
 
 import logging
 logger = logging.getLogger(__name__)
@@ -718,14 +719,13 @@ class DocumentViewSet(viewsets.ModelViewSet, DocumentosFiltradosMixin):
                 status=status.HTTP_404_NOT_FOUND
             )
 
-        # Buscar documentos vu_PT_ asociados a cada partida por pedimento + numero_partida
+        # Buscar documentos asociados a cada partida por nombre de archivo con
+        # frontera real (core.partida_docs). Sin filtro de tipo: barre tanto la
+        # respuesta (1) como REQUEST/ERROR (17/18) de la partida.
+        # incluir_legacy=False: el borrado es destructivo, no se elimina por match difuso.
         doc_ids = []
         for partida in partidas:
-            docs = Document.objects.filter(
-                pedimento_id=partida.pedimento.id,
-                archivo__icontains=f'vu_pt_{partida.pedimento.pedimento_app}_{partida.numero_partida}_'
-            ).values_list('id', flat=True)
-            doc_ids.extend(docs)
+            doc_ids.extend(ids_documentos_entidad(partida, 'partida'))
 
         queryset = self.get_queryset()
         existing_documents = queryset.filter(id__in=doc_ids)
@@ -856,11 +856,7 @@ class DocumentViewSet(viewsets.ModelViewSet, DocumentosFiltradosMixin):
         # Buscar documentos que contengan el numero_cove en el nombre de archivo
         doc_ids = []
         for cove in coves:
-            docs = Document.objects.filter(
-                pedimento_id=cove.pedimento.id,
-                archivo__icontains=cove.numero_cove
-            ).values_list('id', flat=True)
-            doc_ids.extend(docs)
+            doc_ids.extend(ids_documentos_entidad(cove, 'cove'))
 
         queryset = self.get_queryset()
         existing_documents = queryset.filter(id__in=doc_ids)
@@ -989,11 +985,7 @@ class DocumentViewSet(viewsets.ModelViewSet, DocumentosFiltradosMixin):
         # Buscar documentos que contengan el numero_edocument en el nombre de archivo
         doc_ids = []
         for edoc in edocs:
-            docs = Document.objects.filter(
-                pedimento_id=edoc.pedimento.id,
-                archivo__icontains=edoc.numero_edocument
-            ).values_list('id', flat=True)
-            doc_ids.extend(docs)
+            doc_ids.extend(ids_documentos_entidad(edoc, 'edocument'))
 
         queryset = self.get_queryset()
         existing_documents = queryset.filter(id__in=doc_ids)
@@ -1900,6 +1892,10 @@ class DocumentViewSet(viewsets.ModelViewSet, DocumentosFiltradosMixin):
                                 document.delete()
                                 raise Exception(f"El archivo no se encuentra en storage tras guardarlo: {file.name}")
                             document.archivo = ruta
+                            # Ligar explícitamente la sub-entidad recién creada
+                            # (exacto, sin depender del matching por nombre).
+                            _campo_fk = {'partida': 'partida', 'cove': 'cove', 'edoc': 'edocument'}[tab_seccion]
+                            setattr(document, _campo_fk, expediente_obj)
                             document.save()
                         else:
                             document.delete()
@@ -1997,13 +1993,11 @@ class DocumentViewSet(viewsets.ModelViewSet, DocumentosFiltradosMixin):
         if not partidas.exists():
             return Response({"error": "No se encontraron partidas"}, status=status.HTTP_404_NOT_FOUND)
 
+        # Mismo matching por frontera que el borrado, pero incluir_legacy=True:
+        # la descarga no es destructiva, así que sí incluye archivos legacy.
         doc_ids = []
         for partida in partidas:
-            docs = Document.objects.filter(
-                pedimento_id=partida.pedimento.id,
-                archivo__icontains=f'vu_pt_{partida.pedimento.pedimento_app}_{partida.numero_partida}_'
-            ).values_list('id', flat=True)
-            doc_ids.extend(docs)
+            doc_ids.extend(ids_documentos_entidad(partida, 'partida'))
 
         queryset = self.get_queryset()
         docs_qs = queryset.filter(id__in=doc_ids)
@@ -2059,11 +2053,7 @@ class DocumentViewSet(viewsets.ModelViewSet, DocumentosFiltradosMixin):
 
         doc_ids = []
         for cove in coves:
-            docs = Document.objects.filter(
-                pedimento_id=cove.pedimento.id,
-                archivo__icontains=cove.numero_cove
-            ).values_list('id', flat=True)
-            doc_ids.extend(docs)
+            doc_ids.extend(ids_documentos_entidad(cove, 'cove'))
 
         queryset = self.get_queryset()
         docs_qs = queryset.filter(id__in=doc_ids)
@@ -2119,11 +2109,7 @@ class DocumentViewSet(viewsets.ModelViewSet, DocumentosFiltradosMixin):
 
         doc_ids = []
         for edoc in edocs:
-            docs = Document.objects.filter(
-                pedimento_id=edoc.pedimento.id,
-                archivo__icontains=edoc.numero_edocument
-            ).values_list('id', flat=True)
-            doc_ids.extend(docs)
+            doc_ids.extend(ids_documentos_entidad(edoc, 'edocument'))
 
         queryset = self.get_queryset()
         docs_qs = queryset.filter(id__in=doc_ids)

core/document_links.py

@@ -0,0 +1,132 @@
+"""
+Resolución de la sub-entidad (partida / cove / edocument) a la que pertenece un
+Document, para poblar sus FK reales (`document.partida`/`cove`/`edocument`).
+
+Fuente única, reusada por `Document.save()` (alta de filas nuevas) y por el comando
+`backfill_document_links` (filas existentes). Generaliza la frontera de
+[core.partida_docs] al resto de secciones.
+
+Discriminador de sección: el `document_type_id` (lo asigna el microservicio al
+generar el archivo). El nombre de archivo solo decide **cuál** registro dentro de
+la sección, vía frontera tras `vu_<sec>_{pedimento_app}_{numero}` — NO se extrae la
+llave partiendo por `_`, porque `pedimento_app` puede contener `_`
+(p.ej. `vu_AC_0101_230_1703_3004804_4.pdf`): se itera la entidad con su número exacto.
+
+Mapa tipo→sección (autoritativo, microservice/api/api_v2/modules/*/services.py):
+  Partida   : 1 (respuesta), 17 (request), 18 (error)
+  Cove      : 8 (respuesta), 7 (acuse), 19/20 (request/error), 23/24 (acuse request/error)
+  EDocument : 5 (respuesta), 4 (acuse), 21/22 (request/error), 25/26 (acuse request/error)
+  Nativos (sin FK): 2 (PC), 3 (remesa), 6, 13-16, y subidas sin tipo.
+"""
+
+import posixpath
+import re
+
+from core.partida_docs import es_doc_de_partida
+
+PARTIDA_TYPES = {1, 17, 18}
+COVE_TYPES = {7, 8, 19, 20, 23, 24}
+EDOCUMENT_TYPES = {4, 5, 21, 22, 25, 26}
+
+TYPE_TO_SECTION = {}
+for _t in PARTIDA_TYPES:
+    TYPE_TO_SECTION[_t] = 'partida'
+for _t in COVE_TYPES:
+    TYPE_TO_SECTION[_t] = 'cove'
+for _t in EDOCUMENT_TYPES:
+    TYPE_TO_SECTION[_t] = 'edocument'
+
+# Campo FK en Document y campo de llave de negocio en la entidad, por sección.
+SECCION_CAMPO = {'partida': 'partida', 'cove': 'cove', 'edocument': 'edocument'}
+SECCION_LLAVE = {'partida': 'numero_partida', 'cove': 'numero_cove', 'edocument': 'numero_edocument'}
+# related_name del FK a Pedimento en cada entidad (EDocument usa 'documentos').
+SECCION_RELACION = {'partida': 'partidas', 'cove': 'coves', 'edocument': 'documentos'}
+# Prefijos de nombre por sección (cove y edoc tienen variante de acuse).
+SECCION_PREFIJOS = {
+    'partida': ('vu_pt',),
+    'cove': ('vu_cove', 'vu_ac_cove'),
+    'edocument': ('vu_ed', 'vu_ac'),
+}
+
+
+def seccion_de_tipo(document_type_id):
+    """Sección a la que pertenece un document_type_id, o None si es nativo."""
+    if document_type_id is None:
+        return None
+    return TYPE_TO_SECTION.get(int(document_type_id))
+
+
+def _coincide_prefijo(base, prefijo):
+    """True si `base` empieza con `prefijo` seguido de frontera (`_`, `.` o fin)."""
+    return base.startswith(prefijo) and (len(base) == len(prefijo) or base[len(prefijo)] in "_.")
+
+
+def coincide(nombre_archivo, seccion, pedimento_app, numero):
+    """Indica si `nombre_archivo` corresponde a (seccion, pedimento_app, numero)."""
+    if seccion == 'partida':
+        # Reusa el matcher de partidas (incluye formato legacy).
+        return es_doc_de_partida(nombre_archivo, pedimento_app, numero)
+    base = posixpath.basename(nombre_archivo or "").lower()
+    app = str(pedimento_app).strip().lower()
+    num = str(numero).strip().lower()
+    return any(
+        _coincide_prefijo(base, f"{pref}_{app}_{num}")
+        for pref in SECCION_PREFIJOS[seccion]
+    )
+
+
+def numero_en_nombre(nombre_archivo, numero):
+    """True si `numero` aparece como token completo en el basename (con frontera
+    `_`/`.`/inicio/fin), SIN exigir prefijo ni pedimento_app.
+
+    Para documentos LEGADOS cuyo nombre trae otro app/prefijo (p.ej.
+    `vu_EDC_0201_800_..._04382515ZIFF5_hex.pdf`) pero cuyo número de cove/edoc
+    (único y largo) sí está en el nombre. NO usar con partida (enteros cortos →
+    colisiones).
+    """
+    base = posixpath.basename(nombre_archivo or "").lower()
+    num = re.escape(str(numero).strip().lower())
+    if not num:
+        return False
+    return re.search(rf"(?:^|_){num}(?:_|\.|$)", base) is not None
+
+
+def match_entidad(nombre_archivo, seccion, pedimento_app, entidades):
+    """Devuelve la entidad de `entidades` cuyo número coincide con el archivo, o None.
+
+    `entidades` es un iterable de instancias del modelo de la sección (Partida /
+    Cove / EDocument). Pensado para uso en memoria (backfill con listas precargadas).
+    """
+    llave = SECCION_LLAVE[seccion]
+    for ent in entidades:
+        if coincide(nombre_archivo, seccion, pedimento_app, getattr(ent, llave)):
+            return ent
+    return None
+
+
+def resolver_fk(document):
+    """Resuelve la FK de sub-entidad de un Document → (campo, instancia | None).
+
+    `campo` es 'partida' | 'cove' | 'edocument' (el atributo a setear), o (None, None)
+    si el documento es nativo de pedimento o no tiene archivo. Hace las consultas
+    necesarias; para lotes usar `match_entidad` con entidades precargadas.
+    """
+    seccion = seccion_de_tipo(getattr(document, 'document_type_id', None))
+    if not seccion or not document.archivo:
+        return (None, None)
+    pedimento = document.pedimento
+    entidades = getattr(pedimento, SECCION_RELACION[seccion]).all()
+    inst = match_entidad(document.archivo.name, seccion, pedimento.pedimento_app, entidades)
+    return (SECCION_CAMPO[seccion], inst)
+
+
+# --------------------------------------------------------------------------- #
+# Lectura / borrado / descarga: SIEMPRE por la FK real (no por nombre). El nombre
+# solo se usa para ESTABLECER la FK (Document.save() en altas, backfill en filas
+# viejas). Requiere que el backfill esté completo para datos previos.
+# --------------------------------------------------------------------------- #
+
+def ids_documentos_entidad(entidad, seccion):
+    """IDs de los documentos de la entidad (borrado/descarga) por la FK real."""
+    from api.record.models import Document  # import diferido: evita ciclo con record.models
+    return list(Document.objects.filter(**{SECCION_CAMPO[seccion]: entidad}).values_list('id', flat=True))

core/partida_docs.py

@@ -0,0 +1,65 @@
+"""
+Fuente única de la convención de nombres documento → partida.
+
+Las tablas `document` y `partida` no tienen FK entre sí; solo comparten
+`pedimento_id`. La pertenencia de un documento a una partida se infiere del
+nombre del archivo, que tiene tres formatos para los documentos de partida
+(`document_type_id = 1`):
+
+  1. vu_PT_{pedimento_app}_{numero}            (sin extensión — formato original)
+  2. vu_PT_{pedimento_app}_{numero}.xml
+  3. vu_PT_{pedimento_app}_{numero}_{hex}.xml  (sufijo de unicidad del storage)
+
+REQUEST / ERROR son vu_PT_{app}_{numero}_REQUEST.xml / _ERROR.xml (tipos 17/18).
+Formato legacy: vu_PT_..._{numero}.xml (el número de partida queda al final).
+
+El matching debe usar una FRONTERA real tras el número (`_`, `.` o fin de
+cadena) para cubrir los tres formatos sin confundir la partida 1 con la 11/100.
+Aquí viven las dos implementaciones equivalentes de esa regla:
+
+  - patron_regex_partida : para filtros en BD (Document.archivo__iregex).
+  - es_doc_de_partida    : para asignación en memoria (basename).
+"""
+
+import posixpath
+import re
+
+
+def patron_regex_partida(pedimento_app, numero_partida, incluir_legacy=False):
+    """Construye el patrón para `Document.objects.filter(archivo__iregex=...)`.
+
+    PostgreSQL aplica el patrón con `~*` (no anclado), por lo que el prefijo de
+    ruta `documents/` es irrelevante. La frontera `(_|\\.|$)` tras el número
+    cubre los tres formatos y rechaza la colisión de prefijo (la partida 1 no
+    matchea con 11/100 porque tras el `1` vendría un dígito, no la frontera).
+
+    El filtro `document_type_id` se decide en el call site: el patrón también
+    matchea `_REQUEST`/`_ERROR`, así que quien quiera excluirlos debe filtrar
+    por tipo. `pedimento_app` se escapa siempre (es un CharField sin validación).
+
+    incluir_legacy: agrega el formato viejo con el número al final. Es un match
+    difuso (`.+`), úsalo solo en lectura, nunca en operaciones destructivas.
+    """
+    app = re.escape(str(pedimento_app).strip())
+    num = re.escape(str(numero_partida).strip())
+    patron = rf"vu_PT_{app}_{num}(_|\.|$)"
+    if incluir_legacy:
+        patron = rf"({patron}|vu_PT_.+_{num}\.xml$)"
+    return patron
+
+
+def es_doc_de_partida(nombre_archivo, pedimento_app, numero_partida, incluir_legacy=True):
+    """Indica si `nombre_archivo` pertenece a la partida, evaluado en memoria.
+
+    Equivalente a `patron_regex_partida` pero sobre el basename en minúsculas.
+    Fuente única para la lectura (PartidaSerializer) y para el comando
+    fix_partidas_error. La frontera exige que tras `vu_pt_{app}_{numero}` venga
+    `_`, `.` o el fin de la cadena (formato #1 sin extensión).
+    """
+    base = posixpath.basename(nombre_archivo or "").lower()
+    prefijo = f"vu_pt_{pedimento_app}_{numero_partida}".lower()
+    if base.startswith(prefijo) and (len(base) == len(prefijo) or base[len(prefijo)] in "_."):
+        return True
+    if incluir_legacy:
+        return re.match(rf"^vu_pt_.+_{re.escape(str(numero_partida))}\.xml$", base) is not None
+    return False