564 lines
22 KiB
Python
564 lines
22 KiB
Python
"""
|
|
Tareas Celery para importación CSV de Pedimentos.
|
|
Flujo en dos fases: scan_file (validación) → insert_valid_rows (commit).
|
|
"""
|
|
import os
|
|
import base64
|
|
import csv
|
|
import json
|
|
import logging
|
|
import re
|
|
import unicodedata
|
|
from decimal import Decimal, InvalidOperation
|
|
from datetime import datetime
|
|
from typing import Dict, Any, Optional, List, Set
|
|
|
|
from core.celery_app import celery_app
|
|
from core.database import CoreSessionLocal
|
|
|
|
from .template_config import row_from_template
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Redis keys (prefijo propio para no colisionar con otros imports)
|
|
PED_IMPORT_FILE_PREFIX = "ped_import_file:"
|
|
PED_IMPORT_META_PREFIX = "ped_import_meta:"
|
|
PED_IMPORT_ERROR_LINES_PREFIX = "ped_import_error_lines:"
|
|
PED_IMPORT_REDIS_TTL = 3600 # 1 hour
|
|
|
|
|
|
def _get_redis():
|
|
import redis
|
|
url = os.getenv("VALKEY_URL", os.getenv("REDIS_URL", "redis://valkey:6379/0"))
|
|
return redis.Redis.from_url(url, decode_responses=False)
|
|
|
|
|
|
def _worker_upload_dir() -> str:
|
|
return os.path.join(os.getcwd(), "uploads", "temp")
|
|
|
|
|
|
def _ensure_worker_has_file_from_redis(job_id: str) -> Optional[str]:
|
|
r = _get_redis()
|
|
data = r.get(f"{PED_IMPORT_FILE_PREFIX}{job_id}")
|
|
if not data:
|
|
return None
|
|
try:
|
|
raw = base64.b64decode(data)
|
|
except Exception as e:
|
|
logger.warning(f"Pedimentos import: failed to decode file from Redis: {e}")
|
|
return None
|
|
upload_dir = _worker_upload_dir()
|
|
os.makedirs(upload_dir, exist_ok=True)
|
|
file_path = os.path.join(upload_dir, f"ped_{job_id}.csv")
|
|
with open(file_path, "wb") as f:
|
|
f.write(raw)
|
|
return file_path
|
|
|
|
|
|
def _ensure_worker_has_meta_from_redis(job_id: str, file_path: str) -> bool:
|
|
r = _get_redis()
|
|
data = r.get(f"{PED_IMPORT_META_PREFIX}{job_id}")
|
|
if not data:
|
|
return False
|
|
try:
|
|
meta = json.loads(data.decode("utf-8"))
|
|
except Exception as e:
|
|
logger.warning(f"Pedimentos import: failed to decode meta from Redis: {e}")
|
|
return False
|
|
meta_path = file_path.replace(".csv", ".meta.json")
|
|
with open(meta_path, "w", encoding="utf-8") as f:
|
|
json.dump(meta, f)
|
|
return True
|
|
|
|
|
|
def _delete_import_from_redis(job_id: str) -> None:
|
|
try:
|
|
r = _get_redis()
|
|
r.delete(
|
|
f"{PED_IMPORT_FILE_PREFIX}{job_id}",
|
|
f"{PED_IMPORT_META_PREFIX}{job_id}",
|
|
f"{PED_IMPORT_ERROR_LINES_PREFIX}{job_id}",
|
|
)
|
|
except Exception as e:
|
|
logger.warning(f"Pedimentos import: failed to delete Redis keys: {e}")
|
|
|
|
|
|
def normalize_header(name: Optional[str]) -> str:
|
|
if not name:
|
|
return ""
|
|
name = unicodedata.normalize("NFKD", str(name)).upper()
|
|
name = "".join(ch for ch in name if not unicodedata.combining(ch))
|
|
name = re.sub(r"[^A-Z0-9]+", " ", name)
|
|
return re.sub(r"\s+", " ", name).strip()
|
|
|
|
|
|
def _validate_row_pedimento(
|
|
row: Dict[str, Any],
|
|
line_num: int,
|
|
valid_client_ids: Set[int],
|
|
valid_regimes: Set[str],
|
|
valid_pedimento_codes: Set[str],
|
|
) -> Optional[Dict[str, Any]]:
|
|
"""Valida una fila para Pedimento. Retorna error dict o None."""
|
|
# Required: AÑO, ADUANA, PATENTE, NUMERO, CLIENTE_ID, CODIGO_PEDIMENTO, REGIMEN
|
|
year = (row.get("AÑO") or "").strip()
|
|
if not year:
|
|
return {"line": line_num, "col": "AÑO", "msg": "Requerido"}
|
|
if len(year) > 2:
|
|
return {"line": line_num, "col": "AÑO", "msg": "Máximo 2 caracteres"}
|
|
|
|
customs_office = (row.get("ADUANA") or "").strip()
|
|
if not customs_office:
|
|
return {"line": line_num, "col": "ADUANA", "msg": "Requerido"}
|
|
if len(customs_office) > 3:
|
|
return {"line": line_num, "col": "ADUANA", "msg": "Máximo 3 caracteres"}
|
|
|
|
license_val = (row.get("PATENTE") or "").strip()
|
|
if not license_val:
|
|
return {"line": line_num, "col": "PATENTE", "msg": "Requerido"}
|
|
if len(license_val) > 4:
|
|
return {"line": line_num, "col": "PATENTE", "msg": "Máximo 4 caracteres"}
|
|
|
|
pedimento_number = (row.get("NUMERO") or "").strip()
|
|
if not pedimento_number:
|
|
return {"line": line_num, "col": "NUMERO", "msg": "Requerido"}
|
|
if len(pedimento_number) > 7:
|
|
return {"line": line_num, "col": "NUMERO", "msg": "Máximo 7 caracteres"}
|
|
|
|
client_id_str = (row.get("CLIENTE_ID") or "").strip()
|
|
if not client_id_str:
|
|
return {"line": line_num, "col": "CLIENTE_ID", "msg": "Requerido"}
|
|
try:
|
|
client_id = int(client_id_str)
|
|
except ValueError:
|
|
return {"line": line_num, "col": "CLIENTE_ID", "msg": "Debe ser número entero"}
|
|
if client_id not in valid_client_ids:
|
|
return {"line": line_num, "col": "CLIENTE_ID", "msg": "Cliente no existe en catálogo"}
|
|
|
|
pedimento_code = (row.get("CODIGO_PEDIMENTO") or "").strip()
|
|
if not pedimento_code:
|
|
return {"line": line_num, "col": "CODIGO_PEDIMENTO", "msg": "Requerido"}
|
|
if len(pedimento_code) > 2:
|
|
return {"line": line_num, "col": "CODIGO_PEDIMENTO", "msg": "Máximo 2 caracteres"}
|
|
if pedimento_code not in valid_pedimento_codes:
|
|
return {"line": line_num, "col": "CODIGO_PEDIMENTO", "msg": "Código no existe en catálogo"}
|
|
|
|
regime = (row.get("REGIMEN") or "").strip()
|
|
if not regime:
|
|
return {"line": line_num, "col": "REGIMEN", "msg": "Requerido"}
|
|
if len(regime) > 3:
|
|
return {"line": line_num, "col": "REGIMEN", "msg": "Máximo 3 caracteres"}
|
|
if regime not in valid_regimes:
|
|
return {"line": line_num, "col": "REGIMEN", "msg": "Régimen no existe en catálogo"}
|
|
|
|
# Optional numeric/string fields - validate format if present
|
|
status = (row.get("ESTATUS") or "").strip()
|
|
if status and len(status) > 30:
|
|
return {"line": line_num, "col": "ESTATUS", "msg": "Máximo 30 caracteres"}
|
|
|
|
for col, max_len in [("VALOR_USD", 17), ("PRECIO_PAGADO", 17), ("PESO_BRUTO", 19), ("TIPO_CAMBIO", 9)]:
|
|
val = (row.get(col) or "").strip()
|
|
if not val:
|
|
continue
|
|
try:
|
|
Decimal(val.replace(",", "."))
|
|
except (InvalidOperation, ValueError):
|
|
return {"line": line_num, "col": col, "msg": "Valor numérico inválido"}
|
|
|
|
operation_type = (row.get("TIPO_OPERACION") or "").strip().lower()
|
|
if operation_type and operation_type not in ("imp", "exp", ""):
|
|
return {"line": line_num, "col": "TIPO_OPERACION", "msg": "Debe ser imp o exp"}
|
|
|
|
pedimento_type = (row.get("TIPO_PEDIMENTO") or "").strip().lower()
|
|
if pedimento_type and pedimento_type not in ("normal", "consolidated", "complementary", "automobile", ""):
|
|
return {"line": line_num, "col": "TIPO_PEDIMENTO", "msg": "Tipo no válido (normal, consolidated, complementary, automobile)"}
|
|
|
|
return None
|
|
|
|
|
|
def _parse_decimal(val: Any) -> Optional[Decimal]:
|
|
if val is None or (isinstance(val, str) and not val.strip()):
|
|
return None
|
|
try:
|
|
return Decimal(str(val).strip().replace(",", "."))
|
|
except (InvalidOperation, ValueError):
|
|
return None
|
|
|
|
|
|
def _row_to_pedimentos_create(row: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Build dict for PedimentosCreate from normalized CSV row (canonical names)."""
|
|
year = (row.get("AÑO") or "").strip()[:2]
|
|
customs_office = (row.get("ADUANA") or "").strip()[:3]
|
|
license_val = (row.get("PATENTE") or "").strip()[:4]
|
|
pedimento_number = (row.get("NUMERO") or "").strip()[:7]
|
|
client_id_str = (row.get("CLIENTE_ID") or "").strip()
|
|
client_id = int(client_id_str) if client_id_str else None
|
|
pedimento_code = (row.get("CODIGO_PEDIMENTO") or "").strip()[:2]
|
|
regime = (row.get("REGIMEN") or "").strip()[:3]
|
|
|
|
data = {
|
|
"year": year,
|
|
"customs_office": customs_office,
|
|
"license": license_val,
|
|
"pedimento_number": pedimento_number,
|
|
"client_id": client_id,
|
|
"pedimento_code": pedimento_code,
|
|
"regime": regime,
|
|
}
|
|
|
|
op = (row.get("TIPO_OPERACION") or "").strip().lower()
|
|
if op in ("imp", "exp"):
|
|
data["operation_type"] = op
|
|
|
|
ptype = (row.get("TIPO_PEDIMENTO") or "").strip().lower()
|
|
if ptype in ("normal", "consolidated", "complementary", "automobile"):
|
|
data["pedimento_type"] = ptype
|
|
|
|
status = (row.get("ESTATUS") or "").strip()
|
|
if status:
|
|
data["status"] = status[:30]
|
|
|
|
data["usd_value"] = _parse_decimal(row.get("VALOR_USD"))
|
|
data["paid_price"] = _parse_decimal(row.get("PRECIO_PAGADO"))
|
|
data["gross_weight"] = _parse_decimal(row.get("PESO_BRUTO"))
|
|
data["exchange_rate"] = _parse_decimal(row.get("TIPO_CAMBIO"))
|
|
|
|
obs = (row.get("OBSERVACIONES") or "").strip()
|
|
if obs:
|
|
data["observations"] = obs
|
|
|
|
return data
|
|
|
|
|
|
@celery_app.task(bind=True)
|
|
def scan_file(self, job_id: str, config: str = None):
|
|
"""
|
|
Fase 1: Leer CSV, validar filas, escribir errores en JSONL.
|
|
Devuelve waiting_confirmation con total_rows, error_count, valid_rows, errors.
|
|
"""
|
|
logger.info(f"Pedimentos import: starting scan for job {job_id}")
|
|
|
|
file_path = _ensure_worker_has_file_from_redis(job_id)
|
|
if not file_path:
|
|
return {"status": "failed", "error": "Archivo no encontrado (expirado o no subido). Sube de nuevo."}
|
|
_ensure_worker_has_meta_from_redis(job_id, file_path)
|
|
|
|
error_dir = os.path.join(os.path.dirname(file_path).replace("temp", "errors"), "")
|
|
os.makedirs(error_dir, exist_ok=True)
|
|
error_path = os.path.join(error_dir, f"ped_{job_id}.jsonl")
|
|
|
|
total_rows = 0
|
|
try:
|
|
with open(file_path, "r", encoding="utf-8-sig") as f:
|
|
total_rows = sum(1 for _ in f) - 1
|
|
except Exception as e:
|
|
return {"status": "failed", "error": str(e)}
|
|
|
|
meta_path = file_path.replace(".csv", ".meta.json")
|
|
meta = {}
|
|
if os.path.exists(meta_path):
|
|
try:
|
|
with open(meta_path, "r", encoding="utf-8") as f:
|
|
meta = json.load(f) or {}
|
|
except Exception as e:
|
|
logger.warning(f"Pedimentos import: failed to read meta: {e}")
|
|
|
|
tenant_id = meta.get("tenant_id")
|
|
company_id = meta.get("company_id")
|
|
if not tenant_id or not company_id:
|
|
return {"status": "failed", "error": "Falta contexto (tenant/company)"}
|
|
|
|
# Load valid FK sets for validation
|
|
valid_client_ids: Set[int] = set()
|
|
valid_regimes: Set[str] = set()
|
|
valid_pedimento_codes: Set[str] = set()
|
|
try:
|
|
with CoreSessionLocal() as session:
|
|
from api.v1.modules.a76.clients_and_providers.models import ClientProvider
|
|
from api.v1.modules.public.reference_data.pedimento_regimens.models import RegimenPedimento
|
|
from api.v1.modules.public.reference_data.pedimento_codes.models import PedimentoCode
|
|
|
|
for cp in session.query(ClientProvider).filter(
|
|
ClientProvider.tenant_id == tenant_id,
|
|
ClientProvider.company_id == company_id,
|
|
).all():
|
|
valid_client_ids.add(cp.id)
|
|
for r in session.query(RegimenPedimento).all():
|
|
valid_regimes.add(r.code)
|
|
for pc in session.query(PedimentoCode).all():
|
|
valid_pedimento_codes.add(pc.code)
|
|
except Exception as e:
|
|
logger.error(f"Pedimentos import: failed to load FK sets: {e}")
|
|
return {"status": "failed", "error": f"No se pudo cargar catálogos: {e}"}
|
|
|
|
error_count = 0
|
|
processed_rows = 0
|
|
errors_detail: List[Dict[str, Any]] = []
|
|
|
|
try:
|
|
with open(file_path, "r", encoding="utf-8-sig") as f_in, open(
|
|
error_path, "w", encoding="utf-8"
|
|
) as f_err:
|
|
sample = f_in.read(2048)
|
|
f_in.seek(0)
|
|
try:
|
|
dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
|
|
except Exception:
|
|
dialect = "excel"
|
|
reader = csv.DictReader(f_in, dialect=dialect)
|
|
|
|
for i, row in enumerate(reader, start=1):
|
|
if i % 500 == 0:
|
|
self.update_state(
|
|
state="PROGRESS",
|
|
meta={"current": i, "total": total_rows, "errors": error_count},
|
|
)
|
|
|
|
row_norm = row_from_template(row, normalize_header, "pedimentos")
|
|
err = _validate_row_pedimento(
|
|
row_norm, i, valid_client_ids, valid_regimes, valid_pedimento_codes
|
|
)
|
|
if err:
|
|
error_count += 1
|
|
f_err.write(json.dumps(err) + "\n")
|
|
if len(errors_detail) < 500:
|
|
errors_detail.append(
|
|
{"line": err["line"], "col": err.get("col", ""), "msg": err.get("msg", "")}
|
|
)
|
|
processed_rows += 1
|
|
|
|
except Exception as e:
|
|
logger.error(f"Pedimentos import scan failed: {e}")
|
|
return {"status": "failed", "error": str(e)}
|
|
|
|
error_lines_list = []
|
|
try:
|
|
if os.path.exists(error_path):
|
|
with open(error_path, "r", encoding="utf-8") as f:
|
|
for line in f:
|
|
try:
|
|
err = json.loads(line)
|
|
if "line" in err:
|
|
error_lines_list.append(err["line"])
|
|
except Exception:
|
|
pass
|
|
if error_lines_list:
|
|
r = _get_redis()
|
|
r.set(
|
|
f"{PED_IMPORT_ERROR_LINES_PREFIX}{job_id}",
|
|
json.dumps(error_lines_list).encode("utf-8"),
|
|
ex=PED_IMPORT_REDIS_TTL,
|
|
)
|
|
except Exception as e:
|
|
logger.warning(f"Pedimentos import: failed to store error lines in Redis: {e}")
|
|
|
|
return {
|
|
"status": "waiting_confirmation",
|
|
"job_id": job_id,
|
|
"total_rows": processed_rows,
|
|
"error_count": error_count,
|
|
"valid_rows": processed_rows - error_count,
|
|
"errors": errors_detail,
|
|
}
|
|
|
|
|
|
@celery_app.task(bind=True)
|
|
def insert_valid_rows(self, job_id: str):
|
|
"""
|
|
Fase 2: Re-leer CSV, omitir filas con error, insertar Pedimentos vía PedimentosService.create.
|
|
"""
|
|
logger.info(f"Pedimentos import: starting commit for job {job_id}")
|
|
|
|
file_path = _ensure_worker_has_file_from_redis(job_id)
|
|
if not file_path:
|
|
alt_path = os.path.join(_worker_upload_dir(), f"ped_{job_id}.csv")
|
|
if not os.path.exists(alt_path):
|
|
return {
|
|
"status": "failed",
|
|
"error": "Archivo no encontrado (expirado). Sube y confirma de nuevo.",
|
|
}
|
|
file_path = alt_path
|
|
else:
|
|
_ensure_worker_has_meta_from_redis(job_id, file_path)
|
|
|
|
base_dir = os.path.dirname(file_path)
|
|
error_dir = base_dir.replace("temp", "errors")
|
|
error_path = os.path.join(error_dir, f"ped_{job_id}.jsonl")
|
|
|
|
error_lines = set()
|
|
try:
|
|
r = _get_redis()
|
|
raw = r.get(f"{PED_IMPORT_ERROR_LINES_PREFIX}{job_id}")
|
|
if raw:
|
|
error_lines = set(json.loads(raw.decode("utf-8")))
|
|
except Exception as e:
|
|
logger.debug(f"Pedimentos import: could not load error lines from Redis: {e}")
|
|
if not error_lines and os.path.exists(error_path):
|
|
with open(error_path, "r", encoding="utf-8") as f:
|
|
for line in f:
|
|
try:
|
|
err = json.loads(line)
|
|
error_lines.add(err["line"])
|
|
except Exception:
|
|
pass
|
|
|
|
meta_path = file_path.replace(".csv", ".meta.json")
|
|
tenant_id = None
|
|
company_id = None
|
|
meta = {}
|
|
if os.path.exists(meta_path):
|
|
try:
|
|
with open(meta_path, "r", encoding="utf-8") as f:
|
|
meta = json.load(f) or {}
|
|
tenant_id = meta.get("tenant_id")
|
|
company_id = meta.get("company_id")
|
|
except Exception:
|
|
pass
|
|
|
|
if not tenant_id or not company_id:
|
|
return {"status": "failed", "error": "Falta contexto (tenant/company)"}
|
|
|
|
# Reload FK sets for commit-time validation
|
|
valid_client_ids: Set[int] = set()
|
|
valid_regimes: Set[str] = set()
|
|
valid_pedimento_codes: Set[str] = set()
|
|
try:
|
|
with CoreSessionLocal() as session:
|
|
from api.v1.modules.a76.clients_and_providers.models import ClientProvider
|
|
from api.v1.modules.public.reference_data.pedimento_regimens.models import RegimenPedimento
|
|
from api.v1.modules.public.reference_data.pedimento_codes.models import PedimentoCode
|
|
|
|
for cp in session.query(ClientProvider).filter(
|
|
ClientProvider.tenant_id == tenant_id,
|
|
ClientProvider.company_id == company_id,
|
|
).all():
|
|
valid_client_ids.add(cp.id)
|
|
for r in session.query(RegimenPedimento).all():
|
|
valid_regimes.add(r.code)
|
|
for pc in session.query(PedimentoCode).all():
|
|
valid_pedimento_codes.add(pc.code)
|
|
except Exception as e:
|
|
logger.error(f"Pedimentos import: failed to load FK sets: {e}")
|
|
return {"status": "failed", "error": str(e)}
|
|
|
|
from api.v1.modules.a76.pedmientos.dtos.pedimentos import PedimentosCreate
|
|
from api.v1.modules.a76.pedmientos.dtos.pedimento_dates import PedimentoDatesCreate
|
|
from api.v1.modules.a76.pedmientos.services.pedimentos import PedimentosService
|
|
|
|
inserted_count = 0
|
|
skipped_invalid = 0
|
|
skipped_missing_fk = 0
|
|
skipped_duplicate = 0
|
|
skipped_details: List[Dict[str, Any]] = []
|
|
response = None
|
|
|
|
try:
|
|
with CoreSessionLocal() as session:
|
|
with open(file_path, "r", encoding="utf-8-sig") as f:
|
|
sample = f.read(2048)
|
|
f.seek(0)
|
|
try:
|
|
dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
|
|
except Exception:
|
|
dialect = "excel"
|
|
reader = csv.DictReader(f, dialect=dialect)
|
|
|
|
for i, row in enumerate(reader, start=1):
|
|
if i in error_lines:
|
|
continue
|
|
|
|
row_norm = row_from_template(row, normalize_header, "pedimentos")
|
|
err = _validate_row_pedimento(
|
|
row_norm, i, valid_client_ids, valid_regimes, valid_pedimento_codes
|
|
)
|
|
if err:
|
|
skipped_invalid += 1
|
|
skipped_details.append(
|
|
{"line": i, "reason": f"{err.get('col', '')}: {err.get('msg', '')}"}
|
|
)
|
|
continue
|
|
|
|
try:
|
|
data = _row_to_pedimentos_create(row_norm)
|
|
# Service expects at least pedimento_dates with entry_date/end_date for DB NOT NULL
|
|
if "pedimento_dates" not in data or data.get("pedimento_dates") is None:
|
|
data["pedimento_dates"] = PedimentoDatesCreate(
|
|
entry_date=datetime.now(),
|
|
end_date=datetime.now(),
|
|
)
|
|
create_data = PedimentosCreate(**data)
|
|
PedimentosService.create(session, create_data, tenant_id, company_id)
|
|
inserted_count += 1
|
|
except ValueError as ve:
|
|
if "Ya existe" in str(ve) or "duplicate" in str(ve).lower():
|
|
skipped_duplicate += 1
|
|
skipped_details.append({"line": i, "reason": str(ve)})
|
|
else:
|
|
skipped_invalid += 1
|
|
skipped_details.append({"line": i, "reason": str(ve)})
|
|
except Exception as e:
|
|
logger.warning(f"Pedimentos import line {i}: {e}")
|
|
skipped_invalid += 1
|
|
skipped_details.append({"line": i, "reason": str(e)})
|
|
|
|
total_skipped = skipped_invalid + skipped_missing_fk + skipped_duplicate
|
|
if inserted_count == 0 and total_skipped > 0:
|
|
response = {
|
|
"status": "warning",
|
|
"inserted": 0,
|
|
"skipped_invalid": skipped_invalid,
|
|
"skipped_missing_fk": skipped_missing_fk,
|
|
"skipped_duplicate": skipped_duplicate,
|
|
"skipped_details": skipped_details,
|
|
"message": f"No se insertaron registros. {total_skipped} rechazados.",
|
|
}
|
|
elif inserted_count == 0:
|
|
response = {
|
|
"status": "failed",
|
|
"error": "No hay registros válidos en el archivo CSV",
|
|
"inserted": 0,
|
|
"skipped_invalid": skipped_invalid,
|
|
"skipped_missing_fk": skipped_missing_fk,
|
|
"skipped_duplicate": skipped_duplicate,
|
|
"skipped_details": skipped_details,
|
|
}
|
|
else:
|
|
response = {
|
|
"status": "finished",
|
|
"inserted": inserted_count,
|
|
"skipped_invalid": skipped_invalid,
|
|
"skipped_missing_fk": skipped_missing_fk,
|
|
"skipped_duplicate": skipped_duplicate,
|
|
"skipped_details": skipped_details,
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Pedimentos import task failed: {e}")
|
|
import traceback
|
|
logger.error(traceback.format_exc())
|
|
return {"status": "failed", "error": str(e)}
|
|
|
|
try:
|
|
if file_path and os.path.exists(file_path):
|
|
os.remove(file_path)
|
|
if os.path.exists(error_path):
|
|
os.remove(error_path)
|
|
meta_path = file_path.replace(".csv", ".meta.json")
|
|
if os.path.exists(meta_path):
|
|
os.remove(meta_path)
|
|
_delete_import_from_redis(job_id)
|
|
except Exception as cleanup_err:
|
|
logger.warning(f"Pedimentos import cleanup failed: {cleanup_err}")
|
|
|
|
if response is None:
|
|
response = {
|
|
"status": "failed",
|
|
"error": "Error inesperado",
|
|
"inserted": 0,
|
|
"skipped_invalid": skipped_invalid,
|
|
"skipped_missing_fk": skipped_missing_fk,
|
|
"skipped_duplicate": skipped_duplicate,
|
|
"skipped_details": skipped_details,
|
|
}
|
|
return response
|