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

feature/feature/clarion-validaciones-invoices-csv-partidas-def-impo

Browse Source
This commit is contained in:
hreyes
2026-03-09 12:26:23 -06:00
parent bcec74a741
commit 71d3446351
4 changed files with 462 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -607,6 +607,266 @@ def scan_file(self, job_id: str, model_target: str, config: str = None):
logger.exception("Partidas import scan failed: %s", e)
return {"status": "failed", "error": str(e)}
# --- Partidas de Importación Definitiva: flujo específico (Clarion VALIDA_TODA_PARIMPO_DEF / VALIDA_PARCIAL) ---
# Estructura de columnas: misma que partidas TEM (imp_def_details resuelve a imp_temp_details). Facturas DEF/MATDE/EXDEF.
if model_target == "invoice_details" and template_id == "imp_def_details":
try:
from api.v1.modules.a76.invoices.models import InvoiceHeader
from api.v1.modules.a76.items.models import LineItem
from api.v1.modules.a76.classes.models import Class
from api.v1.modules.a76.general_catalogs.units_of_measure.models import UnitOfMeasure
from api.v1.modules.a76.general_catalogs.packages.models import Package
from api.v1.modules.public.reference_data.countries.models import Country
from api.v1.modules.public.reference_data.sectors.models import Sector
from api.v1.modules.public.reference_data.valuation_methods.models import ValuationMethod
from api.v1.modules.public.reference_data.payment_methods.models import PaymentMethod
from api.v1.modules.a76.general_catalogs.fractions.us_tariff_fractions.models import USTariffFraction
from api.v1.modules.a76.general_catalogs.company.models import Company
from api.v1.modules.a76.parts.models import Part
from .validators.partidas_impo_def import validate_row_partidas_impo_def
DEF_INVOICE_TYPES = ("DEF", "MATDE", "EXDEF")
_fc = parse_footer_config(meta.get("footer_config"))
autonumerar = meta.get("autonumerar", True)
actualizar = meta.get("actualizar", False)
levantar_subpartidas = meta.get("levantar_subpartidas", False)
calcular_costo_en_base_a_total = meta.get("calcular_costo_unitario_en_base_a_valor_total", False)
validar_decimales_pza = meta.get("validar_decimales_pza", False)
if _fc:
if "autonumerar" in _fc:
autonumerar = bool(_fc["autonumerar"])
elif _fc.get("autonumber_partidas", "true") is not None:
autonumerar = str(_fc.get("autonumber_partidas", "true")).lower() in ("true", "1", "si", "sí", "yes")
if "actualizar" in _fc:
actualizar = bool(_fc["actualizar"])
if "levantar_subpartidas" in _fc:
levantar_subpartidas = bool(_fc["levantar_subpartidas"])
if "calcular_costo_unitario_en_base_a_valor_total" in _fc:
calcular_costo_en_base_a_total = bool(_fc["calcular_costo_unitario_en_base_a_valor_total"])
if "validar_decimales_pza" in _fc:
validar_decimales_pza = bool(_fc["validar_decimales_pza"])
RFC_EXCEPTION_UPDATED = {"TPI121217SF6", "TCI170502858"}
RFC_EXCEPTION_NUM_PARTE = {"CTE980130518"}
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.in_(DEF_INVOICE_TYPES),
)
)
invoice_id_by_number: Dict[str, int] = {}
invoice_updated_by_number: Dict[str, bool] = {}
for num, iid, is_upd in q_inv.all():
if num:
invoice_id_by_number[str(num).strip()] = iid
invoice_updated_by_number[str(num).strip()] = bool(is_upd)
existing_line_keys_by_invoice: Dict[str, Set[str]] = {}
q_li = (
session.query(InvoiceHeader.invoice_number, LineItem.line_number)
.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.in_(DEF_INVOICE_TYPES),
)
)
for num, ln in q_li.all():
if num is not None:
key = str(num).strip()
if key not in existing_line_keys_by_invoice:
existing_line_keys_by_invoice[key] = set()
existing_line_keys_by_invoice[key].add(str(ln).strip())
partidas_principales_bd: Set[Tuple[str, str]] = set()
try:
from api.v1.modules.a24.fa.fa_item_lines.models import FaLineItem
q_pp = (
session.query(InvoiceHeader.invoice_number, LineItem.line_number)
.join(LineItem, LineItem.invoice_id == InvoiceHeader.id)
.join(FaLineItem, FaLineItem.id == LineItem.id)
.filter(
InvoiceHeader.tenant_id == tenant_id,
InvoiceHeader.company_id == company_id,
InvoiceHeader.operation_type == "imp",
InvoiceHeader.invoice_type.in_(DEF_INVOICE_TYPES),
FaLineItem.is_subitem == False,
FaLineItem.contains_subitems == True,
)
)
for num, ln in q_pp.all():
if num is not None:
partidas_principales_bd.add((str(num).strip(), str(ln).strip()))
except Exception:
pass
valid_class_codes: Set[str] = set()
class_um_by_code: Dict[str, str] = {}
class_fraction_by_code: Dict[str, str] = {}
class_desc_es_by_code: Dict[str, str] = {}
class_desc_en_by_code: Dict[str, str] = {}
for c in session.query(Class).filter(Class.tenant_id == tenant_id, Class.company_id == company_id).all():
code = (c.class_code or "").strip().upper()
if code:
valid_class_codes.add(code)
class_um_by_code[code] = (c.unit_of_measure or "").strip().upper()
class_fraction_by_code[code] = (c.fraction or "").strip()
class_desc_es_by_code[code] = (c.description_es or "").strip()
class_desc_en_by_code[code] = (c.description_en or "").strip()
valid_uom_codes: Set[str] = set()
for u in session.query(UnitOfMeasure.code).filter(UnitOfMeasure.tenant_id == tenant_id, UnitOfMeasure.company_id == company_id).all():
if u[0]:
valid_uom_codes.add((u[0] or "").strip().upper())
valid_bulks_codes: Set[str] = set()
for p in session.query(Package.key).filter(Package.tenant_id == tenant_id, Package.company_id == company_id).all():
if p[0]:
valid_bulks_codes.add((p[0] or "").strip())
valid_country_keys: Set[str] = set()
for row in session.query(Country.m3_key, Country.ame_key).all():
if row[0]:
valid_country_keys.add((row[0] or "").strip().upper())
if row[1]:
valid_country_keys.add((row[1] or "").strip().upper())
valid_fraction_ame: Set[str] = set()
for row in session.query(USTariffFraction.code).filter(USTariffFraction.tenant_id == tenant_id, USTariffFraction.company_id == company_id).all():
if row[0]:
valid_fraction_ame.add((row[0] or "").strip())
authorized_sectors: Set[str] = set()
for row in session.query(Sector.key).filter(Sector.authorized == True).all():
if row[0]:
authorized_sectors.add((row[0] or "").strip().upper())
valid_payment_methods: Set[str] = set()
for row in session.query(PaymentMethod.key).all():
if row[0] is not None:
valid_payment_methods.add(str(row[0]).strip())
valid_valuation_methods: Set[str] = set()
for row in session.query(ValuationMethod.key).all():
if row[0]:
valid_valuation_methods.add((row[0] or "").strip())
company = session.query(Company).filter(Company.id == company_id).first()
company_has_prosec = bool(company.prosec) if company else False
company_rfc = (company.rfc or "").strip().upper() if company else ""
valid_part_numbers: Set[str] = set()
for row in session.query(Part.part_number).filter(Part.tenant_id == tenant_id, Part.company_id == company_id).all():
if row[0]:
valid_part_numbers.add((row[0] or "").strip().upper())
rfc_exception_updated: Set[str] = set()
rfc_exception_num_parte: Set[str] = set()
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)
invoice_numbers_from_csv = set()
for row in rows_list:
inv = (row.get("NUMERO FACTURA") or row.get("NUM FACTURA") or row.get("FACTURA") or "").strip()
if inv:
invoice_numbers_from_csv.add(inv)
if company_rfc in RFC_EXCEPTION_UPDATED:
rfc_exception_updated = invoice_numbers_from_csv
if company_rfc in RFC_EXCEPTION_NUM_PARTE:
rfc_exception_num_parte = invoice_numbers_from_csv
line_counts_csv: Dict[Tuple[str, str], int] = {}
partidas_principales_csv: Set[Tuple[str, str]] = set()
def _get_row_def(row_norm: Dict[str, Any], *keys: str) -> str:
for k in keys:
v = row_norm.get(k)
if v is not None and str(v).strip():
return str(v).strip()
return ""
for row in rows_list:
row_norm = row_from_template(row, "imp_def_details", normalize_header)
inv = _get_row_def(row_norm, "NUMERO FACTURA", "NUM FACTURA", "FACTURA")
linea = _get_row_def(row_norm, "LINEA", "RENGLON", "PARTIDA")
if inv and linea:
key = (inv, linea)
line_counts_csv[key] = line_counts_csv.get(key, 0) + 1
u = _get_row_def(row_norm, "ES PARTIDA O SUBPARTIDA", "ESSUBPARTIDA").upper()
if u == "P" and inv and linea:
partidas_principales_csv.add((inv, linea))
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, "imp_def_details", normalize_header)
err = validate_row_partidas_impo_def(
row_norm,
i,
autonumerar=autonumerar,
actualizar=actualizar,
levantar_subpartidas=levantar_subpartidas,
calcular_costo_en_base_a_total=calcular_costo_en_base_a_total,
validar_decimales_pza=validar_decimales_pza,
invoice_id_by_number=invoice_id_by_number,
invoice_updated_by_number=invoice_updated_by_number,
rfc_exception_updated=rfc_exception_updated,
existing_line_keys_by_invoice=existing_line_keys_by_invoice,
line_counts_csv=line_counts_csv,
partidas_principales_csv=partidas_principales_csv,
partidas_principales_bd=partidas_principales_bd,
valid_class_codes=valid_class_codes,
class_um_by_code=class_um_by_code,
class_fraction_by_code=class_fraction_by_code,
class_desc_es_by_code=class_desc_es_by_code,
class_desc_en_by_code=class_desc_en_by_code,
valid_uom_codes=valid_uom_codes,
valid_bulks_codes=valid_bulks_codes,
valid_country_keys=valid_country_keys,
valid_fraction_ame=valid_fraction_ame,
valid_payment_methods=valid_payment_methods,
valid_valuation_methods=valid_valuation_methods,
authorized_sectors=authorized_sectors,
company_has_prosec=company_has_prosec,
rfc_exception_num_parte=rfc_exception_num_parte or None,
valid_part_numbers=valid_part_numbers,
warnings=None,
)
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", "")})
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)
except Exception as e:
logger.exception("Partidas importación definitiva scan failed: %s", e)
return {"status": "failed", "error": str(e)}
# --- Encabezados de Importación Temporal: flujo específico (Clarion VALIDA_TODA / VALIDA_PARCIAL) ---
if model_target == "invoice_header" and template_id == "imp_temp_header":
try:
@@ -2197,6 +2457,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_details" and _template_id_insert == "imp_def_details":
inv_type_value = "DEF"
logger.info(f"Processing CSV with operation_type={op_type_value}, invoice_type={inv_type_value}, date_format={date_format}")

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

