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

feature/clarion-invoices-temp-partidas-csv

Browse Source
This commit is contained in:
hreyes
2026-03-09 08:41:14 -06:00
parent ecd32d590c
commit 23a9c1c40a
3 changed files with 915 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -301,7 +301,263 @@ def scan_file(self, job_id: str, model_target: str, config: str = None):
except Exception as e:
logger.exception("Series import scan failed: %s", e)
return {"status": "failed", "error": str(e)}
# --- Partidas de Importación Temporal: flujo específico (Clarion VALIDA_TODA / VALIDA_PARCIAL) ---
if model_target == "invoice_details" and template_id == "imp_temp_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_temp import validate_row_partidas_impo_temp
_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 == "TEM",
)
)
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 == "TEM",
)
)
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,
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(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_temp_details", normalize_header)
inv = _get_row(row_norm, "NUMERO FACTURA", "NUM FACTURA", "FACTURA")
linea = _get_row(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(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_temp_details", normalize_header)
err = validate_row_partidas_impo_temp(
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 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
@@ -1099,6 +1355,9 @@ def insert_valid_rows(self, job_id: str, model_target: str):
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
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
footer_config = parse_footer_config(meta.get("footer_config"))
@@ -1136,7 +1395,20 @@ def insert_valid_rows(self, job_id: str, model_target: str):
currency_type_cache: Dict[str, Optional[str]] = {}
customs_section_cache: Dict[str, Optional[str]] = {}
part_cache: Dict[str, Optional[int]] = {}
class_id_by_code: Dict[str, int] = {}
uom_id_by_code: Dict[str, int] = {}
package_id_by_key: Dict[str, int] = {}
if model_target == 'invoice_details':
for c in session.query(Class.id, Class.class_code).filter(Class.tenant_id == tenant_id, Class.company_id == company_id).all():
if c[1]:
class_id_by_code[(c[1] or "").strip().upper()] = c[0]
for u in session.query(UnitOfMeasure.id, UnitOfMeasure.code).filter(UnitOfMeasure.tenant_id == tenant_id, UnitOfMeasure.company_id == company_id).all():
if u[1]:
uom_id_by_code[(u[1] or "").strip().upper()] = u[0]
for p in session.query(Package.id, Package.key).filter(Package.tenant_id == tenant_id, Package.company_id == company_id).all():
if p[1]:
package_id_by_key[(p[1] or "").strip()] = p[0]
validator = ForeignKeyValidator(session, tenant_id, company_id)
with open(file_path, 'r', encoding='utf-8-sig') as f:
@@ -1519,7 +1791,7 @@ def insert_valid_rows(self, job_id: str, model_target: str):
skipped_missing_invoice += 1
continue
part_num = (row_norm.get('NUMPARTE') or row_norm.get('NUMERO PARTE') or '').strip()
part_num = (row_norm.get('NUM. PARTE') or row_norm.get('NUMPARTE') or row_norm.get('NUMERO PARTE') or row_norm.get('NUM PARTE') or '').strip()
if not part_num:
skipped_invalid += 1
reason = "NUMPARTE: Requerido"
@@ -1533,66 +1805,56 @@ def insert_valid_rows(self, job_id: str, model_target: str):
logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
continue
# --- Prevent Duplicates: Clear existing items for this invoice (Once per job) ---
# --- Prevent Duplicates: Clear existing line 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(LineItem).filter(LineItem.invoice_id == invoice_id).delete(synchronize_session=False)
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()
# --- Partidas: LineItem with invoice_id (no Item parent) + full CSV mapping ---
part_num = (row_norm.get('NUM. PARTE') or row_norm.get('NUMPARTE') or row_norm.get('NUMERO PARTE') or row_norm.get('NUM PARTE') or '').strip()
part_id = part_cache.get(part_num) if part_num else None
if part_id is None and part_num:
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(
class_code = (row_norm.get('CLASE') or '').strip().upper()
class_id = class_id_by_code.get(class_code) if class_code else None
uom_code = (row_norm.get('UNIDAD DE MEDIDA') or row_norm.get('UNIDAD MEDIDA') or '').strip().upper()
uom_id = uom_id_by_code.get(uom_code) if uom_code else None
bulk_key = (row_norm.get('CLAVE BULTOS') or row_norm.get('CLAVEBULTOS') or '').strip()
package_id = package_id_by_key.get(bulk_key) if bulk_key else None
line = LineItem(
invoice_id=invoice_id,
line_number=line_num,
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
part_number_id=part_id,
class_id=class_id,
unit_of_measure=uom_id,
order=(row_norm.get('ORDEN DE COMPRA') or row_norm.get('ORDENCOMPRA') or None),
material_type=(row_norm.get('ID TYPE') or row_norm.get('IDTYPE') or None),
tax_payment=(str(row_norm.get('SE PAGO IMPUESTO') or row_norm.get('SEPAGOIMPUESTO') or '').strip().upper() == 'SI'),
payment_method=(row_norm.get('FORMA DE PAGO') or row_norm.get('FORMADEPAGO') or row_norm.get('FORMA PAGO') or None),
valuation_method=(row_norm.get('METODO DE VALORACION') or row_norm.get('METODODEVALORACION') or row_norm.get('METODO VALORACION') or None),
)
session.add(line)
session.flush() # Need line.id
session.flush()
# 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)
price = parse_decimal(row_norm.get('COSTO UNITARIO') or row_norm.get('PRECIO UNITARIO') or row_norm.get('PRECIOUNITARIO'))
if price is None:
total_val = parse_decimal(row_norm.get('TOTAL'))
qty = parse_decimal(row_norm.get('CANTIDAD IMPORTADA') or row_norm.get('CANTIDAD'))
price = (total_val / qty) if (total_val and qty and qty != 0) else None
qty = parse_decimal(row_norm.get('CANTIDAD IMPORTADA') or row_norm.get('CANTIDAD'))
commercial_total = (price * qty) if price and qty else parse_decimal(row_norm.get('TOTAL'))
session.add(LineFinancial(
item_line_id=line.id,
@@ -1600,41 +1862,60 @@ def insert_valid_rows(self, job_id: str, model_target: str):
total_commercial_value=decimal_or_zero(commercial_total),
))
# 4. Quantities (vanilla: nulls -> 0 so we always have a quantity row)
net_w = parse_decimal(row_norm.get('PESO NETO') or row_norm.get('PESONETO'))
gross_w = parse_decimal(row_norm.get('PESO BRUTO') or row_norm.get('PESOBRUTO'))
session.add(LineQuantity(
item_line_id=line.id,
quantity=decimal_or_zero(qty),
net_weight=decimal_or_zero(net_w),
gross_weight=decimal_or_zero(gross_w),
package_quantity=int_or_zero(row_norm.get('CANTIDAD BULTOS') or row_norm.get('CANTIDADBULTOS')),
package_id=package_id,
))
# 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,
))
origin = (row_norm.get('PAIS ORIGEN') or row_norm.get('PAISORIGEN') or row_norm.get('PAIS') or '').strip()
fraction = (row_norm.get('FRACCION ARANCELARIA') or row_norm.get('FRACCION') or row_norm.get('FRACCIONARANCELARIA') or '').strip()
fraction_type = (row_norm.get('PREFERENCIA ARANCELARIA') or row_norm.get('PREFERENCIA') or '').strip()
sector = (row_norm.get('SECTOR') or '').strip()
american_fraction = (row_norm.get('FRACCION AMERICANA') or row_norm.get('FRACCIONAMERICANA') or '').strip()
session.add(LineCustom(
item_line_id=line.id,
origin_country=origin or None,
fraction=fraction or None,
fraction_type=fraction_type or None,
sector=sector or None,
american_fraction=american_fraction or None,
))
# 6. Description
desc = row_norm.get('DESCRIPCION')
if desc:
session.add(LineDescription(
item_line_id=line.id,
description_spanish=desc,
))
desc_es = (row_norm.get('DESCRIPCION ESPAÑOL') or row_norm.get('DESCRIPCIONE') or row_norm.get('DESCRIPCION') or '').strip()
desc_en = (row_norm.get('DESCRIPCION INGLES') or row_norm.get('DESCRIPCIONI') or '').strip()
brand = (row_norm.get('MARCA') or '').strip()
model = (row_norm.get('MODELO') or '').strip()
extra_desc = (row_norm.get('DESCRIPCION EXTRA') or row_norm.get('DESCRIPCIONEXTRA') or '').strip()
additional_info = (row_norm.get('INFORMACION ADICIONAL') or row_norm.get('INFORMACIONADICIONAL') or '').strip()
lot = (row_norm.get('LOTE') or '').strip()
entry_number = (row_norm.get('NUMERO ENTRADA') or row_norm.get('NUMEROENTRADA') or row_norm.get('NUM ENTRADA') or '').strip()
session.add(LineDescription(
item_line_id=line.id,
description_spanish=desc_es or None,
description_english=desc_en or None,
brand=brand or None,
model=model or None,
extra_description=extra_desc or None,
additional_info_spanish=additional_info or None,
lot=lot or None,
entry_number=entry_number or None,
))
# 7. Legacy Sales Details (For specific audit/UI fields; vanilla: nulls -> 0)
detail = InvoiceSalesDetails(
session.add(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
))
details_to_insert.append(line)
# 3. Bulk Insert (ORM Transaction)
try:

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

@@ -52,19 +52,41 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
"imp_def_header": None, # se resuelve igual que imp_temp_header
# --- Encabezado factura: Expo (EstructuraEncFacExpoCamReg.xls) - misma estructura ---
"exp_def_header": None,
# --- Partidas factura: Impo Temp (EstructuraParFacImpoTempAF.xls) ---
# --- Partidas factura: Impo Temp (EstructuraParFacImpoTemp - paridad Clarion A-AG) ---
"imp_temp_details": [
{"canonical": "NUMERO FACTURA", "aliases": ["NUM FACTURA", "FACTURA"]},
{"canonical": "LINEA", "aliases": ["RENGLON", "PARTIDA"]},
{"canonical": "NUMPARTE", "aliases": ["NUMERO PARTE"]},
{"canonical": "PRECIO UNITARIO", "aliases": ["PRECIOUNITARIO"]},
{"canonical": "VALOR COMERCIAL", "aliases": ["VALORCOMERCIAL"]},
{"canonical": "CANTIDAD"},
{"canonical": "CLASE"},
{"canonical": "CANTIDAD IMPORTADA", "aliases": ["CANTIDAD"]},
{"canonical": "UNIDAD DE MEDIDA", "aliases": ["UNIDAD MEDIDA"]},
{"canonical": "COSTO UNITARIO", "aliases": ["COSTOUNITARIO", "PRECIO UNITARIO", "PRECIOUNITARIO"]},
{"canonical": "PESO NETO", "aliases": ["PESONETO"]},
{"canonical": "PESO BRUTO", "aliases": ["PESOBRUTO"]},
{"canonical": "CANTIDAD BULTOS", "aliases": ["CANTIDADBULTOS"]},
{"canonical": "DESCRIPCION"},
{"canonical": "PAIS ORIGEN", "aliases": ["PAISORIGEN"]},
{"canonical": "FRACCION"},
{"canonical": "ORDEN DE COMPRA", "aliases": ["ORDENCOMPRA"]},
{"canonical": "CLAVE BULTOS", "aliases": ["CLAVEBULTOS"]},
{"canonical": "PAIS ORIGEN", "aliases": ["PAISORIGEN", "PAIS"]},
{"canonical": "FRACCION ARANCELARIA", "aliases": ["FRACCION", "FRACCIONARANCELARIA"]},
{"canonical": "PREFERENCIA ARANCELARIA", "aliases": ["PREFERENCIA", "PREFERENCIAARANCELARIA"]},
{"canonical": "SECTOR"},
{"canonical": "FRACCION AMERICANA", "aliases": ["FRACCIONAMERICANA"]},
{"canonical": "ORDEN DE COMPRA", "aliases": ["ORDENCOMPRA", "ORDEN COMPRA"]},
{"canonical": "DESCRIPCION ESPAÑOL", "aliases": ["DESCRIPCIONE", "DESCRIPCION"]},
{"canonical": "DESCRIPCION INGLES", "aliases": ["DESCRIPCION INGLES", "DESCRIPCIONI"]},
{"canonical": "MARCA"},
{"canonical": "MODELO"},
{"canonical": "ES PARTIDA O SUBPARTIDA", "aliases": ["ES PARTIDA O SUBPARTIDA", "ESSUBPARTIDA"]},
{"canonical": "LINEA PRINCIPAL", "aliases": ["LINEAPRINCIPAL", "PARTIDA PRINCIPAL"]},
{"canonical": "NUM. PARTE", "aliases": ["NUMPARTE", "NUMERO PARTE", "NUM PARTE"]},
{"canonical": "SE PAGO IMPUESTO", "aliases": ["SE PAGO IMPUESTO", "SEPAGOIMPUESTO"]},
{"canonical": "FORMA DE PAGO", "aliases": ["FORMADEPAGO", "FORMA PAGO"]},
{"canonical": "METODO DE VALORACION", "aliases": ["METODODEVALORACION", "METODO VALORACION"]},
{"canonical": "DESCRIPCION EXTRA", "aliases": ["DESCRIPCIONEXTRA"]},
{"canonical": "INFORMACION ADICIONAL", "aliases": ["INFORMACIONADICIONAL"]},
{"canonical": "AGREGAR/SUSTITUIR", "aliases": ["AGREGAR SUSTITUIR", "SUSTITUIR"]},
{"canonical": "TOTAL"},
{"canonical": "NUMERO ENTRADA", "aliases": ["NUMEROENTRADA", "NUM ENTRADA"]},
{"canonical": "LOTE"},
{"canonical": "ID TYPE", "aliases": ["IDTYPE"]},
],
# --- Partidas: Impo Def y Expo - misma estructura ---
"imp_def_details": None,

535
backend/api/v1/modules/a76/layouts_csv/facturas/validators/partidas_impo_temp.py Normal file
View File

@@ -0,0 +1,535 @@
"""
Validaciones CSV para Partidas de Importación Temporal.
Paridad Clarion: VALIDA_TODA_PARIMPO_TEM, VALIDA_PARCIAL_PARIMPO_TEM, VALIDACIONES_PARIMPO_TEM.
Estructura: NUMERO FACTURA, LINEA, CLASE, CANTIDAD IMPORTADA, ... hasta ID TYPE (columnas A-AG).
"""
from decimal import Decimal, InvalidOperation
from typing import Dict, Any, Optional, Set, Tuple, List
# Longitudes máximas Clarion
MAX_LEN_FACTURA = 15
MAX_LEN_LINEA = 5
MAX_LEN_CLASE = 9
MAX_LEN_CANTIDAD_STR = 19
MAX_LEN_ORDEN_COMPRA = 20
PREFERENCIAS_VALIDAS = frozenset({"GENERAL", "TLCS", "PROSEC", "ALADI"})
SE_PAGO_IMPUESTO_VALIDOS = frozenset({"SI", "NO"})
APOSTROFE = "'"
def _clip(val: Any) -> str:
if val is None:
return ""
return str(val).strip()
def _parse_decimal(val: Any) -> Optional[Decimal]:
if val is None:
return None
s = _clip(val)
if not s:
return None
s = s.replace(",", "")
try:
return Decimal(s)
except (InvalidOperation, ValueError):
return None
def _get(row: Dict[str, Any], *keys: str) -> str:
for k in keys:
v = row.get(k)
if v is not None and str(v).strip():
return _clip(v)
return ""
def _check_factura_vacia(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
val = _get(row, "NUMERO FACTURA", "NUM FACTURA", "FACTURA")
if not val:
return {
"line": line_num,
"col": "NUMERO FACTURA",
"msg": "Error: (Celda A) La Factura de Importación está vacía y no se pueden hacer las validaciones.",
}
return None
def _check_factura_existe(
invoice_number: str,
line_num: int,
invoice_id_by_number: Dict[str, int],
) -> Optional[Dict[str, Any]]:
if invoice_number not in invoice_id_by_number:
return {
"line": line_num,
"col": "NUMERO FACTURA",
"msg": f"Error: (Celda A) La Factura de Importación {invoice_number} no existe en SCAII y no se pueden hacer las validaciones.",
}
return None
def _check_factura_no_actualizada(
invoice_number: str,
line_num: int,
invoice_updated_by_number: Dict[str, bool],
rfc_exception_updated: Set[str],
) -> Optional[Dict[str, Any]]:
if invoice_number in rfc_exception_updated:
return None
if invoice_updated_by_number.get(invoice_number, False):
return {
"line": line_num,
"col": "NUMERO FACTURA",
"msg": f"Error: (Celda A) La Factura de Importación: {invoice_number} ya existe y está Actualizada, no se puede hacer cambios a las facturas actualizadas.",
}
return None
def _check_linea_si_no_autonumerar(row: Dict[str, Any], line_num: int, autonumerar: bool) -> Optional[Dict[str, Any]]:
if autonumerar:
return None
val = _get(row, "LINEA", "RENGLON", "PARTIDA")
if not val:
return {
"line": line_num,
"col": "LINEA",
"msg": "Error: (Celda B) El campo de la línea de la partida está vacío y no se pueden hacer las validaciones, ya que se tiene la opción de Autonumerar como NO.",
}
return None
def _check_levantar_subpartidas_uv(
row: Dict[str, Any], line_num: int, levantar_subpartidas: bool
) -> Optional[Dict[str, Any]]:
if not levantar_subpartidas:
return None
u = _get(row, "ES PARTIDA O SUBPARTIDA", "ESSUBPARTIDA")
if not u:
return {
"line": line_num,
"col": "ES PARTIDA O SUBPARTIDA",
"msg": "Error: (Celda U) El campo del tipo de la partida (partida o subpartida) está vacío y no se pueden hacer las validaciones, ya que se tiene la opción de Levantar Subpartidas como Si.",
}
v = _get(row, "LINEA PRINCIPAL", "LINEAPRINCIPAL", "PARTIDA PRINCIPAL")
if not v:
return {
"line": line_num,
"col": "LINEA PRINCIPAL",
"msg": "Error: (Celda V) El campo de la partida principal está vacío y no se pueden hacer las validaciones, ya que se tiene la opción de Levantar Subpartidas como Si.",
}
return None
def _valida_toda_obligatorios(
row: Dict[str, Any],
line_num: int,
levantar_subpartidas: bool,
calcular_costo_en_base_a_total: bool,
) -> Optional[Dict[str, Any]]:
"""Obligatorios vacíos: C, D, F (condicional), G, K, M; U, V si LevantarSubpartidas."""
obligatorios: List[str] = []
if not _get(row, "CLASE"):
obligatorios.append("(Col.C) Clases")
if not _get(row, "CANTIDAD IMPORTADA", "CANTIDAD"):
obligatorios.append("(Col.D) Cantidad Importada")
es_subpartida = _get(row, "ES PARTIDA O SUBPARTIDA", "ESSUBPARTIDA").upper() == "S"
if not calcular_costo_en_base_a_total and not es_subpartida:
if not _get(row, "COSTO UNITARIO", "COSTOUNITARIO", "PRECIO UNITARIO", "PRECIOUNITARIO"):
obligatorios.append("(Col.F) Costo Unitario")
if not _get(row, "PESO NETO", "PESONETO"):
obligatorios.append("(Col.G) Peso Neto")
if not _get(row, "PAIS ORIGEN", "PAISORIGEN", "PAIS"):
obligatorios.append("(Col.K) País")
if not _get(row, "PREFERENCIA ARANCELARIA", "PREFERENCIA", "PREFERENCIAARANCELARIA"):
obligatorios.append("(Col.M) Preferencia.")
if levantar_subpartidas:
if not _get(row, "ES PARTIDA O SUBPARTIDA", "ESSUBPARTIDA"):
obligatorios.append("(Col.U) EsSubpartida?.")
if not _get(row, "LINEA PRINCIPAL", "LINEAPRINCIPAL", "PARTIDA PRINCIPAL"):
obligatorios.append("(Col.V) Partida Principal.")
if obligatorios:
return {
"line": line_num,
"col": "CLASE",
"msg": f"Existen campos vacíos que son obligatorios: {', '.join(obligatorios)}.",
}
return None
def _valida_toda_numericos(
row: Dict[str, Any],
line_num: int,
calcular_costo_en_base_a_total: bool,
) -> Optional[Dict[str, Any]]:
"""Costo unitario y peso neto no pueden ser cero (salvo subpartida / costo por total)."""
es_subpartida = _get(row, "ES PARTIDA O SUBPARTIDA", "ESSUBPARTIDA").upper() == "S"
if not calcular_costo_en_base_a_total and not es_subpartida:
costo = _parse_decimal(row.get("COSTO UNITARIO") or row.get("COSTOUNITARIO") or row.get("PRECIO UNITARIO") or row.get("PRECIOUNITARIO"))
if costo is not None and costo == 0:
return {
"line": line_num,
"col": "COSTO UNITARIO",
"msg": f"Error: (Celda F{line_num}) El Costo Unitario no puede ser cero.",
}
peso_neto = _parse_decimal(row.get("PESO NETO") or row.get("PESONETO"))
if peso_neto is not None and peso_neto == 0:
return {
"line": line_num,
"col": "PESO NETO",
"msg": f"Error: (Celda G{line_num}) El Peso Neto no puede ser cero.",
}
return None
def _valida_subpartidas_duplicados(
invoice_number: str,
linea: str,
line_num: int,
line_counts: Dict[Tuple[str, str], int],
) -> Optional[Dict[str, Any]]:
key = (invoice_number.strip(), _clip(linea))
if line_counts.get(key, 0) > 1:
return {
"line": line_num,
"col": "LINEA",
"msg": f"Error: (Celda B{line_num}) El campo de la partida está duplicado entre las partidas.",
}
return None
def _valida_subpartida_tiene_principal(
row: Dict[str, Any],
line_num: int,
partidas_principales_en_csv: Set[Tuple[str, str]],
partidas_principales_en_bd: Set[Tuple[str, str]],
) -> Optional[Dict[str, Any]]:
u = _get(row, "ES PARTIDA O SUBPARTIDA", "ESSUBPARTIDA").upper()
v = _get(row, "LINEA PRINCIPAL", "LINEAPRINCIPAL", "PARTIDA PRINCIPAL")
if u != "S" or not v or v == "0":
return None
inv = _get(row, "NUMERO FACTURA", "NUM FACTURA", "FACTURA")
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:
return None
return {
"line": line_num,
"col": "LINEA PRINCIPAL",
"msg": f"Error: (Celda V{line_num}) La partida principal {v} no existe.",
}
def _valida_subpartida_v_no_cero(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
u = _get(row, "ES PARTIDA O SUBPARTIDA", "ESSUBPARTIDA").upper()
v = _get(row, "LINEA PRINCIPAL", "LINEAPRINCIPAL", "PARTIDA PRINCIPAL")
if u == "S" and v == "0":
return {
"line": line_num,
"col": "LINEA PRINCIPAL",
"msg": f"Error: (Celda V{line_num}) La SubPartida no tiene asignada una partida principal.",
}
return None
def _validaciones_parimpo_tem(
row: Dict[str, Any],
line_num: int,
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,
validar_decimales_pza: bool,
rfc_exception_num_parte: Optional[Set[str]],
invoice_number: str,
) -> Optional[Dict[str, Any]]:
"""VALIDACIONES_PARIMPO_TEM: longitudes, catálogos, reglas de negocio."""
def err(col: str, msg: str) -> Dict[str, Any]:
return {"line": line_num, "col": col, "msg": msg}
clase = _get(row, "CLASE")
um = _get(row, "UNIDAD DE MEDIDA", "UNIDAD MEDIDA")
# Longitudes
factura = _get(row, "NUMERO FACTURA", "NUM FACTURA", "FACTURA")
if factura and len(factura) > MAX_LEN_FACTURA:
return err("NUMERO FACTURA", f"Error: (Celda A{line_num}) La Factura de Importación: {factura} supera la longitud de caracteres.")
linea = _get(row, "LINEA", "RENGLON", "PARTIDA")
if linea and len(linea) > MAX_LEN_LINEA:
return err("LINEA", f"Error: (Celda B{line_num}) La Línea de Importación: {linea} supera la longitud de caracteres.")
if clase:
if len(clase) > MAX_LEN_CLASE:
return err("CLASE", f"Error: (Celda C{line_num}) La Clase: {clase} supera la longitud de caracteres.")
if clase.upper() not in valid_class_codes:
return err("CLASE", f"Error: (Celda C{line_num}) La Clase: {clase} no existe en el Catálogo de Clases.")
if not um and not class_um_by_code.get(clase.upper()):
return err("UNIDAD DE MEDIDA", f"Error: (Celda E{line_num}) Debido a que esta celda es vacía, se asignará la unidad de medida de la Clase pero también está vacía.")
frac = _get(row, "FRACCION ARANCELARIA", "FRACCION", "FRACCIONARANCELARIA")
if not frac and not class_fraction_by_code.get(clase.upper()):
return err("FRACCION ARANCELARIA", f"Error: (Celda L{line_num}) Debido a que esta celda es vacía, se asignará la fracción de la Clase pero también está vacía.")
if not _get(row, "DESCRIPCION ESPAÑOL", "DESCRIPCIONE", "DESCRIPCION") and not class_desc_es_by_code.get(clase.upper()):
return err("DESCRIPCION ESPAÑOL", f"Error: (Celda Q{line_num}) Debido a que esta celda es vacía, se asignará la Descripción en Español de la Clase pero también está vacía.")
if not _get(row, "DESCRIPCION INGLES", "DESCRIPCIONI") and not class_desc_en_by_code.get(clase.upper()):
return err("DESCRIPCION INGLES", f"Error: (Celda R{line_num}) Debido a que esta celda es vacía, se asignará la Descripción en Inglés de la Clase pero también está vacía.")
# D: Cantidad importada
cant_str = _get(row, "CANTIDAD IMPORTADA", "CANTIDAD")
if cant_str:
cant = _parse_decimal(cant_str)
if cant is not None and cant <= 0:
return err("CANTIDAD IMPORTADA", f"Error: (Celda D{line_num}) La Cantidad Importada: {cant_str} es cero.")
if len(cant_str) > MAX_LEN_CANTIDAD_STR:
return err("CANTIDAD IMPORTADA", f"Error: (Celda D{line_num}) La Cantidad Importada: {cant_str} supera la cantidad de caracteres permitidos.")
# E: Unidad de medida en catálogo
if um and um.upper() not in valid_uom_codes:
return err("UNIDAD DE MEDIDA", f"Error: (Celda E{line_num}) La Unidad de Medida: {um} no existe en el Catálogo de Unidades de Medida.")
# I, J: Bultos condicional
clave_bultos = _get(row, "CLAVE BULTOS", "CLAVEBULTOS")
cant_bultos = row.get("CANTIDAD BULTOS") or row.get("CANTIDADBULTOS")
if clave_bultos:
if clave_bultos not in valid_bulks_codes:
return err("CLAVE BULTOS", f"Error: (Celda J{line_num}) La Clave de Bulto: {clave_bultos} no existe en el Catálogo de Claves de Bultos.")
cant_bultos_val = _parse_decimal(cant_bultos)
if cant_bultos_val is None:
return err("CANTIDAD BULTOS", f"Error: (Celda I{line_num}) La Cantidad de Bultos está vacía y en la Celda J{line_num} se tiene la Clave de Bulto.")
if cant_bultos_val == 0:
return err("CANTIDAD BULTOS", f"Error: (Celda I{line_num}) La Cantidad de Bultos es cero y en la Celda J{line_num} se tiene la Clave de Bulto.")
else:
cant_bultos_val = _parse_decimal(cant_bultos)
if cant_bultos_val is not None and cant_bultos_val > 0:
return err("CANTIDAD BULTOS", f"Error: (Celda I{line_num}) La Cantidad de Bultos es {cant_bultos} y en la Celda J{line_num} no se tiene la Clave de Bulto.")
# K: País SAAIM3 o americana
pais = _get(row, "PAIS ORIGEN", "PAISORIGEN", "PAIS")
if pais and pais.upper() not in valid_country_keys:
return err("PAIS ORIGEN", f"Error: (Celda K{line_num}) El País: {pais} no se encontró como Clave SAAIM3 ni Clave Americana en el Catálogo de Paises.")
# M: Preferencia
pref = _get(row, "PREFERENCIA ARANCELARIA", "PREFERENCIA", "PREFERENCIAARANCELARIA").upper()
if pref and pref not in PREFERENCIAS_VALIDAS:
return err("PREFERENCIA ARANCELARIA", f"Error: (Celda M{line_num}) La Preferencia Arancelaria: {pref} no es correcta para el sistema SCAF.")
sector = _get(row, "SECTOR")
if pref == "PROSEC":
if not sector:
return err("PREFERENCIA ARANCELARIA", f"Error: (Celda M{line_num} y N{line_num}) La Preferencia Arancelaria es: {pref} y en la columna N no tiene sector.")
if sector not in authorized_sectors:
return err("SECTOR", f"Error: (Celda N{line_num}) El Sector: {sector} no existe en el Catálogo de Sectores.")
if not company_has_prosec:
return err("SECTOR", f"Error: (Celda N{line_num}) La empresa no cuenta con autorización PROSEC.")
elif pref and pref != "PROSEC" and sector:
return err("PREFERENCIA ARANCELARIA", f"Error: (Celda M{line_num} y N{line_num}) La Preferencia Arancelaria es: {pref} y en la columna N tiene sector.")
# O: Fracción americana
frac_ame = _get(row, "FRACCION AMERICANA", "FRACCIONAMERICANA")
if frac_ame and frac_ame not in valid_fraction_ame:
return err("FRACCION AMERICANA", f"Advertencia: (Celda O{line_num}) La Fracción Americana: {frac_ame} no existe en el Catálogo de Fracciones Americanas.")
# P: Orden de compra máx 20
orden = _get(row, "ORDEN DE COMPRA", "ORDENCOMPRA")
if orden and len(orden) > MAX_LEN_ORDEN_COMPRA:
return err("ORDEN DE COMPRA", f"Error: (Celda P{line_num}) La Orden de Compra: {orden} supera la cantidad de caracteres permitidos.")
# Decimales PZA
if validar_decimales_pza:
um_code = (um or class_um_by_code.get(clase.upper() or "") or "").upper()
if um_code == "PZA" and cant_str:
d = _parse_decimal(cant_str)
if d is not None and d != int(d):
return err("CANTIDAD IMPORTADA", "Error: (Celda D) La Unidad de Medida es PZA, Por lo Tanto no es Válida la Captura de Decimales.")
# Z: Método de valoración
met_val = _get(row, "METODO DE VALORACION", "METODODEVALORACION", "METODO VALORACION")
if met_val and met_val not in valid_valuation_methods:
return err("METODO DE VALORACION", f"Error: (Celda Z{line_num}) El Método de Valoración Capturado: {met_val} No es Válido.")
# RFC excepción: W obligatorio
if rfc_exception_num_parte and invoice_number in rfc_exception_num_parte:
num_parte = _get(row, "NUM. PARTE", "NUMPARTE", "NUMERO PARTE", "NUM PARTE")
if not num_parte:
return err("NUM. PARTE", f"Error: (Celda W{line_num}) No está capturado el número de parte.")
# X: Se pagó impuesto SI/NO
x = _get(row, "SE PAGO IMPUESTO", "SEPAGOIMPUESTO")
if x and x.upper() not in SE_PAGO_IMPUESTO_VALIDOS:
return err("SE PAGO IMPUESTO", f"Error: (Celda X{line_num}) El Valor Capturado para Se Pago Impuesto no es Válido. Capturar SI o NO.")
# Y: Forma de pago en catálogo
forma_pago = _get(row, "FORMA DE PAGO", "FORMADEPAGO", "FORMA PAGO")
if forma_pago and forma_pago not in valid_payment_methods:
return err("FORMA DE PAGO", f"Error: (Celda Y{line_num}) La Forma de Pago Capturado no es Válido.")
return None
def _warn_apostrofes_num_parte(
row: Dict[str, Any], line_num: int, warnings: Optional[List[Dict[str, Any]]]
) -> None:
val = _get(row, "NUM. PARTE", "NUMPARTE", "NUMERO PARTE", "NUM PARTE")
if val and APOSTROFE in val and warnings is not None:
warnings.append({
"line": line_num,
"col": "NUM. PARTE",
"msg": f"Advertencia: El Número de Parte: {val} Contiene Apostrofes.",
"warning": True,
})
def validate_row_partidas_impo_temp(
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]]:
"""
Punto de entrada: valida una fila de CSV de Partidas de Importación Temporal.
Clarion: VALIDA_TODA vs VALIDA_PARCIAL según autonumerar, actualizar y si la partida existe.
"""
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(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
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