EFC/backend
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
backend/api/customs/management/commands/fix_partidas_error.py

542 lines
24 KiB
Python
Raw Permalink Blame History

"""
Diagnóstico y corrección de partidas con descargado=True que NO tienen un XML
de respuesta de partida válido.
Una partida cuenta como realmente descargada solo si alguno de sus documentos
contiene el nodo <consultarPartidaRespuesta> sin <tieneError>true</tieneError>.
Clasificación por contenido de cada documento candidato (excluye types 17/18,
que ya están identificados como REQUEST/ERROR):
- valida : consultarPartidaRespuesta sin tieneError=true
- error : tieneError=true → renombra a _ERROR, type 18
- request : consultarPartidaPeticion → renombra a _REQUEST, type 17
(eco de la petición guardado como si fuera respuesta)
- desconocido : contenido no identificable → solo reporte
- ausente : registro en BD cuyo archivo no existe en storage
- no_verificable : storage inaccesible (excepción al consultar/leer)
Veredicto por partida con descargado=True:
- ≥1 valida → conserva descargado=True
- 0 validas y ≥1 no_verificable → sin cambios (storage inaccesible)
- 0 validas, ≥1 ausente y NINGÚN archivo del pedimento existe en storage
→ sin cambios (canario: probablemente se
está corriendo contra un storage que no
es el de esta BD, p. ej. dev)
- en cualquier otro caso → descargado=False (incluye partidas que
solo tienen el REQUEST, ningún doc, o
registros fantasma con el storage real)
Canario de storage: si al menos un archivo vu_PT_ del pedimento (REQUEST,
ERROR o respuesta) sí existe en storage, el storage es el correcto y los
documentos ausentes son registros fantasma reales (BD sin archivo).
Convenciones de nomenclatura del microservicio:
- REQUEST (type 17): vu_PT_{pedimento_app}_{partida}_REQUEST.xml
- ERROR (type 18): vu_PT_{pedimento_app}_{partida}_ERROR.xml
- Éxito (type 1): vu_PT_{pedimento_app}_{partida}.xml
(el storage puede agregar sufijos de unicidad: vu_PT_{...}_{partida}_Ab12xQ.xml)
- Legacy : vu_PT_..._{partida}.xml (número de partida al final)
Uso:
python manage.py fix_partidas_error --pedimento <UUID> --dry-run
python manage.py fix_partidas_error --organizacion <UUID> --dry-run
python manage.py fix_partidas_error --organizacion <UUID>
python manage.py fix_partidas_error --solo-malformados --dry-run
python manage.py fix_partidas_error --dry-run # todas las orgs
"""
import io
import posixpath
import re
from django.core.management.base import BaseCommand, CommandError
from django.db import transaction
from django.db.models import Q
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
_PT_REQUEST = 17
_PT_ERROR = 18
# Clasificaciones por contenido del XML
_VALIDA = "valida"
_ERROR_VU = "error"
_REQUEST_ECO = "request"
_DESCONOCIDO = "desconocido"
_AUSENTE = "ausente"
_NO_VERIFICABLE = "no_verificable"
# clase → (sufijo de archivo, document_type destino)
_RECLASIFICACION = {
_ERROR_VU: ("ERROR", _PT_ERROR),
_REQUEST_ECO: ("REQUEST", _PT_REQUEST),
}
class Command(BaseCommand):
help = "Corrige partidas descargado=True sin XML de respuesta de partida válido."
def add_arguments(self, parser):
parser.add_argument(
"--organizacion", metavar="UUID",
help="UUID de la organización. Sin este arg: todas las orgs.",
)
parser.add_argument(
"--pedimento", metavar="UUID",
help="UUID del pedimento a diagnosticar/corregir.",
)
# Filtros de fecha (aplican sobre fecha_pago del pedimento)
parser.add_argument(
"--fecha-desde", metavar="YYYY-MM-DD",
help="Procesar pedimentos con fecha_pago >= esta fecha.",
)
parser.add_argument(
"--fecha-hasta", metavar="YYYY-MM-DD",
help="Procesar pedimentos con fecha_pago <= esta fecha.",
)
parser.add_argument(
"--solo-malformados", action="store_true",
help="Limitar a pedimentos con aduana/patente/pedimento/numero_operacion inválidos (comportamiento anterior).",
)
# Control de lote
parser.add_argument(
"--offset", type=int, default=0,
help="Saltar los primeros N pedimentos (default: 0).",
)
parser.add_argument(
"--limit", type=int, default=0,
help="Procesar máximo N pedimentos (default: 0 = todos).",
)
parser.add_argument(
"--dry-run", action="store_true",
help="Solo diagnóstico, sin aplicar cambios.",
)
# ------------------------------------------------------------------ #
# Entry point
# ------------------------------------------------------------------ #
def handle(self, *args, **options):
org_id = options.get("organizacion")
ped_id = options.get("pedimento")
fecha_desde = options.get("fecha_desde")
fecha_hasta = options.get("fecha_hasta")
offset = options["offset"]
limit = options["limit"]
dry_run = options["dry_run"]
if dry_run:
self.stdout.write(self.style.WARNING(
"=== MODO PRUEBA (--dry-run): Sin cambios en BD ni storage ===\n"
))
if ped_id:
self._handle_single(ped_id, dry_run)
return
# Universo: pedimentos con al menos una partida descargado=True
ped_ids = Partida.objects.filter(descargado=True).values_list(
"pedimento_id", flat=True
).distinct()
base_qs = self._malformed_qs() if options["solo_malformados"] else Pedimento.objects.all()
ped_qs = base_qs.filter(id__in=ped_ids)
if org_id:
ped_qs = ped_qs.filter(organizacion_id=org_id)
if fecha_desde:
ped_qs = ped_qs.filter(fecha_pago__gte=fecha_desde)
if fecha_hasta:
ped_qs = ped_qs.filter(fecha_pago__lte=fecha_hasta)
ped_qs = ped_qs.select_related("organizacion").order_by("fecha_pago", "pedimento_app")
total_sin_filtro = ped_qs.count()
if offset:
ped_qs = ped_qs[offset:]
if limit:
ped_qs = ped_qs[:limit]
total = total_sin_filtro if not (offset or limit) else min(
limit or total_sin_filtro, max(0, total_sin_filtro - offset)
)
self.stdout.write(
f"Pedimentos con partidas descargadas (total): {total_sin_filtro}\n"
f"Procesando este lote : {total}"
+ (f" [offset={offset}]" if offset else "")
+ (f" [limit={limit}]" if limit else "")
+ (f" [solo malformados]" if options["solo_malformados"] else "")
+ "\n"
)
if total == 0:
self.stdout.write(self.style.SUCCESS("Nada que revisar en este lote."))
return
stats = self._stats_vacios()
n_peds = 0
for ped in ped_qs:
parciales = self._process_pedimento(ped, dry_run)
n_peds += 1
for k in stats:
stats[k] += parciales[k]
self._print_summary(n_peds, stats, dry_run)
# ------------------------------------------------------------------ #
# Flujo --pedimento
# ------------------------------------------------------------------ #
def _handle_single(self, ped_id, dry_run):
try:
ped = Pedimento.objects.get(id=ped_id)
except Pedimento.DoesNotExist:
raise CommandError(f"Pedimento {ped_id!r} no encontrado.")
# Diagnóstico de campos: informativo, ya no excluye pedimentos válidos
self._print_ped_diagnosis(ped, self._field_checks(ped))
stats = self._process_pedimento(ped, dry_run)
self._print_summary(1, stats, dry_run)
# ------------------------------------------------------------------ #
# Queryset de pedimentos malformados
# ------------------------------------------------------------------ #
def _malformed_qs(self):
return Pedimento.objects.annotate(
aduana_len=Length("aduana"),
patente_len=Length("patente"),
pedimento_len=Length("pedimento"),
).filter(
Q(aduana__isnull=True) | Q(aduana="") | Q(aduana_len__lt=3)
| Q(numero_operacion__isnull=True) | Q(numero_operacion="")
| Q(patente__isnull=True) | Q(patente="") | Q(patente_len__lt=4)
| Q(pedimento__isnull=True) | Q(pedimento="") | Q(pedimento_len__lt=7)
)
# ------------------------------------------------------------------ #
# Diagnóstico de un pedimento
# ------------------------------------------------------------------ #
def _field_checks(self, ped):
return {
"aduana (debe tener 3 dígitos)": not ped.aduana or len(ped.aduana.strip()) < 3,
"numero_operacion (obligatorio)": not ped.numero_operacion or not ped.numero_operacion.strip(),
"patente (debe tener 4 dígitos)": not ped.patente or len(ped.patente.strip()) < 4,
"pedimento_fld (debe tener 7 dígitos)": not ped.pedimento or len(ped.pedimento.strip()) < 7,
}
def _print_ped_diagnosis(self, ped, checks):
es_malo = any(checks.values())
estado = self.style.ERROR("MALFORMADO") if es_malo else self.style.SUCCESS("VÁLIDO")
self.stdout.write(
f"Pedimento {ped.pedimento_app} (id={ped.id}) → {estado}\n"
f" aduana = {ped.aduana!r} (len={len(ped.aduana or '')})\n"
f" patente = {ped.patente!r} (len={len(ped.patente or '')})\n"
f" numero_op = {ped.numero_operacion!r}\n"
f" pedimento_fld = {ped.pedimento!r} (len={len(ped.pedimento or '')})\n"
)
for campo, malo in checks.items():
marca = self.style.ERROR("✗") if malo else self.style.SUCCESS("✓")
self.stdout.write(f" {marca} {campo}")
self.stdout.write("")
# ------------------------------------------------------------------ #
# Procesamiento de un pedimento
# ------------------------------------------------------------------ #
def _stats_vacios(self):
return {
"partidas": 0, # partidas descargado=True revisadas
"corregidas": 0, # partidas marcadas descargado=False
"bloqueadas": 0, # partidas sin cambios (storage inaccesible/equivocado)
"docs_error": 0, # docs renombrados a _ERROR (type 18)
"docs_request": 0, # docs reclasificados a _REQUEST (type 17)
"desconocidos": 0, # docs con contenido no identificable
"fantasmas": 0, # registros en BD sin archivo en storage (no se borran)
}
def _process_pedimento(self, ped, dry_run):
es_malformado = any(self._field_checks(ped).values())
self.stdout.write(
f"Pedimento: {ped.pedimento_app} | "
f"aduana={ped.aduana!r} patente={ped.patente!r} num_op={ped.numero_operacion!r}"
+ (" [MALFORMADO]" if es_malformado else "")
)
stats = self._stats_vacios()
partidas = Partida.objects.filter(pedimento=ped, descargado=True)
n_partidas = partidas.count()
if n_partidas == 0:
self.stdout.write(" → Sin partidas con descargado=True\n")
return stats
self.stdout.write(f" Partidas con descargado=True: {n_partidas}")
# Una sola consulta por pedimento; la asignación por partida es en memoria
docs_pedimento = list(
Document.objects.filter(pedimento=ped, archivo__icontains="vu_PT_")
)
# Canario perezoso: ¿existe en storage al menos un archivo del pedimento?
# Distingue "registro fantasma con storage real" de "storage equivocado".
canario = {"valor": None}
def storage_es_correcto():
if canario["valor"] is None:
canario["valor"] = self._storage_tiene_archivos(docs_pedimento)
return canario["valor"]
for partida in partidas:
self._process_partida(ped, partida, docs_pedimento, storage_es_correcto, dry_run, stats)
self.stdout.write("")
return stats
def _storage_tiene_archivos(self, docs):
"""True si al menos un archivo vu_PT_ del pedimento existe en storage."""
for doc in docs:
try:
if minio_client.file_exists(doc.archivo.name):
return True
except Exception:
return False # storage inaccesible: modo conservador
return False
# ------------------------------------------------------------------ #
# Procesamiento de una partida
# ------------------------------------------------------------------ #
def _process_partida(self, ped, partida, docs_pedimento, storage_es_correcto, dry_run, stats):
stats["partidas"] += 1
docs = self._docs_de_partida(docs_pedimento, ped.pedimento_app, partida.numero_partida)
candidatos = [d for d in docs if d.document_type_id not in (_PT_REQUEST, _PT_ERROR)]
n_requests = sum(1 for d in docs if d.document_type_id == _PT_REQUEST)
n_errores = sum(1 for d in docs if d.document_type_id == _PT_ERROR)
self.stdout.write(
f"\n Partida {partida.numero_partida}: {len(candidatos)} doc(s) de respuesta a revisar"
f" (REQUEST: {n_requests}, ERROR previos: {n_errores})"
)
clasificados = []
for doc in candidatos:
clase, motivo = self._classify_document(doc)
iconos = {
_VALIDA: self.style.SUCCESS("✓ partida válida"),
_ERROR_VU: self.style.ERROR("✗ ERROR VUCEM"),
_REQUEST_ECO: self.style.WARNING("↺ es REQUEST, no respuesta"),
_DESCONOCIDO: self.style.WARNING("? contenido desconocido"),
_AUSENTE: self.style.WARNING("✗ registro sin archivo en storage"),
_NO_VERIFICABLE: self.style.WARNING("⚠ storage inaccesible"),
}
self.stdout.write(f" [{iconos[clase]}] type={doc.document_type_id} | {doc.archivo.name}")
if motivo:
self.stdout.write(f" motivo: {motivo}")
clasificados.append((doc, clase))
validas = [d for d, c in clasificados if c == _VALIDA]
no_verificables = [d for d, c in clasificados if c == _NO_VERIFICABLE]
ausentes = [d for d, c in clasificados if c == _AUSENTE]
corregibles = [(d, c) for d, c in clasificados if c in _RECLASIFICACION]
stats["desconocidos"] += sum(1 for _, c in clasificados if c == _DESCONOCIDO)
# Veredicto: solo una consultarPartidaRespuesta sin error mantiene la
# partida como descargada. Storage inaccesible bloquea el cambio; un
# archivo ausente solo bloquea cuando NINGÚN archivo del pedimento
# existe en storage (canario: posible storage equivocado, p. ej. dev).
if validas:
marcar_no_descargada = False
veredicto = self.style.SUCCESS("OK: tiene respuesta de partida válida")
elif no_verificables:
marcar_no_descargada = False
stats["bloqueadas"] += 1
veredicto = self.style.WARNING(
"SIN CAMBIOS: storage inaccesible — ejecutar donde el storage sea accesible"
)
elif ausentes and not storage_es_correcto():
marcar_no_descargada = False
stats["bloqueadas"] += 1
veredicto = self.style.WARNING(
"SIN CAMBIOS: ningún archivo del pedimento existe en storage — "
"¿se está corriendo contra el storage correcto?"
)
else:
marcar_no_descargada = True
stats["corregidas"] += 1
stats["fantasmas"] += len(ausentes)
veredicto = self.style.ERROR("descargado → False (sin XML de partida válido)")
self.stdout.write(f" Veredicto: {veredicto}")
for _, clase in corregibles:
clave = "docs_error" if clase == _ERROR_VU else "docs_request"
stats[clave] += 1
if not dry_run and (corregibles or marcar_no_descargada):
self._apply_fix(partida, corregibles, marcar_no_descargada, ped.pedimento_app)
# ------------------------------------------------------------------ #
# Asignación de documentos a una partida por nombre de archivo
# ------------------------------------------------------------------ #
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)
"""
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):
asignados[doc.id] = doc
return list(asignados.values())
# ------------------------------------------------------------------ #
# Clasificación del contenido XML
# ------------------------------------------------------------------ #
def _classify_document(self, doc):
"""
Lee el XML desde MinIO y clasifica su contenido.
Retorna (clase, motivo: str | None).
"""
name = doc.archivo.name
try:
if not minio_client.file_exists(name):
return _AUSENTE, "archivo no encontrado en storage"
response = minio_client._client.get_object(minio_client._bucket_name, name)
try:
content = response.read()
finally:
response.close()
response.release_conn()
text = content.decode("utf-8", errors="replace").lower()
except Exception as e:
return _NO_VERIFICABLE, f"excepción al leer archivo: {e}"
if "tieneerror>true<" in text:
return _ERROR_VU, "tieneError=true detectado en XML"
if "consultarpartidarespuesta" in text:
return _VALIDA, None
if "consultarpartidapeticion" in text:
return _REQUEST_ECO, "es la petición SOAP, no la respuesta"
return _DESCONOCIDO, "sin consultarPartidaRespuesta, sin consultarPartidaPeticion y sin tieneError"
# ------------------------------------------------------------------ #
# Aplicación de correcciones
# ------------------------------------------------------------------ #
@transaction.atomic
def _apply_fix(self, partida, corregibles, marcar_no_descargada, pedimento_app):
"""
Renombra/reclasifica documentos y actualiza la partida en una transacción.
Nota: si la transacción revierte, los cambios en storage NO se deshacen;
re-ejecutar el comando converge (ver _rename_in_storage).
"""
for doc, clase in corregibles:
suffix, doc_type = _RECLASIFICACION[clase]
new_name = self._pick_target_name(doc, pedimento_app, partida.numero_partida, suffix)
final_name = self._rename_in_storage(doc.archivo.name, new_name)
doc.archivo = final_name
doc.document_type_id = doc_type
doc.vu = True
doc.save(update_fields=["archivo", "document_type_id", "vu"])
self.stdout.write(self.style.SUCCESS(
f" ✓ Doc {doc.id}: type={doc_type} | {final_name}"
))
if marcar_no_descargada:
partida.descargado = False
partida.save(update_fields=["descargado"])
self.stdout.write(self.style.SUCCESS(
f" ✓ Partida {partida.numero_partida}: descargado=False"
))
def _pick_target_name(self, doc, pedimento_app, numero_partida, suffix):
"""
Primer nombre libre con nomenclatura
{dir}/vu_PT_{pedimento_app}_{numero_partida}_{SUFFIX}[_{n}].xml
verificado contra BD (excluyendo el propio doc) para que dos Documents
nunca terminen apuntando al mismo archivo (p. ej. contra el REQUEST
real type 17 que ya usa el nombre sin índice).
"""
dir_part = posixpath.dirname(doc.archivo.name)
index = 0
while True:
tail = f"_{index}" if index else ""
candidate = posixpath.join(
dir_part, f"vu_PT_{pedimento_app}_{numero_partida}_{suffix}{tail}.xml"
)
if candidate == doc.archivo.name:
return candidate
if not Document.objects.filter(archivo=candidate).exclude(id=doc.id).exists():
return candidate
index += 1
def _rename_in_storage(self, old_name, new_name):
if old_name == new_name:
return old_name
if minio_client.file_exists(new_name):
# Rename ya ocurrió en ejecución previa parcial
self.stderr.write(self.style.WARNING(
f" ⚠ Destino ya existe en storage, usando: {new_name}"
))
if minio_client.file_exists(old_name):
minio_client.delete_file(old_name)
return new_name
if not minio_client.file_exists(old_name):
self.stderr.write(self.style.WARNING(
f" ⚠ Archivo no encontrado en storage: {old_name}"
))
return old_name
response = minio_client._client.get_object(minio_client._bucket_name, old_name)
try:
content = response.read()
finally:
response.close()
response.release_conn()
minio_client.upload_file(new_name, file_data=io.BytesIO(content), content_type="application/xml")
minio_client.delete_file(old_name)
return new_name
# ------------------------------------------------------------------ #
# Resumen final
# ------------------------------------------------------------------ #
def _print_summary(self, total_peds, stats, dry_run):
self.stdout.write(
f"\n{'─' * 60}\nRESUMEN\n"
f" Pedimentos procesados : {total_peds}\n"
f" Partidas revisadas (descargado=True) : {stats['partidas']}\n"
f" Partidas corregidas (descargado=False) : {stats['corregidas']}\n"
f" Partidas sin cambios (no verificables) : {stats['bloqueadas']}\n"
f" Docs renombrados a ERROR (type 18) : {stats['docs_error']}\n"
f" Docs reclasificados a REQUEST (type 17): {stats['docs_request']}\n"
f" Docs con contenido desconocido : {stats['desconocidos']}\n"
f" Registros en BD sin archivo en storage : {stats['fantasmas']} (no se borran)\n"
)
if dry_run:
self.stdout.write(self.style.WARNING(
"\nMODO PRUEBA: ejecuta sin --dry-run para aplicar los cambios."
))
else:
self.stdout.write(self.style.SUCCESS("\nCorrección completada."))
Reference in New Issue View Git Blame Copy Permalink