fix: filtrado de partidas por nomenclatura de documento (core/partida_docs)
Frontera (_|.|$) tras vu_PT_{app}_{numero} para cubrir los 3 formatos sin
confundir partida 1 con 11/100. Fuente unica en core/partida_docs.py, reusada
por get_documentos, handlers de borrado/descarga y fix_partidas_error.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -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())
|
||||
|
||||
|
||||
@@ -14,6 +14,10 @@ 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
|
||||
from core.partida_docs import es_doc_de_partida
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class PedimentoSerializer(serializers.ModelSerializer):
|
||||
documentos_count = serializers.SerializerMethodField()
|
||||
@@ -48,31 +52,34 @@ class PartidaSerializer(serializers.ModelSerializer):
|
||||
documentos = serializers.SerializerMethodField()
|
||||
|
||||
def get_documentos(self, obj):
|
||||
if not obj or not getattr(obj, 'pedimento', None):
|
||||
return []
|
||||
if not obj or not getattr(obj, 'numero_partida', None):
|
||||
if not obj or not getattr(obj, 'pedimento', None) 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}_"
|
||||
# El matching documento→partida se hace por nombre de archivo con
|
||||
# frontera real (core.partida_docs); document_type_id=1 son los
|
||||
# documentos de respuesta de partida (excluye REQUEST/ERROR 17/18).
|
||||
mapa = self.context.get('docs_por_partida')
|
||||
if mapa is not None:
|
||||
# Camino optimizado: la vista precargó el mapa de la página.
|
||||
docs = mapa.get((obj.pedimento_id, obj.numero_partida), [])
|
||||
else:
|
||||
# Fallback (retrieve u otros callers): una consulta por partida.
|
||||
qs = Document.objects.filter(
|
||||
pedimento=obj.pedimento,
|
||||
document_type_id=1,
|
||||
).select_related('pedimento') # evita N+1 en DocumentSerializer.get_pedimento_numero
|
||||
app = obj.pedimento.pedimento_app
|
||||
docs = [d for d in qs if es_doc_de_partida(d.archivo.name, app, obj.numero_partida)]
|
||||
|
||||
# 17 = REQUEST partida, 18 = ERROR partida
|
||||
qs = Document.objects.filter(
|
||||
pedimento=obj.pedimento,
|
||||
archivo__icontains=patron,
|
||||
).exclude(document_type_id__in=[17, 18])
|
||||
org_id = getattr(obj, 'organizacion_id', None)
|
||||
if org_id:
|
||||
docs = [d for d in docs if d.organizacion_id == org_id]
|
||||
|
||||
if hasattr(obj, 'organizacion') and obj.organizacion:
|
||||
qs = qs.filter(organizacion=obj.organizacion)
|
||||
return DocumentSerializer(docs, many=True, context=self.context).data
|
||||
|
||||
serializer = DocumentSerializer(qs, many=True, context=self.context)
|
||||
return serializer.data
|
||||
|
||||
except Exception:
|
||||
except Exception as e:
|
||||
logger.warning("get_documentos partida %s: %s", getattr(obj, 'id', '?'), e)
|
||||
return []
|
||||
class Meta:
|
||||
model = Partida
|
||||
|
||||
@@ -236,7 +236,12 @@ 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
|
||||
from api.customs.views import PartidaViewSet
|
||||
from api.customs.models import Partida
|
||||
from api.record.models import Document, DocumentType
|
||||
|
||||
|
||||
XML_RESPUESTA_VALIDA = (
|
||||
@@ -496,3 +501,161 @@ 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_mismo_resultado_con_y_sin_prefetch(self):
|
||||
self._doc(f"vu_PT_{self.APP}_1.xml")
|
||||
self._doc(f"vu_PT_{self.APP}_11.xml")
|
||||
|
||||
vs = PartidaViewSet()
|
||||
mapa = vs._mapa_docs_partida([self.p1, self.p11])
|
||||
ids_p1 = {d.id for d in mapa[(self.pedimento.id, 1)]}
|
||||
ids_p11 = {d.id for d in mapa[(self.pedimento.id, 11)]}
|
||||
self.assertEqual(len(ids_p1), 1)
|
||||
self.assertEqual(len(ids_p11), 1)
|
||||
self.assertTrue(ids_p1.isdisjoint(ids_p11))
|
||||
|
||||
sin_ctx = PartidaSerializer(self.p1).data
|
||||
con_ctx = PartidaSerializer(self.p1, context={"docs_por_partida": mapa}).data
|
||||
self.assertEqual(len(sin_ctx["documentos"]), 1)
|
||||
self.assertEqual(len(con_ctx["documentos"]), 1)
|
||||
self.assertIn(f"vu_PT_{self.APP}_1.xml", self._blob(con_ctx))
|
||||
self.assertNotIn("_11", self._blob(con_ctx))
|
||||
|
||||
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_mapa_docs_partida_es_una_sola_consulta(self):
|
||||
# documentos para varias partidas del mismo pedimento
|
||||
self._doc(f"vu_PT_{self.APP}_1.xml")
|
||||
self._doc(f"vu_PT_{self.APP}_11.xml")
|
||||
partidas = list(
|
||||
Partida.objects.filter(pedimento=self.pedimento).select_related("pedimento")
|
||||
)
|
||||
vs = PartidaViewSet()
|
||||
# Una sola consulta sin importar cuántas partidas (evita el N+1).
|
||||
with self.assertNumQueries(1):
|
||||
vs._mapa_docs_partida(partidas)
|
||||
|
||||
@@ -60,6 +60,8 @@ 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 core.partida_docs import es_doc_de_partida
|
||||
from collections import defaultdict
|
||||
from unicodedata import normalize
|
||||
from datetime import datetime
|
||||
from django.utils import timezone
|
||||
@@ -2354,6 +2356,41 @@ class PartidaViewSet(viewsets.ModelViewSet):
|
||||
return qs.filter(pedimento__contribuyente__in=user.rfc.all())
|
||||
return Partida.objects.none()
|
||||
|
||||
def list(self, request, *args, **kwargs):
|
||||
# Precarga los documentos de la página en una sola consulta para evitar
|
||||
# el N+1 de get_documentos (una consulta regex por cada partida).
|
||||
queryset = self.filter_queryset(self.get_queryset()).select_related('pedimento')
|
||||
page = self.paginate_queryset(queryset)
|
||||
objetos = page if page is not None else list(queryset)
|
||||
ctx = self.get_serializer_context()
|
||||
ctx['docs_por_partida'] = self._mapa_docs_partida(objetos)
|
||||
serializer = self.get_serializer(objetos, many=True, context=ctx)
|
||||
if page is not None:
|
||||
return self.get_paginated_response(serializer.data)
|
||||
return Response(serializer.data)
|
||||
|
||||
def _mapa_docs_partida(self, partidas):
|
||||
"""Asigna los documentos de respuesta (tipo 1) de los pedimentos de la
|
||||
página a cada partida por nombre de archivo, en memoria.
|
||||
Devuelve {(pedimento_id, numero_partida): [Document, ...]}."""
|
||||
ped_ids = {p.pedimento_id for p in partidas}
|
||||
if not ped_ids:
|
||||
return {}
|
||||
docs_por_ped = defaultdict(list)
|
||||
qs = Document.objects.filter(
|
||||
pedimento_id__in=ped_ids, document_type_id=1,
|
||||
).select_related('pedimento')
|
||||
for d in qs:
|
||||
docs_por_ped[d.pedimento_id].append(d)
|
||||
mapa = {}
|
||||
for p in partidas:
|
||||
app = p.pedimento.pedimento_app
|
||||
mapa[(p.pedimento_id, p.numero_partida)] = [
|
||||
d for d in docs_por_ped.get(p.pedimento_id, [])
|
||||
if es_doc_de_partida(d.archivo.name, app, p.numero_partida)
|
||||
]
|
||||
return mapa
|
||||
|
||||
def perform_create(self, serializer):
|
||||
if is_internal_service_request(self.request):
|
||||
serializer.save()
|
||||
|
||||
@@ -33,6 +33,7 @@ from core.permissions import (
|
||||
user_has_permission,
|
||||
IsInternalService,
|
||||
)
|
||||
from core.partida_docs import patron_regex_partida
|
||||
|
||||
import logging
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -718,12 +719,18 @@ 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}_'
|
||||
pedimento_id=partida.pedimento_id,
|
||||
archivo__iregex=patron_regex_partida(
|
||||
partida.pedimento.pedimento_app, partida.numero_partida,
|
||||
incluir_legacy=False,
|
||||
),
|
||||
).values_list('id', flat=True)
|
||||
doc_ids.extend(docs)
|
||||
|
||||
@@ -1997,11 +2004,16 @@ 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}_'
|
||||
pedimento_id=partida.pedimento_id,
|
||||
archivo__iregex=patron_regex_partida(
|
||||
partida.pedimento.pedimento_app, partida.numero_partida,
|
||||
incluir_legacy=True,
|
||||
),
|
||||
).values_list('id', flat=True)
|
||||
doc_ids.extend(docs)
|
||||
|
||||
|
||||
65
core/partida_docs.py
Normal file
65
core/partida_docs.py
Normal file
@@ -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
|
||||
Reference in New Issue
Block a user