1739 lines
80 KiB
Python
1739 lines
80 KiB
Python
import os
|
|
from datetime import datetime
|
|
from decimal import Decimal
|
|
import csv
|
|
import json
|
|
import logging
|
|
import re
|
|
import unicodedata
|
|
from typing import Dict, Any, Optional, List, Set, Tuple
|
|
|
|
from core.celery_app import celery_app
|
|
from core.database import CoreSessionLocal
|
|
from core.paths import layout_path
|
|
|
|
from ..common import storage as common_storage
|
|
from ..common import meta as common_meta
|
|
from ..common import responses as common_responses
|
|
from .template_config import row_from_template
|
|
# Models are imported inside tasks to avoid circular dependencies and mapper initialization issues in the API process
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Job type vacío para facturas (prefijo Redis "import_" sin tipo, ver common/storage.py)
|
|
JOB_TYPE = ""
|
|
|
|
# Redis keys and TTL for import file/meta (exportados para routes; coinciden con common_storage cuando job_type="")
|
|
IMPORT_FILE_KEY_PREFIX = "import_file:"
|
|
IMPORT_META_KEY_PREFIX = "import_meta:"
|
|
IMPORT_ERROR_LINES_KEY_PREFIX = "import_error_lines:"
|
|
IMPORT_REDIS_TTL = common_storage.IMPORT_REDIS_TTL
|
|
|
|
|
|
def _ensure_worker_has_file_from_redis(job_id: str) -> Optional[str]:
|
|
"""Usa common storage con job_type vacío (prefijo import_)."""
|
|
return common_storage.ensure_file_from_redis(JOB_TYPE, job_id, "Invoices import")
|
|
|
|
|
|
def _ensure_worker_has_meta_from_redis(job_id: str, file_path: str) -> bool:
|
|
return common_storage.ensure_meta_from_redis(JOB_TYPE, job_id, file_path, "Invoices import")
|
|
|
|
|
|
def _delete_import_from_redis(job_id: str) -> None:
|
|
common_storage.delete_import_from_redis(JOB_TYPE, job_id)
|
|
|
|
class ForeignKeyValidator:
|
|
def __init__(self, session, tenant_id, company_id):
|
|
self.session = session
|
|
self.tenant_id = tenant_id
|
|
self.company_id = company_id
|
|
self.cache = {} # {(model_name, value): bool}
|
|
|
|
def check_exists(self, model, value, field_name="id", is_public=False):
|
|
if value is None:
|
|
return True # Assume optional if None, or let DB handle not-null
|
|
|
|
key = (model.__name__, value)
|
|
if key in self.cache:
|
|
return self.cache[key]
|
|
|
|
query = self.session.query(getattr(model, field_name)).filter(getattr(model, field_name) == value)
|
|
if not is_public:
|
|
query = query.filter(model.tenant_id == self.tenant_id, model.company_id == self.company_id)
|
|
|
|
exists = query.first() is not None
|
|
self.cache[key] = exists
|
|
return exists
|
|
|
|
|
|
TRANSPORT_TYPE_VALUES = {
|
|
"none",
|
|
"transport",
|
|
"box",
|
|
"licence plates",
|
|
"truck",
|
|
"vessel",
|
|
"rail_barge",
|
|
"container",
|
|
"airplane",
|
|
"gondola",
|
|
"flatbed",
|
|
}
|
|
|
|
|
|
def normalize_public_code(value: Optional[str]) -> Optional[str]:
|
|
if value is None:
|
|
return None
|
|
text = str(value).strip().upper()
|
|
return text or None
|
|
|
|
|
|
def validate_public_code(
|
|
validator: ForeignKeyValidator,
|
|
model,
|
|
value: Optional[str],
|
|
line_num: int,
|
|
col_name: str,
|
|
field_name: str = "code",
|
|
required: bool = False,
|
|
) -> Optional[Dict[str, Any]]:
|
|
code = normalize_public_code(value)
|
|
if not code:
|
|
if required:
|
|
return {"line": line_num, "col": col_name, "msg": "Requerido"}
|
|
return None
|
|
if not validator.check_exists(model, code, field_name=field_name, is_public=True):
|
|
return {"line": line_num, "col": col_name, "msg": "No existe en el catalogo"}
|
|
return None
|
|
|
|
|
|
def validate_tenant_fk_id(
|
|
validator: ForeignKeyValidator,
|
|
model,
|
|
value: Optional[int],
|
|
line_num: int,
|
|
col_name: str,
|
|
required: bool = False,
|
|
) -> Optional[Dict[str, Any]]:
|
|
if value is None:
|
|
if required:
|
|
return {"line": line_num, "col": col_name, "msg": "Requerido"}
|
|
return None
|
|
if not validator.check_exists(model, value):
|
|
return {"line": line_num, "col": col_name, "msg": "No existe en el catalogo"}
|
|
return None
|
|
|
|
@celery_app.task(bind=True)
|
|
def scan_file(self, job_id: str, model_target: str, config: str = None):
|
|
"""
|
|
Pass 1: Read CSV, Validate types, Write Errors to JSONL.
|
|
File content is loaded from Redis (written by API on upload) so worker does not need shared filesystem.
|
|
"""
|
|
logger.info(f"Starting scan for job {job_id} target {model_target}")
|
|
|
|
# 1. Get file from Redis and write to worker local disk
|
|
file_path = _ensure_worker_has_file_from_redis(job_id)
|
|
if not file_path:
|
|
return {"status": "failed", "error": "File not found (missing or expired in queue). Please upload again."}
|
|
_ensure_worker_has_meta_from_redis(job_id, file_path)
|
|
|
|
error_path = common_storage.error_path_for_job(JOB_TYPE, job_id)
|
|
|
|
total_rows = 0
|
|
error_count = 0
|
|
processed_rows = 0
|
|
|
|
# 3. Count Total (Quick Pass) or just estimate
|
|
try:
|
|
with open(file_path, 'r', encoding='utf-8-sig') as f:
|
|
total_rows = sum(1 for _ in f) - 1 # Minus header
|
|
except Exception as e:
|
|
return {"status": "failed", "error": f"Cannot read file: {e}"}
|
|
|
|
footer_config = parse_footer_config(config)
|
|
date_format = footer_config.get("dateFormat")
|
|
|
|
# Validate and set default date_format if not provided
|
|
if not date_format:
|
|
date_format = "yyyy-mm-dd" # Default to ISO format
|
|
logger.info(f"No date_format specified in config, using default: {date_format}")
|
|
|
|
try:
|
|
tenant_id, company_id = common_meta.require_tenant_context(file_path)
|
|
except ValueError as e:
|
|
return {"status": "failed", "error": str(e)}
|
|
|
|
meta = common_meta.load_meta(file_path)
|
|
template_id = meta.get("template_id") or (
|
|
"imp_temp_header" if model_target == "invoice_header" else
|
|
"imp_temp_details" if model_target == "invoice_details" else "imp_temp_series"
|
|
)
|
|
|
|
inv_type_value = normalize_public_code(footer_config.get("invoice_type") or meta.get("invoice_type") or "TEM")
|
|
if not inv_type_value:
|
|
inv_type_value = "TEM"
|
|
|
|
# --- Series de Importación Temporal: flujo específico (Clarion VALIDA_TODA / VALIDA_PARCIAL) ---
|
|
if model_target == "invoice_series":
|
|
try:
|
|
from api.v1.modules.a76.invoices.models import InvoiceHeader
|
|
from api.v1.modules.a76.items.models import LineItem
|
|
from api.v1.modules.a76.items.series.models import Serie
|
|
from .validators.series_impo_temp import (
|
|
validate_row_series_impo_temp,
|
|
)
|
|
|
|
actualizar = meta.get("actualizar", False)
|
|
autonumerar = meta.get("autonumerar", True)
|
|
validar_series_exception = meta.get("validar_series", False)
|
|
_fc = parse_footer_config(meta.get("footer_config"))
|
|
if _fc:
|
|
if "actualizar" in _fc:
|
|
actualizar = bool(_fc["actualizar"])
|
|
elif _fc.get("mode") == "update":
|
|
actualizar = True
|
|
elif _fc.get("mode") == "replace":
|
|
actualizar = False
|
|
if "autonumerar" in _fc:
|
|
autonumerar = bool(_fc["autonumerar"])
|
|
else:
|
|
as_val = _fc.get("autonumber_series", "true")
|
|
autonumerar = str(as_val).lower() in ("true", "1", "si", "sí", "yes")
|
|
if "validar_series" in _fc:
|
|
validar_series_exception = bool(_fc["validar_series"])
|
|
|
|
with CoreSessionLocal() as session:
|
|
# Invoice lookup: imp + TEM
|
|
q = (
|
|
session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.is_updated)
|
|
.filter(
|
|
InvoiceHeader.tenant_id == tenant_id,
|
|
InvoiceHeader.company_id == company_id,
|
|
InvoiceHeader.operation_type == "imp",
|
|
InvoiceHeader.invoice_type == "TEM",
|
|
)
|
|
)
|
|
rows_inv = q.all()
|
|
invoice_id_by_number: Dict[str, int] = {}
|
|
invoice_updated_by_number: Dict[str, bool] = {}
|
|
for num, iid, is_upd in rows_inv:
|
|
if num:
|
|
invoice_id_by_number[str(num).strip()] = iid
|
|
invoice_updated_by_number[str(num).strip()] = bool(is_upd)
|
|
|
|
# Existing series keys: (invoice_number, linea_factura, linea_serie)
|
|
existing_series_keys: Set[Tuple[str, str, str]] = set()
|
|
existing_series_data: Dict[Tuple[str, str, str], Dict[str, Any]] = {}
|
|
if actualizar and not autonumerar:
|
|
q_ser = (
|
|
session.query(
|
|
InvoiceHeader.invoice_number,
|
|
LineItem.line_number,
|
|
Serie.row,
|
|
Serie.serial_numbers,
|
|
Serie.model,
|
|
Serie.sub_model,
|
|
Serie.number_id,
|
|
)
|
|
.join(LineItem, LineItem.invoice_id == InvoiceHeader.id)
|
|
.join(Serie, Serie.line_item_id == LineItem.id)
|
|
.filter(
|
|
InvoiceHeader.tenant_id == tenant_id,
|
|
InvoiceHeader.company_id == company_id,
|
|
InvoiceHeader.operation_type == "imp",
|
|
InvoiceHeader.invoice_type == "TEM",
|
|
)
|
|
)
|
|
for num, ln, rw, sn, md, sm, nid in q_ser.all():
|
|
if num is not None:
|
|
k = (str(num).strip(), str(ln).strip(), str(rw).strip())
|
|
existing_series_keys.add(k)
|
|
existing_series_data.setdefault(k, {
|
|
"serial_numbers": sn or "",
|
|
"model": md or "",
|
|
"sub_model": sm or "",
|
|
"number_id": nid or "",
|
|
})
|
|
|
|
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)
|
|
errors_detail: List[Dict[str, Any]] = []
|
|
error_lines_list: List[int] = []
|
|
for i, row in enumerate(reader, start=1):
|
|
if i % 1000 == 0:
|
|
self.update_state(state="PROGRESS", meta={"current": i, "total": total_rows, "errors": len(error_lines_list)})
|
|
row_norm = row_from_template(row, "imp_temp_series", normalize_header)
|
|
warnings_list: List[Dict[str, Any]] = []
|
|
err = validate_row_series_impo_temp(
|
|
row_norm,
|
|
i,
|
|
actualizar=actualizar,
|
|
autonumerar=autonumerar,
|
|
validar_series_exception=validar_series_exception,
|
|
invoice_id_by_number=invoice_id_by_number,
|
|
invoice_updated_by_number=invoice_updated_by_number,
|
|
existing_series_keys=existing_series_keys,
|
|
existing_series_data=existing_series_data,
|
|
warnings=warnings_list,
|
|
)
|
|
if err and not err.get("warning"):
|
|
error_count += 1
|
|
error_lines_list.append(err["line"])
|
|
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", "")})
|
|
for w in warnings_list:
|
|
if len(errors_detail) < 500:
|
|
errors_detail.append({"line": w["line"], "col": w.get("col", ""), "msg": w.get("msg", ""), "warning": True})
|
|
processed_rows += 1
|
|
|
|
if error_lines_list:
|
|
common_storage.store_error_lines(JOB_TYPE, job_id, error_lines_list)
|
|
return common_responses.scan_result(
|
|
job_id, processed_rows, error_count, errors_detail, total_rows_in_file=total_rows
|
|
)
|
|
except Exception as e:
|
|
logger.exception("Series import scan failed: %s", e)
|
|
return {"status": "failed", "error": str(e)}
|
|
|
|
try:
|
|
from api.v1.modules.a76.invoices.models import InvoiceHeader
|
|
from api.v1.modules.a76.clients_and_providers.models import ClientProvider
|
|
from api.v1.modules.a76.customs_brokers.models import CustomsBroker
|
|
from api.v1.modules.public.reference_data.currency_types.models import CurrencyType
|
|
from api.v1.modules.public.reference_data.pedimento_codes.models import PedimentoCode
|
|
from api.v1.modules.public.reference_data.code_pedimento_regimens.models import (
|
|
CodePedimentoRegimen,
|
|
)
|
|
from api.v1.modules.public.reference_data.pedimento_regimens.models import RegimenPedimento
|
|
from api.v1.modules.public.reference_data.invoice_types.models import InvoiceType
|
|
from api.v1.modules.public.reference_data.customs_sections.models import CustomsSection
|
|
from api.v1.modules.public.reference_data.incoterms.models import Incoterm
|
|
from api.v1.modules.a76.parts.models import Part
|
|
|
|
models = {
|
|
"InvoiceHeader": InvoiceHeader,
|
|
"InvoiceType": InvoiceType,
|
|
"ClientProvider": ClientProvider,
|
|
"CustomsBroker": CustomsBroker,
|
|
"RegimenPedimento": RegimenPedimento,
|
|
"CodePedimentoRegimen": CodePedimentoRegimen,
|
|
"PedimentoCode": PedimentoCode,
|
|
"CurrencyType": CurrencyType,
|
|
"CustomsSection": CustomsSection,
|
|
"Incoterm": Incoterm,
|
|
"Part": Part,
|
|
}
|
|
|
|
with CoreSessionLocal() as session, \
|
|
open(file_path, 'r', encoding='utf-8-sig') as f_in, \
|
|
open(error_path, 'w', encoding='utf-8') as f_err:
|
|
validator = ForeignKeyValidator(session, tenant_id, company_id)
|
|
invoice_id_cache: Dict[str, Optional[int]] = {}
|
|
|
|
# Detect Delimiter
|
|
sample = f_in.read(2048)
|
|
f_in.seek(0)
|
|
try:
|
|
dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
|
|
except:
|
|
dialect = 'excel'
|
|
|
|
reader = csv.DictReader(f_in, dialect=dialect)
|
|
|
|
for i, row in enumerate(reader, start=1):
|
|
# Check for Progress Update
|
|
if i % 1000 == 0:
|
|
self.update_state(state='PROGRESS', meta={
|
|
'current': i,
|
|
'total': total_rows,
|
|
'errors': error_count
|
|
})
|
|
|
|
# Solo columnas de la plantilla (respetar plantilla tal cual)
|
|
row_norm = row_from_template(row, template_id, normalize_header)
|
|
errors = validate_row_strict(
|
|
row_norm,
|
|
model_target,
|
|
i,
|
|
date_format,
|
|
validator,
|
|
inv_type_value,
|
|
invoice_id_cache,
|
|
models,
|
|
)
|
|
|
|
if errors:
|
|
error_count += 1
|
|
# Write simple JSON error
|
|
f_err.write(json.dumps(errors) + "\n")
|
|
|
|
processed_rows += 1
|
|
|
|
except Exception as e:
|
|
logger.error(f"Scan failed: {e}")
|
|
return {"status": "failed", "error": str(e)}
|
|
|
|
# 4. Store error line numbers in Redis so insert_valid_rows can skip them (any worker)
|
|
error_lines_list = []
|
|
errors_detail: List[Dict[str, Any]] = []
|
|
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"])
|
|
if len(errors_detail) < 500:
|
|
errors_detail.append(
|
|
{
|
|
"line": err["line"],
|
|
"col": err.get("col", ""),
|
|
"msg": err.get("msg", ""),
|
|
}
|
|
)
|
|
except Exception:
|
|
pass
|
|
if error_lines_list:
|
|
common_storage.store_error_lines(JOB_TYPE, job_id, error_lines_list)
|
|
except Exception as e:
|
|
logger.warning(f"Failed to store error lines in Redis: {e}")
|
|
|
|
return common_responses.scan_result(job_id, processed_rows, error_count, errors_detail)
|
|
|
|
def validate_row_phase_1(
|
|
row: Dict[str, Any],
|
|
target: str,
|
|
line_num: int,
|
|
date_format: Optional[str],
|
|
) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Validation: Unique IDs, Dates, and Numeric constraint checks.
|
|
Target: 'invoice_header' or 'invoice_details'
|
|
"""
|
|
def check_decimal(col_name):
|
|
val = row.get(col_name)
|
|
if val and str(val).strip():
|
|
if parse_decimal(val) is None:
|
|
return {"line": line_num, "col": col_name, "msg": "Debe ser un número decimal válido"}
|
|
return None
|
|
|
|
def check_int(col_name):
|
|
val = row.get(col_name)
|
|
if val and str(val).strip():
|
|
if parse_int(val) is None:
|
|
return {"line": line_num, "col": col_name, "msg": "Debe ser un número entero válido"}
|
|
return None
|
|
|
|
def check_date(col_name):
|
|
date_str = row.get(col_name)
|
|
if date_str and str(date_str).strip():
|
|
if not is_valid_date(date_str, date_format):
|
|
expected = display_date_format(date_format)
|
|
return {
|
|
"line": line_num,
|
|
"col": col_name,
|
|
"msg": f"Formato de fecha inválido ({expected})",
|
|
}
|
|
return None
|
|
|
|
def check_weight(col_name):
|
|
val = row.get(col_name)
|
|
if val and str(val).strip():
|
|
if parse_weight_unit(val) is None:
|
|
return {"line": line_num, "col": col_name, "msg": "Unidad de peso inválida (ej. KGS, LBS)"}
|
|
return None
|
|
|
|
def check_currency(col_name):
|
|
val = row.get(col_name)
|
|
if val and str(val).strip():
|
|
parsed_currency = parse_currency(val, None)
|
|
val_norm = normalize_header(val)
|
|
# parse_currency returns MANUAL if unknown, so if it wasn't explicitly MANUAL, it's invalid
|
|
if parsed_currency.value == "manual" and "MANUAL" not in val_norm:
|
|
return {"line": line_num, "col": col_name, "msg": "Moneda inválida (ej. MN, ME, USD, PESOS)"}
|
|
return None
|
|
|
|
def check_transport_type(col_name):
|
|
val = row.get(col_name)
|
|
if val and str(val).strip():
|
|
if str(val).strip().lower() not in TRANSPORT_TYPE_VALUES:
|
|
return {"line": line_num, "col": col_name, "msg": "Tipo de transporte inválido (ej. box, truck, container)"}
|
|
return None
|
|
|
|
# A. Invoice Header
|
|
if target == 'invoice_header':
|
|
# 1. Unique ID
|
|
if not row.get('NUMERO FACTURA') and not row.get('NUM FACTURA') and not row.get('ID'):
|
|
return {"line": line_num, "col": "NUMERO FACTURA", "msg": "Requerido"}
|
|
|
|
# 2. Date Format
|
|
date_str = row.get('FECHA FACTURA') or row.get('FECHA')
|
|
if not date_str or not str(date_str).strip():
|
|
return {"line": line_num, "col": "FECHA FACTURA", "msg": "Requerido"}
|
|
|
|
err = check_date('FECHA FACTURA') or check_date('FECHA')
|
|
if err: return err
|
|
|
|
err = check_date('FECHA EMISION')
|
|
if err: return err
|
|
|
|
# 3. Numeric Fields
|
|
for col in ['TIPO DE CAMBIO', 'FLETES', 'VALOR SEGUROS', 'SEGUROS', 'EMBALAJES', 'OTROS INCREMENTABLES']:
|
|
err = check_decimal(col)
|
|
if err: return err
|
|
|
|
# 4. Integer FKs
|
|
for col in ['CLAVE PROVEEDOR', 'CLAVE VENDIDO A', 'CLAVE ENVIADO A', 'AGENTE ADUANAL', 'REMESA']:
|
|
err = check_int(col)
|
|
if err: return err
|
|
|
|
# 5. Enums
|
|
for col in ['TIPO PESO']:
|
|
err = check_weight(col)
|
|
if err: return err
|
|
|
|
for col in ['TIPO MONEDA']:
|
|
err = check_currency(col)
|
|
if err: return err
|
|
|
|
for col in ['TIPO TRANSPORTE']:
|
|
err = check_transport_type(col)
|
|
if err: return err
|
|
|
|
# B. Invoice Details (Parts)
|
|
elif target == 'invoice_details':
|
|
# 1. Line Number
|
|
if not row.get('LINEA') and not row.get('RENGLON') and not row.get('PARTIDA'):
|
|
return {"line": line_num, "col": "LINEA", "msg": "Requerido"}
|
|
|
|
# 2. Parent Link (Invoice Number)
|
|
if not (row.get('NUMERO FACTURA') or row.get('NUM FACTURA') or row.get('FACTURA')):
|
|
return {"line": line_num, "col": "NUMERO FACTURA", "msg": "Requerido"}
|
|
|
|
# 3. Numeric Fields
|
|
for col in ['PRECIO UNITARIO', 'PRECIOUNITARIO', 'VALOR COMERCIAL', 'VALORCOMERCIAL', 'CANTIDAD']:
|
|
err = check_decimal(col)
|
|
if err: return err
|
|
|
|
for col in ['CANTIDAD BULTOS', 'CANTIDADBULTOS', 'LINEA', 'RENGLON', 'PARTIDA']:
|
|
err = check_int(col)
|
|
if err: return err
|
|
|
|
return None
|
|
|
|
|
|
def validate_row_strict(
|
|
row: Dict[str, Any],
|
|
target: str,
|
|
line_num: int,
|
|
date_format: Optional[str],
|
|
validator: ForeignKeyValidator,
|
|
inv_type_value: str,
|
|
invoice_id_cache: Dict[str, Optional[int]],
|
|
models: Dict[str, Any],
|
|
) -> Optional[Dict[str, Any]]:
|
|
err = validate_row_phase_1(row, target, line_num, date_format)
|
|
if err:
|
|
return err
|
|
|
|
InvoiceHeader = models["InvoiceHeader"]
|
|
InvoiceType = models["InvoiceType"]
|
|
ClientProvider = models["ClientProvider"]
|
|
CustomsBroker = models["CustomsBroker"]
|
|
RegimenPedimento = models["RegimenPedimento"]
|
|
CurrencyType = models["CurrencyType"]
|
|
CustomsSection = models["CustomsSection"]
|
|
Incoterm = models["Incoterm"]
|
|
Part = models["Part"]
|
|
|
|
if target == "invoice_header":
|
|
if not validator.check_exists(InvoiceType, inv_type_value, field_name="key", is_public=True):
|
|
return {"line": line_num, "col": "TIPO FACTURA", "msg": "No existe en el catalogo"}
|
|
|
|
provider_id = parse_int(row.get("CLAVE PROVEEDOR"))
|
|
err = validate_tenant_fk_id(validator, ClientProvider, provider_id, line_num, "CLAVE PROVEEDOR", required=True)
|
|
if err:
|
|
return err
|
|
|
|
sold_to_id = parse_int(row.get("CLAVE VENDIDO A"))
|
|
err = validate_tenant_fk_id(validator, ClientProvider, sold_to_id, line_num, "CLAVE VENDIDO A", required=True)
|
|
if err:
|
|
return err
|
|
|
|
shipped_to_id = parse_int(row.get("CLAVE ENVIADO A"))
|
|
err = validate_tenant_fk_id(validator, ClientProvider, shipped_to_id, line_num, "CLAVE ENVIADO A", required=True)
|
|
if err:
|
|
return err
|
|
|
|
broker_id = parse_int(row.get("AGENTE ADUANAL"))
|
|
err = validate_tenant_fk_id(validator, CustomsBroker, broker_id, line_num, "AGENTE ADUANAL")
|
|
if err:
|
|
return err
|
|
|
|
err = validate_public_code(
|
|
validator,
|
|
RegimenPedimento,
|
|
row.get("REGIMEN") or row.get("CLAVEDOCUMENTO"),
|
|
line_num,
|
|
"CLAVEDOCUMENTO",
|
|
)
|
|
if err:
|
|
return err
|
|
|
|
err = validate_public_code(
|
|
validator,
|
|
CustomsSection,
|
|
row.get("ADUANA DE CRUCE"),
|
|
line_num,
|
|
"ADUANA DE CRUCE",
|
|
field_name="customs_code",
|
|
)
|
|
if err:
|
|
return err
|
|
|
|
err = validate_public_code(
|
|
validator,
|
|
CurrencyType,
|
|
row.get("CLAVE MONEDA"),
|
|
line_num,
|
|
"CLAVE MONEDA",
|
|
)
|
|
if err:
|
|
return err
|
|
|
|
err = validate_public_code(
|
|
validator,
|
|
Incoterm,
|
|
row.get("CLAVE INCOTERM"),
|
|
line_num,
|
|
"CLAVE INCOTERM",
|
|
)
|
|
if err:
|
|
return err
|
|
|
|
elif target == "invoice_details":
|
|
invoice_number = (row.get("NUMERO FACTURA") or row.get("NUM FACTURA") or row.get("FACTURA") or "").strip()
|
|
if not invoice_number:
|
|
return {"line": line_num, "col": "NUMERO FACTURA", "msg": "Requerido"}
|
|
|
|
cache_key = f"{invoice_number}|{inv_type_value}"
|
|
if cache_key in invoice_id_cache:
|
|
invoice_id = invoice_id_cache[cache_key]
|
|
else:
|
|
invoice_id = (
|
|
validator.session.query(InvoiceHeader.id)
|
|
.filter(
|
|
InvoiceHeader.tenant_id == validator.tenant_id,
|
|
InvoiceHeader.company_id == validator.company_id,
|
|
InvoiceHeader.invoice_number == invoice_number,
|
|
InvoiceHeader.invoice_type == inv_type_value,
|
|
)
|
|
.scalar()
|
|
)
|
|
invoice_id_cache[cache_key] = invoice_id
|
|
if not invoice_id:
|
|
return {"line": line_num, "col": "NUMERO FACTURA", "msg": "Factura no existe"}
|
|
|
|
part_num = (row.get("NUMPARTE") or row.get("NUMERO PARTE") or "").strip()
|
|
if not part_num:
|
|
return {"line": line_num, "col": "NUMPARTE", "msg": "Requerido"}
|
|
if not validator.check_exists(Part, part_num, field_name="part_number"):
|
|
return {"line": line_num, "col": "NUMPARTE", "msg": "No existe en el catalogo"}
|
|
|
|
return None
|
|
|
|
def parse_footer_config(config: Optional[str]) -> Dict[str, Any]:
|
|
if not config:
|
|
return {}
|
|
try:
|
|
if isinstance(config, str):
|
|
return json.loads(config)
|
|
if isinstance(config, dict):
|
|
return config
|
|
except Exception:
|
|
return {}
|
|
return {}
|
|
|
|
|
|
def display_date_format(date_format: Optional[str]) -> str:
|
|
if not date_format:
|
|
return "YYYY-MM-DD"
|
|
return date_format.upper()
|
|
|
|
|
|
def parse_date(date_text: Optional[str], date_format: Optional[str]) -> Optional[datetime.date]:
|
|
if not date_text:
|
|
return None
|
|
candidates = []
|
|
fmt_map = {
|
|
"dd/mm/yyyy": "%d/%m/%Y",
|
|
"mm/dd/yyyy": "%m/%d/%Y",
|
|
"yyyy-mm-dd": "%Y-%m-%d",
|
|
}
|
|
if date_format and date_format in fmt_map:
|
|
candidates.append(fmt_map[date_format])
|
|
candidates.extend(["%Y-%m-%d", "%d/%m/%Y", "%m/%d/%Y"])
|
|
for fmt in candidates:
|
|
try:
|
|
return datetime.strptime(str(date_text).strip(), fmt).date()
|
|
except ValueError:
|
|
continue
|
|
return None
|
|
|
|
|
|
def is_valid_date(date_text: Optional[str], date_format: Optional[str]) -> bool:
|
|
return parse_date(date_text, date_format) is not None
|
|
|
|
|
|
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 normalize_row(row: Dict[str, Any]) -> Dict[str, Any]:
|
|
return {normalize_header(k): v for k, v in row.items()}
|
|
|
|
|
|
def parse_int(value: Any) -> Optional[int]:
|
|
if value is None:
|
|
return None
|
|
text = str(value).strip()
|
|
if not text:
|
|
return None
|
|
try:
|
|
return int(text)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def parse_decimal(value: Any) -> Optional[Decimal]:
|
|
if value is None:
|
|
return None
|
|
text = str(value).strip()
|
|
if not text:
|
|
return None
|
|
text = text.replace(",", "")
|
|
try:
|
|
return Decimal(text)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def decimal_or_zero(value: Any) -> Decimal:
|
|
"""Return parsed decimal or Decimal('0') for CSV nulls/empty (vanilla default)."""
|
|
return parse_decimal(value) or Decimal("0")
|
|
|
|
|
|
def int_or_zero(value: Any) -> int:
|
|
"""Return parsed int or 0 for CSV nulls/empty (vanilla default)."""
|
|
return parse_int(value) if parse_int(value) is not None else 0
|
|
|
|
|
|
def parse_currency(value: Optional[str], currency_type: Optional[str]):
|
|
from api.v1.modules.a76.invoices.models import Currency
|
|
if value:
|
|
normalized = normalize_header(value)
|
|
if normalized in {"MN", "M N", "NACIONAL", "LOCAL", "PESOS", "PESO"}:
|
|
return Currency.LOCAL
|
|
if normalized in {"ME", "M E", "EXTRANJERA", "EXTRANJERO", "FOREIGN", "USD", "DOLAR", "DOLARES"}:
|
|
return Currency.FOREIGN
|
|
if "MANUAL" in normalized:
|
|
return Currency.MANUAL
|
|
if currency_type and str(currency_type).strip().upper() == "MXN":
|
|
return Currency.LOCAL
|
|
if currency_type:
|
|
return Currency.FOREIGN
|
|
return Currency.MANUAL
|
|
|
|
|
|
def parse_weight_unit(value: Optional[str]):
|
|
from api.v1.modules.a76.invoices.models import WeightUnit
|
|
if not value:
|
|
return None
|
|
normalized = normalize_header(value)
|
|
if normalized in {"KG", "KGS", "KILOS", "KILOGRAMOS"}:
|
|
return WeightUnit.KGS
|
|
if normalized in {"LB", "LBS", "LIBRAS"}:
|
|
return WeightUnit.LBS
|
|
return None
|
|
|
|
|
|
def resolve_tenant_fk_id(
|
|
session: CoreSessionLocal,
|
|
model,
|
|
value: Optional[int],
|
|
tenant_id: int,
|
|
company_id: int,
|
|
cache: Dict[int, Optional[int]],
|
|
) -> Optional[int]:
|
|
if value is None:
|
|
return None
|
|
if value in cache:
|
|
return cache[value]
|
|
exists = (
|
|
session.query(model.id)
|
|
.filter(
|
|
model.id == value,
|
|
model.tenant_id == tenant_id,
|
|
model.company_id == company_id,
|
|
)
|
|
.scalar()
|
|
)
|
|
cache[value] = value if exists is not None else None
|
|
return cache[value]
|
|
|
|
|
|
def resolve_public_code(
|
|
session: CoreSessionLocal,
|
|
model,
|
|
column,
|
|
value: Optional[str],
|
|
cache: Dict[str, Optional[str]],
|
|
) -> Optional[str]:
|
|
if not value:
|
|
return None
|
|
normalized = str(value).strip().upper()
|
|
if not normalized:
|
|
return None
|
|
if normalized in cache:
|
|
return cache[normalized]
|
|
exists = session.query(column).filter(column == normalized).scalar()
|
|
cache[normalized] = normalized if exists is not None else None
|
|
return cache[normalized]
|
|
|
|
@celery_app.task(bind=True)
|
|
def insert_valid_rows(self, job_id: str, model_target: str):
|
|
"""
|
|
Pass 2: Re-read CSV, Skip Errors, Bulk Insert.
|
|
File and meta are loaded from Redis if present (same as scan_file), so worker does not need shared filesystem.
|
|
"""
|
|
logger.info(f"Starting Commit for {job_id} target {model_target}")
|
|
|
|
# Ensure we have the file on this worker: prefer Redis (so any worker can run commit)
|
|
file_path = _ensure_worker_has_file_from_redis(job_id)
|
|
if not file_path:
|
|
alt_path = common_storage.file_path_for_job(JOB_TYPE, job_id)
|
|
if not os.path.exists(alt_path):
|
|
return {"status": "failed", "error": "File not found (missing or expired). Please upload and confirm again."}
|
|
file_path = alt_path
|
|
else:
|
|
_ensure_worker_has_meta_from_redis(job_id, file_path)
|
|
|
|
try:
|
|
tenant_id, company_id = common_meta.require_tenant_context(file_path)
|
|
except ValueError as e:
|
|
return {"status": "failed", "error": str(e)}
|
|
|
|
meta = common_meta.load_meta(file_path)
|
|
meta_path = common_meta.get_meta_path(file_path)
|
|
error_path = common_storage.error_path_for_job(JOB_TYPE, job_id)
|
|
error_lines = common_storage.get_error_lines(JOB_TYPE, job_id, error_path)
|
|
|
|
# Si el upload fue de series (template_id imp_temp_series), usar flujo series aunque model_target venga mal
|
|
use_series_flow = (
|
|
model_target == "invoice_series"
|
|
or meta.get("template_id") == "imp_temp_series"
|
|
)
|
|
|
|
# --- Series de Importación Temporal: commit (INSERT/UPDATE item_line_series) ---
|
|
if use_series_flow:
|
|
try:
|
|
from api.v1.modules.a76.invoices.models import InvoiceHeader
|
|
from api.v1.modules.a76.items.models import LineItem
|
|
from api.v1.modules.a76.items.series.models import Serie
|
|
from .validators.series_impo_temp import (
|
|
validate_row_series_impo_temp,
|
|
row_to_series_normalized,
|
|
)
|
|
|
|
actualizar = meta.get("actualizar", False)
|
|
autonumerar = meta.get("autonumerar", True)
|
|
validar_series_exception = meta.get("validar_series", False)
|
|
_fc = parse_footer_config(meta.get("footer_config"))
|
|
if _fc:
|
|
if "actualizar" in _fc:
|
|
actualizar = bool(_fc["actualizar"])
|
|
elif _fc.get("mode") == "update":
|
|
actualizar = True
|
|
elif _fc.get("mode") == "replace":
|
|
actualizar = False
|
|
if "autonumerar" in _fc:
|
|
autonumerar = bool(_fc["autonumerar"])
|
|
else:
|
|
as_val = _fc.get("autonumber_series", "true")
|
|
autonumerar = str(as_val).lower() in ("true", "1", "si", "sí", "yes")
|
|
if "validar_series" in _fc:
|
|
validar_series_exception = bool(_fc["validar_series"])
|
|
|
|
with CoreSessionLocal() as session:
|
|
q = (
|
|
session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.is_updated)
|
|
.filter(
|
|
InvoiceHeader.tenant_id == tenant_id,
|
|
InvoiceHeader.company_id == company_id,
|
|
InvoiceHeader.operation_type == "imp",
|
|
InvoiceHeader.invoice_type == "TEM",
|
|
)
|
|
)
|
|
rows_inv = q.all()
|
|
invoice_id_by_number: Dict[str, int] = {}
|
|
invoice_updated_by_number: Dict[str, bool] = {}
|
|
for num, iid, is_upd in rows_inv:
|
|
if num:
|
|
invoice_id_by_number[str(num).strip()] = iid
|
|
invoice_updated_by_number[str(num).strip()] = bool(is_upd)
|
|
|
|
existing_series_keys: Set[Tuple[str, str, str]] = set()
|
|
existing_series_data: Dict[Tuple[str, str, str], Dict[str, Any]] = {}
|
|
if actualizar and not autonumerar:
|
|
q_ser = (
|
|
session.query(
|
|
InvoiceHeader.invoice_number,
|
|
LineItem.line_number,
|
|
Serie.row,
|
|
Serie.serial_numbers,
|
|
Serie.model,
|
|
Serie.sub_model,
|
|
Serie.number_id,
|
|
)
|
|
.join(LineItem, LineItem.invoice_id == InvoiceHeader.id)
|
|
.join(Serie, Serie.line_item_id == LineItem.id)
|
|
.filter(
|
|
InvoiceHeader.tenant_id == tenant_id,
|
|
InvoiceHeader.company_id == company_id,
|
|
InvoiceHeader.operation_type == "imp",
|
|
InvoiceHeader.invoice_type == "TEM",
|
|
)
|
|
)
|
|
for num, ln, rw, sn, md, sm, nid in q_ser.all():
|
|
if num is not None:
|
|
k = (str(num).strip(), str(ln).strip(), str(rw).strip())
|
|
existing_series_keys.add(k)
|
|
existing_series_data.setdefault(k, {
|
|
"serial_numbers": sn or "",
|
|
"model": md or "",
|
|
"sub_model": sm or "",
|
|
"number_id": nid or "",
|
|
})
|
|
|
|
inserted_count = 0
|
|
updated_count = 0
|
|
skipped_invalid = 0
|
|
skipped_details: List[Dict[str, Any]] = []
|
|
|
|
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, "imp_temp_series", normalize_header)
|
|
err = validate_row_series_impo_temp(
|
|
row_norm,
|
|
i,
|
|
actualizar=actualizar,
|
|
autonumerar=autonumerar,
|
|
validar_series_exception=validar_series_exception,
|
|
invoice_id_by_number=invoice_id_by_number,
|
|
invoice_updated_by_number=invoice_updated_by_number,
|
|
existing_series_keys=existing_series_keys,
|
|
existing_series_data=existing_series_data,
|
|
warnings=None,
|
|
)
|
|
if err:
|
|
skipped_invalid += 1
|
|
skipped_details.append({
|
|
"line": i,
|
|
"invoice": (row_norm.get("NUMERO FACTURA") or row_norm.get("NUM FACTURA") or "").strip(),
|
|
"reason": err.get("msg", ""),
|
|
})
|
|
continue
|
|
|
|
data = row_to_series_normalized(row_norm)
|
|
invoice_number = data["NUMERO FACTURA"]
|
|
linea_factura = data["LINEA FACTURA"]
|
|
linea_serie = data["LINEA SERIE"]
|
|
|
|
if not invoice_number or invoice_number not in invoice_id_by_number:
|
|
skipped_invalid += 1
|
|
continue
|
|
invoice_id = invoice_id_by_number[invoice_number]
|
|
line_number_val = parse_int(linea_factura)
|
|
if line_number_val is None:
|
|
skipped_invalid += 1
|
|
skipped_details.append({"line": i, "invoice": invoice_number, "reason": "LINEA FACTURA debe ser numérico."})
|
|
continue
|
|
line_item = (
|
|
session.query(LineItem)
|
|
.filter(
|
|
LineItem.invoice_id == invoice_id,
|
|
LineItem.line_number == line_number_val,
|
|
LineItem.tenant_id == tenant_id,
|
|
LineItem.company_id == company_id,
|
|
)
|
|
.first()
|
|
)
|
|
if not line_item:
|
|
skipped_invalid += 1
|
|
skipped_details.append({
|
|
"line": i,
|
|
"invoice": invoice_number,
|
|
"reason": f"Partida línea {linea_factura} no existe en la factura.",
|
|
})
|
|
continue
|
|
|
|
if autonumerar:
|
|
max_row = (
|
|
session.query(Serie.row)
|
|
.filter(Serie.line_item_id == line_item.id)
|
|
.order_by(Serie.row.desc())
|
|
.limit(1)
|
|
.scalar()
|
|
)
|
|
row_num = (max_row or 0) + 1
|
|
else:
|
|
row_num = parse_int(linea_serie)
|
|
if row_num is None:
|
|
skipped_invalid += 1
|
|
skipped_details.append({"line": i, "invoice": invoice_number, "reason": "LINEA SERIE debe ser numérico."})
|
|
continue
|
|
|
|
existing_serie = (
|
|
session.query(Serie)
|
|
.filter(
|
|
Serie.line_item_id == line_item.id,
|
|
Serie.row == row_num,
|
|
)
|
|
.first()
|
|
)
|
|
|
|
if existing_serie:
|
|
if actualizar:
|
|
existing_serie.serial_numbers = data["SERIE"] or existing_serie.serial_numbers
|
|
existing_serie.model = data["MODELO"] or existing_serie.model
|
|
existing_serie.sub_model = data["SUB MODELO"] or existing_serie.sub_model
|
|
existing_serie.number_id = data["NUMERO ID"] or existing_serie.number_id
|
|
session.add(existing_serie)
|
|
updated_count += 1
|
|
else:
|
|
skipped_invalid += 1
|
|
skipped_details.append({"line": i, "invoice": invoice_number, "reason": "Serie ya existe (use actualizar)."})
|
|
else:
|
|
new_serie = Serie(
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
line_item_id=line_item.id,
|
|
row=row_num,
|
|
serial_numbers=data["SERIE"] or None,
|
|
model=data["MODELO"] or None,
|
|
sub_model=data["SUB MODELO"] or None,
|
|
number_id=data["NUMERO ID"] or None,
|
|
)
|
|
session.add(new_serie)
|
|
inserted_count += 1
|
|
|
|
session.commit()
|
|
|
|
common_storage.cleanup_import_job(JOB_TYPE, job_id, file_path=file_path, error_path=error_path, meta_path=meta_path)
|
|
status = "finished" if (inserted_count + updated_count) > 0 else ("warning" if skipped_invalid else "failed")
|
|
out = {
|
|
"status": status,
|
|
"inserted": inserted_count,
|
|
"updated": updated_count,
|
|
"skipped_invalid": skipped_invalid,
|
|
"skipped_missing_fk": 0,
|
|
"skipped_duplicate": 0,
|
|
"skipped_details": skipped_details,
|
|
}
|
|
if status == "failed":
|
|
out["error"] = "No hay registros válidos en el archivo CSV."
|
|
elif status == "warning" and skipped_invalid:
|
|
out["message"] = f"No se insertaron registros. {skipped_invalid} fueron rechazados."
|
|
return out
|
|
except Exception as e:
|
|
logger.exception("Series import commit failed: %s", e)
|
|
return {"status": "failed", "error": str(e)}
|
|
|
|
try:
|
|
from api.v1.modules.a76.invoices.models import (
|
|
InvoiceHeader,
|
|
InvoiceComplianceMx,
|
|
InvoiceFinancials,
|
|
InvoiceLogistics,
|
|
InvoiceSalesDetails,
|
|
OperationType,
|
|
TransportType,
|
|
WeightUnit,
|
|
)
|
|
from api.v1.modules.a76.clients_and_providers.models import ClientProvider
|
|
from api.v1.modules.a76.customs_brokers.models import CustomsBroker
|
|
from api.v1.modules.public.reference_data.currency_types.models import CurrencyType
|
|
from api.v1.modules.public.reference_data.pedimento_codes.models import PedimentoCode
|
|
from api.v1.modules.public.reference_data.code_pedimento_regimens.models import CodePedimentoRegimen
|
|
from api.v1.modules.public.reference_data.pedimento_regimens.models import RegimenPedimento
|
|
from api.v1.modules.public.reference_data.invoice_types.models import InvoiceType
|
|
from api.v1.modules.public.reference_data.customs_sections.models import CustomsSection
|
|
from api.v1.modules.public.reference_data.incoterms.models import Incoterm
|
|
|
|
from api.v1.modules.a76.items.models import LineItem
|
|
from api.v1.modules.a76.items.line_financials.models import LineFinancial
|
|
from api.v1.modules.a76.items.line_quantities.models import LineQuantity
|
|
from api.v1.modules.a76.items.line_customs.models import LineCustom
|
|
from api.v1.modules.a76.items.line_descriptions.models import LineDescription
|
|
from api.v1.modules.a76.parts.models import Part
|
|
|
|
footer_config = parse_footer_config(meta.get("footer_config"))
|
|
|
|
date_format = footer_config.get("dateFormat")
|
|
# Validate and set default date_format if not provided
|
|
if not date_format:
|
|
date_format = "yyyy-mm-dd" # Default to ISO format
|
|
logger.info(f"No date_format specified in config, using default: {date_format}")
|
|
else:
|
|
logger.info(f"Using date_format from config: {date_format}")
|
|
|
|
# Default types from config or fallback
|
|
op_type_value = OperationType(meta.get('operation_type', 'imp').lower())
|
|
inv_type_value = normalize_public_code(footer_config.get('invoice_type') or 'TEM') or 'TEM'
|
|
|
|
logger.info(f"Processing CSV with operation_type={op_type_value}, invoice_type={inv_type_value}, date_format={date_format}")
|
|
|
|
headers_to_insert = []
|
|
details_to_insert = []
|
|
skipped_invalid = 0
|
|
skipped_missing_invoice = 0
|
|
skipped_missing_fk = 0
|
|
skipped_fk_details = []
|
|
inserted_count = 0
|
|
response = None
|
|
|
|
with CoreSessionLocal() as session:
|
|
invoice_id_cache = {}
|
|
cleared_invoices = set() # Track invoices where we've already cleared items in this job
|
|
provider_cache: Dict[int, Optional[int]] = {}
|
|
sold_to_cache: Dict[int, Optional[int]] = {}
|
|
shipped_to_cache: Dict[int, Optional[int]] = {}
|
|
broker_cache: Dict[int, Optional[int]] = {}
|
|
regimen_cache: Dict[str, Optional[str]] = {}
|
|
currency_type_cache: Dict[str, Optional[str]] = {}
|
|
customs_section_cache: Dict[str, Optional[str]] = {}
|
|
part_cache: Dict[str, Optional[int]] = {}
|
|
|
|
validator = ForeignKeyValidator(session, tenant_id, company_id)
|
|
|
|
with open(file_path, 'r', encoding='utf-8-sig') as f:
|
|
# Detect Delimiter
|
|
sample = f.read(2048)
|
|
f.seek(0)
|
|
try:
|
|
dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
|
|
except:
|
|
dialect = 'excel'
|
|
|
|
reader = csv.DictReader(f, dialect=dialect)
|
|
|
|
template_id = meta.get("template_id") or (
|
|
"imp_temp_header" if model_target == "invoice_header" else "imp_temp_details"
|
|
)
|
|
|
|
for i, row in enumerate(reader, start=1):
|
|
if i in error_lines:
|
|
continue
|
|
|
|
row_norm = row_from_template(row, template_id, normalize_header)
|
|
|
|
# Mapping Logic (solo campos que acepta el modelo de facturas)
|
|
if model_target == 'invoice_header':
|
|
invoice_number = (row_norm.get('NUMERO FACTURA') or row_norm.get('NUM FACTURA') or row_norm.get('FACTURA') or '').strip()
|
|
invoice_date = parse_date(row_norm.get('FECHA FACTURA') or row_norm.get('FECHA'), date_format)
|
|
|
|
if not invoice_number or not invoice_date:
|
|
skipped_invalid += 1
|
|
logger.debug(f"Row {i}: Skipped - missing invoice_number or invalid invoice_date. "
|
|
f"Invoice: {invoice_number}, Date: {row_norm.get('FECHA FACTURA') or row_norm.get('FECHA')}")
|
|
continue
|
|
|
|
# --- NEW: Foreign Key Validations ---
|
|
# 1. Invoice Type (Public)
|
|
if not validator.check_exists(InvoiceType, inv_type_value, field_name="key", is_public=True):
|
|
skipped_missing_fk += 1
|
|
reason = f"Tipo de factura '{inv_type_value}' no existe"
|
|
skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
|
|
logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
|
|
continue
|
|
|
|
provider_id = parse_int(row_norm.get('CLAVE PROVEEDOR'))
|
|
err = validate_tenant_fk_id(
|
|
validator,
|
|
ClientProvider,
|
|
provider_id,
|
|
i,
|
|
"CLAVE PROVEEDOR",
|
|
required=True,
|
|
)
|
|
if err:
|
|
skipped_invalid += 1
|
|
reason = f"{err['col']}: {err['msg']}"
|
|
skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
|
|
logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
|
|
continue
|
|
|
|
sold_to_id = parse_int(row_norm.get('CLAVE VENDIDO A'))
|
|
err = validate_tenant_fk_id(
|
|
validator,
|
|
ClientProvider,
|
|
sold_to_id,
|
|
i,
|
|
"CLAVE VENDIDO A",
|
|
required=True,
|
|
)
|
|
if err:
|
|
skipped_invalid += 1
|
|
reason = f"{err['col']}: {err['msg']}"
|
|
skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
|
|
logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
|
|
continue
|
|
|
|
shipped_to_id = parse_int(row_norm.get('CLAVE ENVIADO A'))
|
|
err = validate_tenant_fk_id(
|
|
validator,
|
|
ClientProvider,
|
|
shipped_to_id,
|
|
i,
|
|
"CLAVE ENVIADO A",
|
|
required=True,
|
|
)
|
|
if err:
|
|
skipped_invalid += 1
|
|
reason = f"{err['col']}: {err['msg']}"
|
|
skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
|
|
logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
|
|
continue
|
|
|
|
broker_id = parse_int(row_norm.get('AGENTE ADUANAL'))
|
|
err = validate_tenant_fk_id(
|
|
validator,
|
|
CustomsBroker,
|
|
broker_id,
|
|
i,
|
|
"AGENTE ADUANAL",
|
|
)
|
|
if err:
|
|
skipped_missing_fk += 1
|
|
reason = f"{err['col']}: {err['msg']}"
|
|
skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
|
|
logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
|
|
continue
|
|
|
|
err = validate_public_code(
|
|
validator,
|
|
RegimenPedimento,
|
|
row_norm.get('REGIMEN') or row_norm.get('CLAVEDOCUMENTO'),
|
|
i,
|
|
"CLAVEDOCUMENTO",
|
|
)
|
|
if err:
|
|
skipped_missing_fk += 1
|
|
reason = f"{err['col']}: {err['msg']}"
|
|
skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
|
|
logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
|
|
continue
|
|
|
|
err = validate_public_code(
|
|
validator,
|
|
CustomsSection,
|
|
row_norm.get('ADUANA DE CRUCE'),
|
|
i,
|
|
"ADUANA DE CRUCE",
|
|
field_name="customs_code",
|
|
)
|
|
if err:
|
|
skipped_missing_fk += 1
|
|
reason = f"{err['col']}: {err['msg']}"
|
|
skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
|
|
logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
|
|
continue
|
|
|
|
err = validate_public_code(
|
|
validator,
|
|
CurrencyType,
|
|
row_norm.get('CLAVE MONEDA'),
|
|
i,
|
|
"CLAVE MONEDA",
|
|
)
|
|
if err:
|
|
skipped_missing_fk += 1
|
|
reason = f"{err['col']}: {err['msg']}"
|
|
skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
|
|
logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
|
|
continue
|
|
|
|
err = validate_public_code(
|
|
validator,
|
|
Incoterm,
|
|
row_norm.get('CLAVE INCOTERM'),
|
|
i,
|
|
"CLAVE INCOTERM",
|
|
)
|
|
if err:
|
|
skipped_missing_fk += 1
|
|
reason = f"{err['col']}: {err['msg']}"
|
|
skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
|
|
logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
|
|
continue
|
|
|
|
transport_type_val = row_norm.get('TIPO TRANSPORTE')
|
|
if transport_type_val and str(transport_type_val).strip().lower() not in TRANSPORT_TYPE_VALUES:
|
|
skipped_invalid += 1
|
|
reason = "TIPO TRANSPORTE: Tipo de transporte invalido"
|
|
skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
|
|
logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
|
|
continue
|
|
|
|
currency_val = row_norm.get('TIPO MONEDA')
|
|
if currency_val and str(currency_val).strip():
|
|
parsed_currency = parse_currency(currency_val, None)
|
|
val_norm = normalize_header(currency_val)
|
|
if parsed_currency.value == "manual" and "MANUAL" not in val_norm:
|
|
skipped_invalid += 1
|
|
reason = "TIPO MONEDA: Moneda invalida"
|
|
skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
|
|
logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
|
|
continue
|
|
|
|
# 2. Client/Provider and broker checks are handled above
|
|
|
|
# --- 4. Check for Existing Invoice (Upsert Logic) ---
|
|
existing_header = None
|
|
if invoice_number:
|
|
existing_header = (
|
|
session.query(InvoiceHeader)
|
|
.filter(
|
|
InvoiceHeader.tenant_id == tenant_id,
|
|
InvoiceHeader.company_id == company_id,
|
|
InvoiceHeader.invoice_number == invoice_number,
|
|
InvoiceHeader.invoice_type == inv_type_value
|
|
)
|
|
.first()
|
|
)
|
|
|
|
if existing_header:
|
|
# UPDATE existing header
|
|
header = existing_header
|
|
header.invoice_date = invoice_date
|
|
header.operation_type = op_type_value
|
|
header.is_updated = True # Mark as updated
|
|
header.updated_date = datetime.utcnow()
|
|
header.document_type = resolve_public_code(
|
|
session,
|
|
RegimenPedimento,
|
|
RegimenPedimento.code,
|
|
(row_norm.get('REGIMEN') or row_norm.get('CLAVEDOCUMENTO')),
|
|
regimen_cache,
|
|
)
|
|
header.project_number = (row_norm.get('NUM PROYECTO') or row_norm.get('NUMPROYECTO') or None)
|
|
header.purchase_order = (row_norm.get('ORDEN COMPRA') or row_norm.get('ORDENCOMPRA') or None)
|
|
header.alternate_invoice = (row_norm.get('FACTURA ALTERNA') or None)
|
|
header.invoice_ref = (row_norm.get('FACTURA EXPO REF') or row_norm.get('FACTURAEXPOREF') or None)
|
|
header.emission_date = parse_date(row_norm.get('FECHA EMISION'), date_format)
|
|
header.observation_es = (row_norm.get('OBSERVACIONES E') or None)
|
|
header.observation_en = (row_norm.get('OBSERVACIONES I') or None)
|
|
|
|
logger.info(f"Row {i}: Updating existing invoice {invoice_number}")
|
|
|
|
# Clean up related data that will be re-inserted/updated
|
|
# Note: compliance, financials, logistics are 1-to-1 relationships and will be updated by assignment below
|
|
# but we might want to be explicit if ORM doesn't handle replace well.
|
|
# SQLAlchemy relationship assignment usually handles 1-to-1 updates correctly.
|
|
|
|
else:
|
|
# CREATE new header
|
|
header = InvoiceHeader(
|
|
invoice_number=invoice_number,
|
|
invoice_date=invoice_date,
|
|
operation_type=op_type_value,
|
|
is_updated=False,
|
|
system="CSV",
|
|
capture_date=datetime.utcnow(),
|
|
invoice_type=inv_type_value,
|
|
document_type=resolve_public_code(
|
|
session,
|
|
RegimenPedimento,
|
|
RegimenPedimento.code,
|
|
(row_norm.get('REGIMEN') or row_norm.get('CLAVEDOCUMENTO')),
|
|
regimen_cache,
|
|
),
|
|
project_number=(row_norm.get('NUM PROYECTO') or row_norm.get('NUMPROYECTO') or None),
|
|
purchase_order=(row_norm.get('ORDEN COMPRA') or row_norm.get('ORDENCOMPRA') or None),
|
|
alternate_invoice=(row_norm.get('FACTURA ALTERNA') or None),
|
|
invoice_ref=(row_norm.get('FACTURA EXPO REF') or row_norm.get('FACTURAEXPOREF') or None),
|
|
emission_date=parse_date(row_norm.get('FECHA EMISION'), date_format),
|
|
observation_es=(row_norm.get('OBSERVACIONES E') or None),
|
|
observation_en=(row_norm.get('OBSERVACIONES I') or None),
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
)
|
|
|
|
compliance = InvoiceComplianceMx(
|
|
remesa=parse_int(row_norm.get('REMESA')),
|
|
aduana=resolve_public_code(
|
|
session,
|
|
CustomsSection,
|
|
CustomsSection.customs_code,
|
|
row_norm.get('ADUANA DE CRUCE'),
|
|
customs_section_cache,
|
|
),
|
|
provider_id=resolve_tenant_fk_id(
|
|
session,
|
|
ClientProvider,
|
|
parse_int(row_norm.get('CLAVE PROVEEDOR')),
|
|
tenant_id,
|
|
company_id,
|
|
provider_cache,
|
|
),
|
|
sold_to_id=resolve_tenant_fk_id(
|
|
session,
|
|
ClientProvider,
|
|
parse_int(row_norm.get('CLAVE VENDIDO A')),
|
|
tenant_id,
|
|
company_id,
|
|
sold_to_cache,
|
|
),
|
|
shipped_to_id=resolve_tenant_fk_id(
|
|
session,
|
|
ClientProvider,
|
|
parse_int(row_norm.get('CLAVE ENVIADO A')),
|
|
tenant_id,
|
|
company_id,
|
|
shipped_to_cache,
|
|
),
|
|
customs_broker_id=resolve_tenant_fk_id(
|
|
session,
|
|
CustomsBroker,
|
|
parse_int(row_norm.get('AGENTE ADUANAL')),
|
|
tenant_id,
|
|
company_id,
|
|
broker_cache,
|
|
),
|
|
edocument=(row_norm.get('E DOCUMENT') or None),
|
|
vucem_operation_num=(row_norm.get('NUM OPERACION') or None),
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
)
|
|
|
|
financials_currency_type = resolve_public_code(
|
|
session,
|
|
CurrencyType,
|
|
CurrencyType.code,
|
|
row_norm.get('CLAVE MONEDA'),
|
|
currency_type_cache,
|
|
)
|
|
financials = InvoiceFinancials(
|
|
currency=parse_currency(row_norm.get('TIPO MONEDA'), financials_currency_type),
|
|
currency_type=financials_currency_type,
|
|
exchange_rate=decimal_or_zero(row_norm.get('TIPO DE CAMBIO')),
|
|
freight=decimal_or_zero(row_norm.get('FLETES')),
|
|
insurance_value=decimal_or_zero(row_norm.get('VALOR SEGUROS')),
|
|
insurance=decimal_or_zero(row_norm.get('SEGUROS')),
|
|
packaging=decimal_or_zero(row_norm.get('EMBALAJES')),
|
|
other_increments=decimal_or_zero(row_norm.get('OTROS INCREMENTABLES')),
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
)
|
|
|
|
weight_type = parse_weight_unit(row_norm.get('TIPO PESO'))
|
|
logistics = None
|
|
if weight_type or row_norm.get('TIPO TRANSPORTE') or row_norm.get('NUMERO TRANSPORTE'):
|
|
raw_transport = (row_norm.get('TIPO TRANSPORTE') or "none")
|
|
transport_str = str(raw_transport).strip().lower() or "none"
|
|
try:
|
|
transport_type = TransportType(transport_str)
|
|
except ValueError:
|
|
transport_type = TransportType.NONE
|
|
logistics = InvoiceLogistics(
|
|
carrier_id=(row_norm.get('CLAVE TRANSPORTISTA') or None),
|
|
driver_name=(row_norm.get('NOMBRE CONDUCTOR') or None),
|
|
transport_type=transport_type,
|
|
transport_num=(row_norm.get('NUMERO TRANSPORTE') or None),
|
|
weight_type=weight_type or WeightUnit.KGS,
|
|
seal_number=(row_norm.get('PRECINTO') or None),
|
|
incoterm=(row_norm.get('CLAVE INCOTERM') or None),
|
|
entry_exit_date=parse_date(row_norm.get('FECHA EMISION'), date_format),
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
)
|
|
|
|
header.compliance_mx = compliance
|
|
header.financials = financials
|
|
if logistics:
|
|
header.logistics = logistics
|
|
|
|
headers_to_insert.append(header)
|
|
|
|
elif model_target == 'invoice_details':
|
|
invoice_number = (row_norm.get('NUMERO FACTURA') or row_norm.get('NUM FACTURA') or '').strip()
|
|
if not invoice_number:
|
|
skipped_invalid += 1
|
|
continue
|
|
|
|
cache_key = f"{invoice_number}|{inv_type_value}"
|
|
if cache_key in invoice_id_cache:
|
|
invoice_id = invoice_id_cache[cache_key]
|
|
else:
|
|
invoice_id = (
|
|
session.query(InvoiceHeader.id)
|
|
.filter(
|
|
InvoiceHeader.tenant_id == tenant_id,
|
|
InvoiceHeader.company_id == company_id,
|
|
InvoiceHeader.invoice_number == invoice_number,
|
|
InvoiceHeader.invoice_type == inv_type_value,
|
|
)
|
|
.scalar()
|
|
)
|
|
invoice_id_cache[cache_key] = invoice_id
|
|
|
|
if not invoice_id:
|
|
logger.warning(
|
|
"Invoice not found for details row %s (invoice_number=%s)",
|
|
i,
|
|
invoice_number,
|
|
)
|
|
skipped_missing_invoice += 1
|
|
continue
|
|
|
|
part_num = (row_norm.get('NUMPARTE') or row_norm.get('NUMERO PARTE') or '').strip()
|
|
if not part_num:
|
|
skipped_invalid += 1
|
|
reason = "NUMPARTE: Requerido"
|
|
skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
|
|
logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
|
|
continue
|
|
if not validator.check_exists(Part, part_num, field_name="part_number"):
|
|
skipped_missing_fk += 1
|
|
reason = f"NUMPARTE '{part_num}' no existe"
|
|
skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
|
|
logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
|
|
continue
|
|
|
|
# --- Prevent Duplicates: Clear existing items for this invoice (Once per job) ---
|
|
if invoice_id not in cleared_invoices:
|
|
logger.info(f"Clearing existing details for Invoice {invoice_number} (ID: {invoice_id}) to prevent duplicates")
|
|
|
|
# 1. Delete Items (Cascades to LineItem, LineFinancial, etc. if DB configured, check models)
|
|
# Checking Item model, we usually need to be careful.
|
|
# Assuming Cascade delete is set up on FKs or we rely on ORM cascade if using relationships.
|
|
# Here we use bulk delete.
|
|
session.query(Item).filter(Item.invoice_id == invoice_id).delete(synchronize_session=False)
|
|
|
|
# 2. Delete InvoiceSalesDetails
|
|
session.query(InvoiceSalesDetails).filter(InvoiceSalesDetails.invoice_id == invoice_id).delete(synchronize_session=False)
|
|
|
|
cleared_invoices.add(invoice_id)
|
|
|
|
# --- NEW LOGIC: Expanded Anexo 76 Structure ---
|
|
|
|
# A. Find/Cache Part
|
|
part_id = None
|
|
part_id = part_cache.get(part_num)
|
|
if part_id is None:
|
|
p = session.query(Part.id).filter(
|
|
Part.part_number == part_num,
|
|
Part.tenant_id == tenant_id,
|
|
Part.company_id == company_id
|
|
).first()
|
|
if p:
|
|
part_id = p.id
|
|
part_cache[part_num] = part_id
|
|
|
|
line_num_val = (row_norm.get('LINEA') or row_norm.get('RENGLON') or row_norm.get('PARTIDA'))
|
|
line_num = parse_int(line_num_val) or (len(details_to_insert) + 1)
|
|
|
|
# 1. Parent Item
|
|
item = Item(
|
|
invoice_id=invoice_id,
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
item_type="N", # Default to Normal
|
|
system_origin="CSV"
|
|
)
|
|
session.add(item)
|
|
session.flush() # Need item.id
|
|
|
|
# 2. Main Line
|
|
line = LineItem(
|
|
item_id=item.id,
|
|
line_number=line_num,
|
|
part_number=part_id,
|
|
tenant_id=tenant_id,
|
|
company_id=company_id
|
|
)
|
|
session.add(line)
|
|
session.flush() # Need line.id
|
|
|
|
# 3. Financial Data (vanilla: nulls from CSV -> 0)
|
|
price = parse_decimal(row_norm.get('PRECIO UNITARIO') or row_norm.get('PRECIOUNITARIO'))
|
|
val_com = parse_decimal(row_norm.get('VALOR COMERCIAL') or row_norm.get('VALORCOMERCIAL'))
|
|
qty = parse_decimal(row_norm.get('CANTIDAD'))
|
|
commercial_total = val_com or (price * qty if price and qty else None)
|
|
|
|
session.add(LineFinancial(
|
|
item_line_id=line.id,
|
|
unit_cost_capture=decimal_or_zero(price),
|
|
total_commercial_value=decimal_or_zero(commercial_total),
|
|
))
|
|
|
|
# 4. Quantities (vanilla: nulls -> 0 so we always have a quantity row)
|
|
session.add(LineQuantity(
|
|
item_line_id=line.id,
|
|
quantity=decimal_or_zero(qty),
|
|
))
|
|
|
|
# 5. Customs/Fraction
|
|
origin = row_norm.get('PAIS ORIGEN') or row_norm.get('PAISORIGEN')
|
|
fraction = row_norm.get('FRACCION')
|
|
if origin or fraction:
|
|
session.add(LineCustom(
|
|
item_line_id=line.id,
|
|
fraction=fraction,
|
|
origin_country=origin,
|
|
))
|
|
|
|
# 6. Description
|
|
desc = row_norm.get('DESCRIPCION')
|
|
if desc:
|
|
session.add(LineDescription(
|
|
item_line_id=line.id,
|
|
description_spanish=desc,
|
|
))
|
|
|
|
# 7. Legacy Sales Details (For specific audit/UI fields; vanilla: nulls -> 0)
|
|
detail = InvoiceSalesDetails(
|
|
invoice_id=invoice_id,
|
|
line_number=line_num,
|
|
sales_order=(row_norm.get('ORDEN DE COMPRA') or row_norm.get('ORDENCOMPRA') or None),
|
|
line_bundles=int_or_zero(row_norm.get('CANTIDAD BULTOS') or row_norm.get('CANTIDADBULTOS')),
|
|
tenant_id=tenant_id,
|
|
company_id=company_id,
|
|
)
|
|
session.add(detail)
|
|
details_to_insert.append(item) # Use as counter/ref
|
|
|
|
# 3. Bulk Insert (ORM Transaction)
|
|
try:
|
|
if model_target == 'invoice_header':
|
|
if headers_to_insert:
|
|
logger.info(f"Attempting to commit {len(headers_to_insert)} headers")
|
|
session.add_all(headers_to_insert)
|
|
session.commit()
|
|
inserted_count = len(headers_to_insert)
|
|
logger.info(f"Headers commit successful. Inserted: {inserted_count}")
|
|
else:
|
|
logger.warning(f"No headers to insert for job {job_id}")
|
|
else:
|
|
if details_to_insert:
|
|
logger.info(f"Attempting to commit {len(details_to_insert)} items and related data")
|
|
session.commit() # Everything was already added with session.add()
|
|
inserted_count = len(details_to_insert)
|
|
logger.info(f"Details commit successful. Inserted: {inserted_count}")
|
|
else:
|
|
logger.warning(f"No details to insert for job {job_id}")
|
|
|
|
except Exception as db_err:
|
|
session.rollback()
|
|
logger.error(f"DB Error during {model_target} commit: {db_err}")
|
|
import traceback
|
|
logger.error(traceback.format_exc())
|
|
return {"status": "failed", "error": str(db_err)}
|
|
|
|
# 4. Determine final status and prepare response (inside session block to access variables)
|
|
total_skipped = skipped_invalid + skipped_missing_fk + skipped_missing_invoice
|
|
|
|
# Log summary
|
|
logger.info(f"Job {job_id} completed. Inserted: {inserted_count}, Skipped: {total_skipped} "
|
|
f"(invalid: {skipped_invalid}, missing_fk: {skipped_missing_fk}, missing_invoice: {skipped_missing_invoice})")
|
|
|
|
# Prepare response based on results
|
|
if inserted_count == 0:
|
|
if total_skipped > 0:
|
|
logger.warning(f"No valid records to insert for job {job_id}. All {total_skipped} records were rejected.")
|
|
response = {
|
|
"status": "warning",
|
|
"inserted": 0,
|
|
"skipped_invalid": skipped_invalid,
|
|
"skipped_missing_invoice": skipped_missing_invoice,
|
|
"skipped_missing_fk": skipped_missing_fk,
|
|
"skipped_details": skipped_fk_details,
|
|
"message": f"No se insertaron registros. {total_skipped} fueron rechazados."
|
|
}
|
|
else:
|
|
logger.error(f"No valid records found in CSV for job {job_id}")
|
|
response = {
|
|
"status": "failed",
|
|
"error": "No hay registros válidos en el archivo CSV",
|
|
"inserted": 0,
|
|
"skipped_invalid": skipped_invalid,
|
|
"skipped_missing_invoice": skipped_missing_invoice,
|
|
"skipped_missing_fk": skipped_missing_fk,
|
|
"skipped_details": skipped_fk_details
|
|
}
|
|
else:
|
|
# Success case - at least some records were inserted
|
|
response = {
|
|
"status": "finished",
|
|
"inserted": inserted_count,
|
|
"skipped_invalid": skipped_invalid,
|
|
"skipped_missing_invoice": skipped_missing_invoice,
|
|
"skipped_missing_fk": skipped_missing_fk,
|
|
"skipped_details": skipped_fk_details
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Task failed: {e}")
|
|
import traceback
|
|
logger.error(traceback.format_exc())
|
|
return {"status": "failed", "error": str(e)}
|
|
|
|
# 5. Cleanup: remove temp files and Redis keys so data is not kept indefinitely
|
|
try:
|
|
common_storage.cleanup_import_job(
|
|
JOB_TYPE, job_id,
|
|
file_path=file_path,
|
|
error_path=error_path,
|
|
meta_path=meta_path,
|
|
)
|
|
except Exception as cleanup_err:
|
|
logger.warning("Failed to cleanup temp files or Redis: %s", cleanup_err)
|
|
|
|
# Ensure response is defined (fallback in case of unexpected errors)
|
|
if response is None:
|
|
logger.error(f"Unexpected error: response not set for job {job_id}")
|
|
response = {
|
|
"status": "failed",
|
|
"error": "Error inesperado durante el procesamiento",
|
|
"inserted": 0,
|
|
"skipped_invalid": skipped_invalid,
|
|
"skipped_missing_invoice": skipped_missing_invoice,
|
|
"skipped_missing_fk": skipped_missing_fk,
|
|
"skipped_details": skipped_fk_details
|
|
}
|
|
|
|
return response
|