ADUANASOFT/plantillas-proyectos
13
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feature/clarion-validaciones-invoices-csv-headers-comp-mex

Browse Source
This commit is contained in:
hreyes
2026-03-09 14:34:54 -06:00
parent b0e788d0cd
commit 09e20fb109
7 changed files with 623 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

375
backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py
View File

@@ -1703,6 +1703,186 @@ def scan_file(self, job_id: str, model_target: str, config: str = None):
logger.exception("Encabezados importación definitiva scan failed: %s", e)
return {"status": "failed", "error": str(e)}
# --- Encabezados Compras Mexicanas: flujo específico (Clarion VALIDA_TODA_FAC_COM_MEX / VALIDA_PARCIAL) ---
if model_target == "invoice_header" and template_id == "cmex_header":
try:
from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceFinancials
from api.v1.modules.a76.items.models import LineItem
from api.v1.modules.a76.clients_and_providers.models import ClientProvider
from api.v1.modules.public.reference_data.incoterms.models import Incoterm
from api.v1.modules.public.reference_data.currency_types.models import CurrencyType
from api.v1.modules.a76.general_catalogs.exchange_rate.models import ExchangeRate
from api.v1.modules.a76.transportation.transporters.models import Transporter
from .validators.encabezados_cmex import validate_row_encabezados_cmex
_fc = parse_footer_config(meta.get("footer_config"))
actualizar = meta.get("actualizar", False)
if _fc and "actualizar" in _fc:
actualizar = bool(_fc["actualizar"])
with CoreSessionLocal() as session:
q_inv = (
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 == "MEX",
)
)
invoice_exists_by_number = {}
invoice_updated_by_number = {}
for num, iid, is_upd in q_inv.all():
if num:
n = str(num).strip()
invoice_exists_by_number[n] = True
invoice_updated_by_number[n] = bool(is_upd)
valid_provider_ids = set()
valid_sold_to_ids = set()
valid_shipped_to_ids = set()
valid_provider_short_names = set()
valid_sold_to_short_names = set()
valid_shipped_to_short_names = set()
for cp in session.query(ClientProvider.id, ClientProvider.short_name).filter(
ClientProvider.tenant_id == tenant_id,
ClientProvider.company_id == company_id,
).all():
valid_provider_ids.add(cp[0])
valid_sold_to_ids.add(cp[0])
valid_shipped_to_ids.add(cp[0])
if cp[1] and str(cp[1]).strip():
sn_upper = str(cp[1]).strip().upper()
valid_provider_short_names.add(sn_upper)
valid_sold_to_short_names.add(sn_upper)
valid_shipped_to_short_names.add(sn_upper)
valid_transporter_keys = set()
for t in session.query(Transporter.transporter_key).filter(
Transporter.tenant_id == tenant_id,
Transporter.company_id == company_id,
).all():
if t[0]:
valid_transporter_keys.add((t[0] or "").strip().upper())
valid_incoterms = set()
for inc in session.query(Incoterm.code).all():
if inc[0]:
valid_incoterms.add((inc[0] or "").strip().upper())
valid_currency_codes = set()
for ct in session.query(CurrencyType.code).all():
if ct[0]:
valid_currency_codes.add((ct[0] or "").strip().upper())
exchange_rate_by_date = {}
for er in session.query(ExchangeRate.date, ExchangeRate.value).filter(
ExchangeRate.tenant_id == tenant_id,
ExchangeRate.company_id == company_id,
).all():
if er[0] and er[1] is not None:
dk = er[0].strftime("%Y-%m-%d") if hasattr(er[0], "strftime") else str(er[0])[:10]
exchange_rate_by_date[dk] = er[1]
invoice_has_partidas_by_number = {}
existing_tipo_moneda_by_number = {}
q_li_count = (
session.query(InvoiceHeader.invoice_number, func.count(LineItem.id))
.join(LineItem, LineItem.invoice_id == InvoiceHeader.id)
.filter(
InvoiceHeader.tenant_id == tenant_id,
InvoiceHeader.company_id == company_id,
InvoiceHeader.operation_type == "imp",
InvoiceHeader.invoice_type == "MEX",
)
.group_by(InvoiceHeader.invoice_number)
)
for num, cnt in q_li_count.all():
if num:
invoice_has_partidas_by_number[str(num).strip()] = cnt > 0
q_fin = (
session.query(InvoiceHeader.invoice_number, InvoiceFinancials.currency)
.join(InvoiceFinancials, InvoiceFinancials.invoice_id == InvoiceHeader.id)
.filter(
InvoiceHeader.tenant_id == tenant_id,
InvoiceHeader.company_id == company_id,
InvoiceHeader.operation_type == "imp",
InvoiceHeader.invoice_type == "MEX",
)
)
for num, cur in q_fin.all():
if num and cur:
cur_str = (cur or "").strip().lower()
if cur_str == "foreign":
existing_tipo_moneda_by_number[str(num).strip()] = "ME"
elif cur_str == "local":
existing_tipo_moneda_by_number[str(num).strip()] = "MN"
else:
existing_tipo_moneda_by_number[str(num).strip()] = (cur_str or "").upper()[:2]
with open(file_path, "r", encoding="utf-8-sig") as f_in:
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)
rows_list = list(reader)
error_count = 0
processed_rows = 0
error_lines_list = []
errors_detail = []
with open(error_path, "w", encoding="utf-8") as f_err:
for i, row in enumerate(rows_list, start=1):
if i % 1000 == 0:
self.update_state(state="PROGRESS", meta={"current": i, "total": total_rows, "errors": error_count})
row_norm = row_from_template(row, "cmex_header", normalize_header)
warnings_row = []
err = validate_row_encabezados_cmex(
row_norm,
i,
actualizar=actualizar,
invoice_exists_by_number=invoice_exists_by_number,
invoice_updated_by_number=invoice_updated_by_number,
valid_provider_ids=valid_provider_ids,
valid_sold_to_ids=valid_sold_to_ids,
valid_shipped_to_ids=valid_shipped_to_ids,
valid_transporter_keys=valid_transporter_keys,
valid_incoterms=valid_incoterms,
valid_currency_codes=valid_currency_codes,
exchange_rate_by_date=exchange_rate_by_date,
invoice_has_partidas_by_number=invoice_has_partidas_by_number,
existing_tipo_moneda_by_number=existing_tipo_moneda_by_number,
valid_provider_short_names=valid_provider_short_names,
valid_sold_to_short_names=valid_sold_to_short_names,
valid_shipped_to_short_names=valid_shipped_to_short_names,
date_format=date_format,
parse_date_fn=parse_date,
warnings=warnings_row,
)
if err:
error_count += 1
error_lines_list.append(err["line"])
f_err.write(json.dumps({"line": err["line"], "col": err.get("col", ""), "msg": err.get("msg", "")}) + "\n")
if len(errors_detail) < 500:
errors_detail.append({"line": err["line"], "col": err.get("col", ""), "msg": err.get("msg", "")})
for w in warnings_row:
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("Encabezados Compras Mexicanas 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
@@ -2905,6 +3085,8 @@ def insert_valid_rows(self, job_id: str, model_target: str):
)
if model_target == "invoice_header" and _template_id_insert == "imp_def_header":
inv_type_value = "DEF"
if model_target == "invoice_header" and _template_id_insert == "cmex_header":
inv_type_value = "MEX"
if model_target == "invoice_details" and _template_id_insert == "imp_def_details":
inv_type_value = "DEF"
@@ -3035,63 +3217,94 @@ def insert_valid_rows(self, job_id: str, model_target: str):
logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
continue
err = _validate_customs_broker_ref(
err = _validate_client_provider_ref(
validator,
CustomsBroker,
row_norm.get('AGENTE ADUANAL'),
ClientProvider,
row_norm.get('CLAVE ENVIADO A'),
i,
"AGENTE ADUANAL",
required=False,
"CLAVE ENVIADO A",
required=True,
)
if err:
skipped_missing_fk += 1
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
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
if inv_type_value != "MEX":
err = _validate_customs_broker_ref(
validator,
CustomsBroker,
row_norm.get('AGENTE ADUANAL'),
i,
"AGENTE ADUANAL",
required=False,
)
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,
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,
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,
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
if inv_type_value == "MEX":
err = validate_public_code(
validator,
CurrencyType,
row_norm.get('CLAVE MONEDA'),
i,
"CLAVE MONEDA",
)
if err and (row_norm.get('TIPO MONEDA') or '').strip().upper() == 'MC':
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
else:
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,
@@ -3145,8 +3358,8 @@ def insert_valid_rows(self, job_id: str, model_target: str):
# --- Resolve PEDIMENTO (Col A) to pedimento_id and REMESA (Col B) ---
pedimento_id = None
remesa_val = parse_int(row_norm.get('REMESA'))
ped_str = (row_norm.get('PEDIMENTO') or '').strip()
remesa_val = parse_int(row_norm.get('REMESA')) if inv_type_value != "MEX" else None
ped_str = (row_norm.get('PEDIMENTO') or '').strip() if inv_type_value != "MEX" else ''
if ped_str:
parsed = parse_pedimento_col_a(ped_str)
if parsed:
@@ -3182,12 +3395,15 @@ def insert_valid_rows(self, job_id: str, model_target: str):
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.document_type = (
None if inv_type_value == "MEX" else
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)
@@ -3214,12 +3430,15 @@ def insert_valid_rows(self, job_id: str, model_target: str):
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,
document_type=(
None if inv_type_value == "MEX" else
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),
@@ -3233,14 +3452,17 @@ def insert_valid_rows(self, job_id: str, model_target: str):
)
compliance = InvoiceComplianceMx(
pedimento_id=pedimento_id,
remesa=remesa_val,
aduana=resolve_public_code(
session,
CustomsSection,
CustomsSection.customs_code,
row_norm.get('ADUANA DE CRUCE'),
customs_section_cache,
pedimento_id=pedimento_id if inv_type_value != "MEX" else None,
remesa=remesa_val if inv_type_value != "MEX" else None,
aduana=(
None if inv_type_value == "MEX" else
resolve_public_code(
session,
CustomsSection,
CustomsSection.customs_code,
row_norm.get('ADUANA DE CRUCE'),
customs_section_cache,
)
),
provider_id=resolve_client_provider_id(
session,
@@ -3266,13 +3488,16 @@ def insert_valid_rows(self, job_id: str, model_target: str):
company_id,
shipped_to_cache,
),
customs_broker_id=resolve_customs_broker_id(
session,
CustomsBroker,
row_norm.get('AGENTE ADUANAL'),
tenant_id,
company_id,
broker_cache,
customs_broker_id=(
None if inv_type_value == "MEX" else
resolve_customs_broker_id(
session,
CustomsBroker,
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),

27
backend/api/v1/modules/a76/layouts_csv/facturas/template_config.py
View File

@@ -84,6 +84,33 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
],
# --- Encabezado factura: Expo (EstructuraEncFacExpoCamReg.xls) - misma estructura ---
"exp_def_header": None,
# --- Encabezado factura: Compras Mexicanas (Clarion VALIDA_TODA_FAC_COM_MEX / VALIDA_PARCIAL) ---
# Estructura CSV: A,B=CAPTURAR CMEX; C=NUMERO FACTURA; D=FECHA FACTURA; E=TIPO DE CAMBIO; F=CAPTURAR CMEX;
# G=CLAVE PROVEEDOR; H=CLAVE VENDIDO A; I=CLAVE ENVIADO A; J=CAPTURAR CMEX; K=CLAVE TRANSPORTISTA; ...; Z=OBSERVACIONES E
"cmex_header": [
{"canonical": "NUMERO FACTURA", "aliases": ["NUM FACTURA", "FACTURA"]},
{"canonical": "FECHA FACTURA", "aliases": ["FECHA"]},
{"canonical": "TIPO DE CAMBIO"},
{"canonical": "CLAVE PROVEEDOR"},
{"canonical": "CLAVE VENDIDO A", "aliases": ["CLAVE VENDIDO A:"]},
{"canonical": "CLAVE ENVIADO A"},
{"canonical": "CLAVE TRANSPORTISTA"},
{"canonical": "NOMBRE CONDUCTOR"},
{"canonical": "TIPO TRANSPORTE"},
{"canonical": "NUMERO TRANSPORTE"},
{"canonical": "TIPO MONEDA"},
{"canonical": "CLAVE MONEDA"},
{"canonical": "FLETES"},
{"canonical": "VALOR SEGUROS"},
{"canonical": "SEGUROS"},
{"canonical": "EMBALAJES"},
{"canonical": "OTROS INCREMENTABLES"},
{"canonical": "CLAVE INCOTERM"},
{"canonical": "PRECINTO"},
{"canonical": "FECHA EMISION"},
{"canonical": "TIPO PESO"},
{"canonical": "OBSERVACIONES E"},
],
# --- Partidas factura: Impo Temp (EstructuraParFacImpoTemp - paridad Clarion A-AG) ---
"imp_temp_details": [
{"canonical": "NUMERO FACTURA", "aliases": ["NUM FACTURA", "FACTURA"]},

2
backend/api/v1/modules/a76/layouts_csv/facturas/validators/__init__.py
View File

@@ -9,6 +9,7 @@ from .encabezados_impo_def import (
validate_row_encabezados_impo_def,
parse_pedimento_col_a_impo_def,
)
from .encabezados_cmex import validate_row_encabezados_cmex
from .partidas_impo_def import validate_row_partidas_impo_def
from .series_impo_def import (
validate_row_series_impo_def,
@@ -18,6 +19,7 @@ from .series_impo_def import (
__all__ = [
"validate_row_encabezados_impo_temp",
"validate_row_encabezados_impo_def",
"validate_row_encabezados_cmex",
"validate_row_partidas_impo_def",
"validate_row_series_impo_def",
"row_to_series_normalized_def",

280
backend/api/v1/modules/a76/layouts_csv/facturas/validators/encabezados_cmex.py Normal file
View File

@@ -0,0 +1,280 @@
"""
Validaciones CSV para Encabezados de Facturas de Compras Mexicanas.
Paridad Clarion: VALIDA_TODA_FAC_COM_MEX, VALIDA_PARCIAL_FAC_COM_MEX, VALIDACIONES_FAC_COM_MEX.
Sin pedimento, remesa, agente aduanal ni aduana de cruce.
Estructura CSV: NUMERO FACTURA (C), FECHA FACTURA (D), TIPO DE CAMBIO (E), CLAVE PROVEEDOR (G), ...
"""
from datetime import datetime
from decimal import Decimal
from typing import Any, Dict, List, Optional, Set
from .encabezados_impo_temp import (
MAX_LEN_FACTURA,
TIPO_PESO_VALIDOS,
TIPOS_MONEDA_VALIDOS,
_clip,
_err,
_get,
_parse_decimal,
_parse_int,
_validaciones_factura_longitud,
_validaciones_moneda,
_validaciones_tipo_cambio,
_validaciones_tipo_peso,
)
# Clarion Col M: Compras Mexicanas incluye "FERRO BARCAZA" (con espacio) y "PLATAFORMA"
TIPO_TRANSPORTE_VALIDOS_CMEX = frozenset({
"NINGUNO", "TRANSPORTE", "CAJA", "PLACAS", "CAMION", "BUQUE",
"FERROBARCAZA", "FERRO BARCAZA", "CONTENEDOR", "PLATAFORMA", "AVION",
})
def _validaciones_transporte_cmex(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""Tipo transporte (Col M) y número (Col N). Acepta FERRO BARCAZA y FERROBARCAZA."""
m_raw = _get(row, "TIPO TRANSPORTE")
m = m_raw.upper().replace(" ", "") if m_raw else ""
m_with_space = m_raw.upper() if m_raw else ""
n = _get(row, "NUMERO TRANSPORTE")
if m_raw and m_with_space not in TIPO_TRANSPORTE_VALIDOS_CMEX and m not in TIPO_TRANSPORTE_VALIDOS_CMEX:
return _err(
line_num,
"TIPO TRANSPORTE",
"Error: (Celda M{}) El Tipo de Transporte: {} no es válido. "
"Válidos: NINGUNO, TRANSPORTE, CAJA, PLACAS, CAMION, BUQUE, FERRO BARCAZA, CONTENEDOR, PLATAFORMA, AVION.".format(
line_num, m_raw
),
)
if not m_raw and n:
return _err(
line_num,
"NUMERO TRANSPORTE",
"Error: (Celda N{}) El Tipo de Transporte está vacío y está capturado un número de transporte.".format(line_num),
)
if m_raw and (m == "NINGUNO" or m_with_space == "NINGUNO") and n:
return _err(
line_num,
"NUMERO TRANSPORTE",
"Error: (Celda N{}) El Tipo de Transporte es NINGUNO y está capturado un número de transporte.".format(line_num),
)
if m_raw and m != "NINGUNO" and m_with_space != "NINGUNO" and not n:
return _err(
line_num,
"NUMERO TRANSPORTE",
"Error: (Celda N{}) El Tipo de Transporte es {} y no está capturado el número de transporte.".format(line_num, m_raw),
)
return None
def _validaciones_obligatorios_toda_cmex(
row: Dict[str, Any],
line_num: int,
actualizar: bool,
) -> Optional[Dict[str, Any]]:
"""Obligatorios VALIDA_TODA_FAC_COM_MEX: C siempre; si no es actualizar, también D, G, H, I."""
obligatorios: List[str] = []
if not _get(row, "NUMERO FACTURA", "NUM FACTURA", "FACTURA"):
obligatorios.append("(Col.C) Factura")
if not actualizar:
if not _get(row, "FECHA FACTURA", "FECHA"):
obligatorios.append("(Col.D) Fecha de la Factura")
if not _get(row, "CLAVE PROVEEDOR"):
obligatorios.append("(Col.G) Clave del Proveedor")
if not _get(row, "CLAVE VENDIDO A"):
obligatorios.append("(Col.H) Clave del Vendido A")
if not _get(row, "CLAVE ENVIADO A"):
obligatorios.append("(Col.I) Clave del Enviado A")
if obligatorios:
return _err(
line_num,
"ARCHIVO CSV",
"Existen campos vacíos que son obligatorios: {}. Revisar la línea del archivo y capturar los campos con la información correcta.".format(
", ".join(obligatorios)
),
)
return None
def _validaciones_catalogos_cmex(
row: Dict[str, Any],
line_num: int,
valid_provider_ids: Set[int],
valid_sold_to_ids: Set[int],
valid_shipped_to_ids: Set[int],
valid_provider_short_names: Set[str],
valid_sold_to_short_names: Set[str],
valid_shipped_to_short_names: Set[str],
valid_transporter_keys: Set[str],
valid_incoterms: Set[str],
) -> Optional[Dict[str, Any]]:
"""Catálogos para Compras Mexicanas: Proveedor, Vendido A, Enviado A, Transportista, Incoterm (sin Agente Aduanal ni Aduana)."""
def check_id_or_rfc(
val: Any,
col: str,
catalog_name: str,
valid_ids: Set[int],
valid_short_names: Set[str],
) -> Optional[Dict[str, Any]]:
if val is None or str(val).strip() == "":
return None
v = _parse_int(val)
if v is not None:
if valid_ids and v not in valid_ids:
return _err(line_num, col, "Error: La clave en {} no existe en el Catálogo de {}.".format(col, catalog_name))
return None
sn_norm = str(val).strip().upper()
if valid_short_names and sn_norm not in valid_short_names:
return _err(line_num, col, "Error: La clave/corta en {} no existe en el Catálogo de {}.".format(col, catalog_name))
if not valid_short_names:
return _err(line_num, col, "Error: (Celda) {} debe ser un número entero o clave corta (short name) válida.".format(col))
return None
err = check_id_or_rfc(
row.get("CLAVE PROVEEDOR"), "CLAVE PROVEEDOR", "Clientes/Proveedores",
valid_provider_ids, valid_provider_short_names,
)
if err:
return err
err = check_id_or_rfc(
row.get("CLAVE VENDIDO A"), "CLAVE VENDIDO A", "Clientes/Proveedores",
valid_sold_to_ids, valid_sold_to_short_names,
)
if err:
return err
err = check_id_or_rfc(
row.get("CLAVE ENVIADO A"), "CLAVE ENVIADO A", "Clientes/Proveedores",
valid_shipped_to_ids, valid_shipped_to_short_names,
)
if err:
return err
k = _get(row, "CLAVE TRANSPORTISTA")
if k and valid_transporter_keys and k.upper() not in valid_transporter_keys:
return _err(
line_num,
"CLAVE TRANSPORTISTA",
"Error: (Celda K{}) La Clave del Transportista: {} no existe en el Catálogo de Transportistas.".format(line_num, k),
)
v = _get(row, "CLAVE INCOTERM")
if v and valid_incoterms and v.upper() not in valid_incoterms:
return _err(
line_num,
"CLAVE INCOTERM",
"Error: (Celda V{}) La Clave de INCOTERM: {} no existe en el Catálogo de INCOTERMS.".format(line_num, v),
)
return None
def validate_row_encabezados_cmex(
row: Dict[str, Any],
line_num: int,
actualizar: bool,
invoice_exists_by_number: Dict[str, bool],
invoice_updated_by_number: Dict[str, bool],
valid_provider_ids: Set[int],
valid_sold_to_ids: Set[int],
valid_shipped_to_ids: Set[int],
valid_transporter_keys: Set[str],
valid_incoterms: Set[str],
valid_currency_codes: Set[str],
exchange_rate_by_date: Optional[Dict[str, Any]] = None,
invoice_has_partidas_by_number: Optional[Dict[str, bool]] = None,
existing_tipo_moneda_by_number: Optional[Dict[str, str]] = None,
valid_provider_short_names: Optional[Set[str]] = None,
valid_sold_to_short_names: Optional[Set[str]] = None,
valid_shipped_to_short_names: Optional[Set[str]] = None,
date_format: Optional[str] = None,
parse_date_fn=None,
warnings: Optional[List[Dict[str, Any]]] = None,
) -> Optional[Dict[str, Any]]:
"""
Valida una fila de CSV de Encabezados de Compras Mexicanas.
Clarion: VALIDA_TODA_FAC_COM_MEX (factura nueva o no actualizar) vs VALIDA_PARCIAL_FAC_COM_MEX (actualizar existente).
Siempre ejecuta VALIDACIONES_FAC_COM_MEX (longitud C, tipo cambio, catálogos G/H/I/K, transporte M/N, moneda O/P, incoterm V, tipo peso Y).
"""
factura = _get(row, "NUMERO FACTURA", "NUM FACTURA", "FACTURA")
if not factura:
return _err(
line_num,
"NUMERO FACTURA",
"Error: (Col.C) La columna de Número de Factura está vacía y no se pueden hacer las validaciones.",
)
if invoice_updated_by_number.get(factura.strip(), False):
return _err(
line_num,
"NUMERO FACTURA",
"Error: (Celda C{}) El Número de Factura: {} ya existe y está Actualizada, no se puede hacer cambios.".format(
line_num, factura
),
)
if actualizar and factura.strip() not in invoice_exists_by_number:
return _err(
line_num,
"NUMERO FACTURA",
"Error: (Col.C) Factura de importación {} no existe (modo Actualizar).".format(factura),
)
use_partial = actualizar and invoice_exists_by_number.get(factura.strip(), False)
if not use_partial:
err = _validaciones_obligatorios_toda_cmex(row, line_num, actualizar)
if err:
return err
err = _validaciones_factura_longitud(row, line_num)
if err:
return err
err = _validaciones_transporte_cmex(row, line_num)
if err:
return err
has_partidas = invoice_has_partidas_by_number.get(factura.strip(), False) if invoice_has_partidas_by_number else False
existing_moneda = existing_tipo_moneda_by_number.get(factura.strip()) if existing_tipo_moneda_by_number else None
err = _validaciones_moneda(
row,
line_num,
valid_currency_codes or set(),
has_partidas if use_partial else None,
existing_moneda if use_partial else None,
)
if err:
return err
err = _validaciones_tipo_peso(row, line_num)
if err:
return err
err = _validaciones_catalogos_cmex(
row,
line_num,
valid_provider_ids or set(),
valid_sold_to_ids or set(),
valid_shipped_to_ids or set(),
valid_provider_short_names or set(),
valid_sold_to_short_names or set(),
valid_shipped_to_short_names or set(),
valid_transporter_keys or set(),
valid_incoterms or set(),
)
if err:
return err
invoice_date_parsed = None
if parse_date_fn:
date_str = _get(row, "FECHA FACTURA", "FECHA")
if date_str:
invoice_date_parsed = parse_date_fn(date_str, date_format)
err = _validaciones_tipo_cambio(
row, line_num, invoice_date_parsed, exchange_rate_by_date or {}, warnings
)
if err:
return err
return None