@@ -9,10 +9,12 @@ from .encabezados_impo_def import (
validate_row_encabezados_impo_def,
parse_pedimento_col_a_impo_def,
)
from .partidas_impo_def import validate_row_partidas_impo_def
__all__ = [
"validate_row_encabezados_impo_temp",
"validate_row_encabezados_impo_def",
"validate_row_partidas_impo_def",
"row_to_transport_type_clarion",
"parse_pedimento_col_a",
"parse_pedimento_col_a_impo_def",

197
backend/api/v1/modules/a76/layouts_csv/facturas/validators/partidas_impo_def.py Normal file
View File

@@ -0,0 +1,197 @@
"""
Validaciones CSV para Partidas de Importación Definitiva.
Paridad Clarion: VALIDA_TODA_PARIMPO_DEF, VALIDA_PARCIAL_PARIMPO_DEF, VALIDACIONES_PARIMPO_DEF.
Reutiliza la lógica de partidas_impo_temp; solo cambia el mensaje cuando la factura no existe
(«no existe en el catálogo de Importación Definitiva») y el origen de facturas (DEF/MATDE/EXDEF en tasks.py).
Estructura de columnas: misma que partidas TEM (NUMERO FACTURA, LINEA, CLASE, ... ID TYPE).
"""
from typing import Dict, Any, Optional, Set, Tuple, List
from .partidas_impo_temp import (
_clip,
_get,
_check_factura_vacia,
_check_factura_no_actualizada,
_check_linea_si_no_autonumerar,
_check_levantar_subpartidas_uv,
_valida_toda_obligatorios,
_valida_toda_numericos,
_valida_subpartidas_duplicados,
_valida_subpartida_tiene_principal,
_valida_subpartida_v_no_cero,
_validaciones_parimpo_tem,
_warn_apostrofes_num_parte,
)
def _check_factura_existe_def(
invoice_number: str,
line_num: int,
invoice_id_by_number: Dict[str, int],
) -> Optional[Dict[str, Any]]:
"""Misma lógica que _check_factura_existe; mensaje específico Importación Definitiva (Clarion)."""
if invoice_number not in invoice_id_by_number:
return {
"line": line_num,
"col": "NUMERO FACTURA",
"msg": (
f"Error: (Celda A{line_num}) La Factura de Importación {invoice_number} "
"no existe en el catálogo de Importación Definitiva y no se pueden hacer las validaciones. "
),
}
return None
def validate_row_partidas_impo_def(
row: Dict[str, Any],
line_num: int,
autonumerar: bool,
actualizar: bool,
levantar_subpartidas: bool,
calcular_costo_en_base_a_total: bool,
validar_decimales_pza: bool,
invoice_id_by_number: Dict[str, int],
invoice_updated_by_number: Dict[str, bool],
rfc_exception_updated: Set[str],
existing_line_keys_by_invoice: Dict[str, Set[str]],
line_counts_csv: Dict[Tuple[str, str], int],
partidas_principales_csv: Set[Tuple[str, str]],
partidas_principales_bd: Set[Tuple[str, str]],
valid_class_codes: Set[str],
class_um_by_code: Dict[str, str],
class_fraction_by_code: Dict[str, str],
class_desc_es_by_code: Dict[str, str],
class_desc_en_by_code: Dict[str, str],
valid_uom_codes: Set[str],
valid_bulks_codes: Set[str],
valid_country_keys: Set[str],
valid_fraction_ame: Set[str],
valid_payment_methods: Set[str],
valid_valuation_methods: Set[str],
authorized_sectors: Set[str],
company_has_prosec: bool,
rfc_exception_num_parte: Optional[Set[str]],
valid_part_numbers: Optional[Set[str]],
warnings: Optional[List[Dict[str, Any]]] = None,
) -> Optional[Dict[str, Any]]:
"""
Valida una fila de CSV de Partidas de Importación Definitiva.
Clarion: VALIDA_TODA_PARIMPO_DEF vs VALIDA_PARCIAL_PARIMPO_DEF según autonumerar, actualizar y si la partida existe.
Reutiliza todo de partidas_impo_temp salvo el check de factura existente (mensaje DEF).
"""
err = _check_factura_vacia(row, line_num)
if err:
return err
invoice_number = _get(row, "NUMERO FACTURA", "NUM FACTURA", "FACTURA")
if not invoice_number:
return {"line": line_num, "col": "NUMERO FACTURA", "msg": "Requerido"}
err = _check_factura_existe_def(invoice_number, line_num, invoice_id_by_number)
if err:
return err
err = _check_factura_no_actualizada(
invoice_number, line_num, invoice_updated_by_number, rfc_exception_updated
)
if err:
return err
err = _check_linea_si_no_autonumerar(row, line_num, autonumerar)
if err:
return err
err = _check_levantar_subpartidas_uv(row, line_num, levantar_subpartidas)
if err:
return err
_warn_apostrofes_num_parte(row, line_num, warnings)
linea = _get(row, "LINEA", "RENGLON", "PARTIDA")
existing_lines = existing_line_keys_by_invoice.get(invoice_number.strip(), set())
partida_existe = bool(linea and linea in existing_lines)
use_partial = actualizar and not autonumerar and partida_existe
if use_partial:
return _validaciones_parimpo_tem(
row,
line_num,
valid_class_codes=valid_class_codes,
class_um_by_code=class_um_by_code,
class_fraction_by_code=class_fraction_by_code,
class_desc_es_by_code=class_desc_es_by_code,
class_desc_en_by_code=class_desc_en_by_code,
valid_uom_codes=valid_uom_codes,
valid_bulks_codes=valid_bulks_codes,
valid_country_keys=valid_country_keys,
valid_fraction_ame=valid_fraction_ame,
valid_payment_methods=valid_payment_methods,
valid_valuation_methods=valid_valuation_methods,
authorized_sectors=authorized_sectors,
company_has_prosec=company_has_prosec,
validar_decimales_pza=validar_decimales_pza,
rfc_exception_num_parte=rfc_exception_num_parte,
invoice_number=invoice_number,
)
else:
err = _valida_toda_obligatorios(
row, line_num, levantar_subpartidas, calcular_costo_en_base_a_total
)
if err:
return err
# Importación Definitiva: NUM. PARTE es obligatorio en todas las partidas (el insert lo exige).
num_parte = _get(row, "NUM. PARTE", "NUMPARTE", "NUMERO PARTE", "NUM PARTE")
if not (num_parte and str(num_parte).strip()):
return {
"line": line_num,
"col": "NUM. PARTE",
"msg": "NUM. PARTE: Requerido (obligatorio para partidas de Importación Definitiva).",
}
err = _valida_toda_numericos(row, line_num, calcular_costo_en_base_a_total)
if err:
return err
if levantar_subpartidas:
err = _valida_subpartidas_duplicados(
invoice_number, linea, line_num, line_counts_csv
)
if err:
return err
err = _valida_subpartida_tiene_principal(
row, line_num, partidas_principales_csv, partidas_principales_bd
)
if err:
return err
err = _valida_subpartida_v_no_cero(row, line_num)
if err:
return err
err = _validaciones_parimpo_tem(
row,
line_num,
valid_class_codes=valid_class_codes,
class_um_by_code=class_um_by_code,
class_fraction_by_code=class_fraction_by_code,
class_desc_es_by_code=class_desc_es_by_code,
class_desc_en_by_code=class_desc_en_by_code,
valid_uom_codes=valid_uom_codes,
valid_bulks_codes=valid_bulks_codes,
valid_country_keys=valid_country_keys,
valid_fraction_ame=valid_fraction_ame,
valid_payment_methods=valid_payment_methods,
valid_valuation_methods=valid_valuation_methods,
authorized_sectors=authorized_sectors,
company_has_prosec=company_has_prosec,
validar_decimales_pza=validar_decimales_pza,
rfc_exception_num_parte=rfc_exception_num_parte,
invoice_number=invoice_number,
)
if err:
return err
if rfc_exception_num_parte and invoice_number in rfc_exception_num_parte and valid_part_numbers is not None:
num_parte = _get(row, "NUM. PARTE", "NUMPARTE", "NUMERO PARTE", "NUM PARTE")
if num_parte and num_parte.upper() not in valid_part_numbers:
return {
"line": line_num,
"col": "NUM. PARTE",
"msg": f"Error: (Celda W{line_num}) El número de parte Capturado: {num_parte} no existe.",
}
return None

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

@@ -214,7 +214,7 @@ def _valida_subpartida_tiene_principal(
if not inv:
return None
key_principal = (inv.strip(), _clip(v))
if key_principal in partidas_principales_en_csv or key_principal in partidas_principales_bd:
if key_principal in partidas_principales_en_csv or key_principal in partidas_principales_en_bd:
return None
return {
"line": line_num,