feat: FK polimorfica Document -> {partida, cove, edocument} + backfill (T2025-09-004)

Reemplaza el matching fragil por nombre de archivo con FK reales:
- 3 FK nullables (CASCADE) en Document; resolucion central en save() por
  document_type + nombre (core.document_links), cubre toda ruta de creacion
  incluida la ingesta del microservicio; set explicito en create_vu_record.
- Comando backfill_document_links (idempotente, dry-run) para filas existentes.
- Lectura/descarga/borrado SIEMPRE por la FK (id); el nombre solo ESTABLECE la
  FK en save()/backfill. Prefetch con select_related(pedimento, fuente) sin N+1.
- Migraciones: 0004 (campos), 0005 (indices CONCURRENTLY IF NOT EXISTS, idempotente
  via SeparateDatabaseAndState), 0006 (ANALYZE document para estadisticas del planner).

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

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
+                )