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

feature/clarion-partidas-series-csv-comp-mex

Browse Source
This commit is contained in:
hreyes
2026-03-09 15:11:15 -06:00
parent 09e20fb109
commit 423d0ac3fd
6 changed files with 474 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
backend/api/v1/modules/a76/csv_templates/registry.py
View File

@@ -59,15 +59,19 @@ def _canonicals_from_columns(cols: Optional[List[Dict]]) -> List[str]:
def _build_registry() -> Dict[str, List[str]]:
registry: Dict[str, List[str]] = {}
# a76/imports (facturas): imp_temp_header, imp_temp_details, imp_def_*, exp_def_*, cmex_header
# a76/imports (facturas): imp_temp_header, imp_temp_details, imp_def_*, exp_def_*, cmex_*, series
for tid in (
"imp_temp_header",
"imp_temp_details",
"imp_temp_series",
"imp_def_header",
"imp_def_details",
"imp_def_series",
"exp_def_header",
"exp_def_details",
"cmex_header",
"cmex_details",
"cmex_series",
):
cols = resolve_imports_template(tid)
registry[tid] = _canonicals_from_columns(cols)
@@ -145,9 +149,13 @@ TEMPLATE_FILENAMES: Dict[str, str] = {
"transporters": "EstructuraCatTransportistas.csv",
"imp_temp_header": "EstructuraEncFacImpoTemp.csv",
"imp_temp_details": "EstructuraParFacImpoTempAF.csv",
"imp_temp_series": "EstructuraSeriesFacImpoTemp.csv",
"imp_def_header": "EstructuraEncFacImpoDef.csv",
"imp_def_details": "EstructuraParFacImpoDefAF.csv",
"imp_def_series": "EstructuraSeriesFacImpoDef.csv",
"cmex_header": "EstructuraEncFacComprasMex.csv",
"cmex_details": "EstructuraParFacComprasMex.csv",
"cmex_series": "EstructuraSeriesFacComprasMex.csv",
"exp_def_header": "EstructuraEncFacExpoCamReg.csv",
"exp_def_details": "EstructuraParExpoCamReg.csv",
}

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

@@ -393,6 +393,167 @@ def scan_file(self, job_id: str, model_target: str, config: str = None):
logger.exception("Series importación definitiva scan failed: %s", e)
return {"status": "failed", "error": str(e)}
# --- Series Compras Mexicanas: misma lógica que Impo Def, facturas MEX ---
if model_target == "invoice_series" and template_id == "cmex_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 api.v1.modules.a76.items.line_quantities.models import LineQuantity
from .validators.series_impo_def import validate_row_series_impo_def
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 == "MEX",
)
)
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)
partida_max_series: Dict[Tuple[str, str], int] = {}
q_qty = (
session.query(
InvoiceHeader.invoice_number,
LineItem.line_number,
LineQuantity.quantity,
)
.join(LineItem, LineItem.invoice_id == InvoiceHeader.id)
.outerjoin(LineQuantity, LineQuantity.item_line_id == LineItem.id)
.filter(
InvoiceHeader.tenant_id == tenant_id,
InvoiceHeader.company_id == company_id,
InvoiceHeader.operation_type == "imp",
InvoiceHeader.invoice_type == "MEX",
)
)
for num, ln, qty in q_qty.all():
if num is not None and ln is not None:
key = (str(num).strip(), str(ln).strip())
if qty is not None:
partida_max_series[key] = int(qty) if qty else 0
else:
partida_max_series[key] = 0
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 == "MEX",
)
)
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 "",
})
csv_series_count_so_far: Dict[Tuple[str, str], int] = {}
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 = []
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, "cmex_series", normalize_header)
warnings_list: List[Dict[str, Any]] = []
err = validate_row_series_impo_def(
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,
partida_max_series=partida_max_series,
csv_series_count_so_far=csv_series_count_so_far,
existing_series_keys=existing_series_keys,
existing_series_data=existing_series_data,
warnings=warnings_list,
catalog_label="Compras Mexicanas",
)
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", "")})
else:
inv_num = (row_norm.get("NUMERO FACTURA") or row_norm.get("NUM FACTURA") or "").strip()
line_fac = (row_norm.get("LINEA FACTURA") or row_norm.get("LINEA") or row_norm.get("PARTIDA") or "").strip()
if inv_num and line_fac:
key_csv = (inv_num, line_fac)
csv_series_count_so_far[key_csv] = csv_series_count_so_far.get(key_csv, 0) + 1
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 Compras Mexicanas scan failed: %s", e)
return {"status": "failed", "error": str(e)}
# --- Series de Importación Temporal: flujo específico (Clarion VALIDA_TODA / VALIDA_PARCIAL) ---
if model_target == "invoice_series":
try:
@@ -1038,6 +1199,265 @@ def scan_file(self, job_id: str, model_target: str, config: str = None):
logger.exception("Partidas importación definitiva scan failed: %s", e)
return {"status": "failed", "error": str(e)}
# --- Partidas Compras Mexicanas: misma lógica que Impo Def, facturas MEX ---
if model_target == "invoice_details" and template_id == "cmex_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
_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 == "MEX",
)
)
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 == "MEX",
)
)
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 == "MEX",
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_cmex(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, "cmex_details", normalize_header)
inv = _get_row_cmex(row_norm, "NUMERO FACTURA", "NUM FACTURA", "FACTURA")
linea = _get_row_cmex(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_cmex(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, "cmex_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,
catalog_label="Compras Mexicanas",
)
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 Compras Mexicanas 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:
@@ -2525,7 +2945,7 @@ def insert_valid_rows(self, job_id: str, model_target: str):
# Si el upload fue de series (template_id imp_temp_series o imp_def_series), usar flujo series aunque model_target venga mal
use_series_flow = (
model_target == "invoice_series"
or meta.get("template_id") in ("imp_temp_series", "imp_def_series")
or meta.get("template_id") in ("imp_temp_series", "imp_def_series", "cmex_series")
)
_footer_for_series = parse_footer_config(meta.get("footer_config")) or {}
@@ -2534,11 +2954,12 @@ def insert_valid_rows(self, job_id: str, model_target: str):
use_series_flow
and (
meta.get("template_id") == "imp_def_series"
or meta.get("template_id") == "cmex_series"
or _inv_type_series in ("DEF", "MATDE", "EXDEF")
)
)
# --- Series de Importación Definitiva: commit (INSERT/UPDATE item_line_series para facturas DEF) ---
# --- Series de Importación Definitiva: commit (INSERT/UPDATE item_line_series para facturas DEF o MEX) ---
if use_def_series_commit:
try:
from api.v1.modules.a76.invoices.models import InvoiceHeader
@@ -2550,7 +2971,12 @@ def insert_valid_rows(self, job_id: str, model_target: str):
row_to_series_normalized_def,
)
DEF_INVOICE_TYPES = ("DEF", "MATDE", "EXDEF")
_series_template_id = meta.get("template_id")
is_cmex_series = _series_template_id == "cmex_series"
SERIES_INV_TYPES = ("MEX",) if is_cmex_series else ("DEF", "MATDE", "EXDEF")
series_row_template_id = "cmex_series" if is_cmex_series else "imp_def_series"
DEF_INVOICE_TYPES = ("DEF", "MATDE", "EXDEF") # keep for any legacy reference
actualizar = meta.get("actualizar", False)
autonumerar = meta.get("autonumerar", True)
validar_series_exception = meta.get("validar_series", False)
@@ -2577,7 +3003,7 @@ def insert_valid_rows(self, job_id: str, model_target: str):
InvoiceHeader.tenant_id == tenant_id,
InvoiceHeader.company_id == company_id,
InvoiceHeader.operation_type == "imp",
InvoiceHeader.invoice_type.in_(DEF_INVOICE_TYPES),
InvoiceHeader.invoice_type.in_(SERIES_INV_TYPES),
)
)
rows_inv = q.all()
@@ -2601,7 +3027,7 @@ def insert_valid_rows(self, job_id: str, model_target: str):
InvoiceHeader.tenant_id == tenant_id,
InvoiceHeader.company_id == company_id,
InvoiceHeader.operation_type == "imp",
InvoiceHeader.invoice_type.in_(DEF_INVOICE_TYPES),
InvoiceHeader.invoice_type.in_(SERIES_INV_TYPES),
)
)
for num, ln, qty in q_qty.all():
@@ -2628,7 +3054,7 @@ def insert_valid_rows(self, job_id: str, model_target: str):
InvoiceHeader.tenant_id == tenant_id,
InvoiceHeader.company_id == company_id,
InvoiceHeader.operation_type == "imp",
InvoiceHeader.invoice_type.in_(DEF_INVOICE_TYPES),
InvoiceHeader.invoice_type.in_(SERIES_INV_TYPES),
)
)
for num, ln, rw, sn, md, sm, nid in q_ser.all():
@@ -2660,7 +3086,7 @@ def insert_valid_rows(self, job_id: str, model_target: str):
for i, row in enumerate(reader, start=1):
if i in error_lines:
continue
row_norm = row_from_template(row, "imp_def_series", normalize_header)
row_norm = row_from_template(row, series_row_template_id, normalize_header)
err = validate_row_series_impo_def(
row_norm,
i,
@@ -2674,6 +3100,7 @@ def insert_valid_rows(self, job_id: str, model_target: str):
existing_series_keys=existing_series_keys,
existing_series_data=existing_series_data,
warnings=None,
catalog_label="Compras Mexicanas" if is_cmex_series else "Importación Definitiva",
)
if err:
skipped_invalid += 1
@@ -3089,7 +3516,11 @@ def insert_valid_rows(self, job_id: str, model_target: str):
inv_type_value = "MEX"
if model_target == "invoice_details" and _template_id_insert == "imp_def_details":
inv_type_value = "DEF"
if model_target == "invoice_details" and _template_id_insert == "cmex_details":
inv_type_value = "MEX"
if _template_id_insert == "cmex_series":
inv_type_value = "MEX"
logger.info(f"Processing CSV with operation_type={op_type_value}, invoice_type={inv_type_value}, date_format={date_format}")
headers_to_insert = []

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

@@ -186,6 +186,10 @@ def _resolve_template_columns(template_id: str) -> Optional[List[Dict[str, Any]]
return TEMPLATE_COLUMNS.get("imp_temp_header")
if template_id in ("imp_def_details", "exp_def_details"):
return TEMPLATE_COLUMNS.get("imp_temp_details")
if template_id == "cmex_details":
return TEMPLATE_COLUMNS.get("imp_temp_details")
if template_id == "cmex_series":
return TEMPLATE_COLUMNS.get("imp_def_series")
return None

12
backend/api/v1/modules/a76/layouts_csv/facturas/validators/partidas_impo_def.py
View File

@@ -28,15 +28,16 @@ def _check_factura_existe_def(
invoice_number: str,
line_num: int,
invoice_id_by_number: Dict[str, int],
catalog_label: str = "Importación Definitiva",
) -> Optional[Dict[str, Any]]:
"""Misma lógica que _check_factura_existe; mensaje específico Importación Definitiva (Clarion)."""
"""Misma lógica que _check_factura_existe; mensaje específico según catalog_label (DEF o Compras Mexicanas)."""
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. "
f"no existe en el catálogo de {catalog_label} y no se pueden hacer las validaciones. "
),
}
return None
@@ -73,6 +74,7 @@ def validate_row_partidas_impo_def(
rfc_exception_num_parte: Optional[Set[str]],
valid_part_numbers: Optional[Set[str]],
warnings: Optional[List[Dict[str, Any]]] = None,
catalog_label: str = "Importación Definitiva",
) -> Optional[Dict[str, Any]]:
"""
Valida una fila de CSV de Partidas de Importación Definitiva.
@@ -87,7 +89,7 @@ def validate_row_partidas_impo_def(
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)
err = _check_factura_existe_def(invoice_number, line_num, invoice_id_by_number, catalog_label)
if err:
return err
@@ -139,13 +141,13 @@ def validate_row_partidas_impo_def(
)
if err:
return err
# Importación Definitiva: NUM. PARTE es obligatorio en todas las partidas (el insert lo exige).
# Importación Definitiva / Compras Mexicanas: 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).",
"msg": f"NUM. PARTE: Requerido (obligatorio para partidas de {catalog_label}).",
}
err = _valida_toda_numericos(row, line_num, calcular_costo_en_base_a_total)
if err:

17
backend/api/v1/modules/a76/layouts_csv/facturas/validators/series_impo_def.py
View File

@@ -23,19 +23,20 @@ def _check_factura_existe_def(
invoice_number: str,
line_num: int,
invoice_id_by_number: Dict[str, int],
catalog_label: str = "Importación Definitiva",
) -> Optional[Dict[str, Any]]:
"""Factura debe existir en BD (Importación Definitiva: DEF/MATDE/EXDEF)."""
"""Factura debe existir en BD (Importación Definitiva o Compras Mexicanas según catalog_label)."""
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 SCAII y no se pueden hacer las validaciones. "
f"no existe en SCAII y no se pueden hacer las validaciones. "
),
"solution": (
f"Capturar en la Celda A{line_num} un número de Factura existente "
"al cual desee agregar o actualizar series"
f"al cual desee agregar o actualizar series"
),
"identifier": "FAC_IMPO_DEF",
"fields": invoice_number,
@@ -47,14 +48,15 @@ def _check_factura_no_actualizada_def(
invoice_number: str,
line_num: int,
invoice_updated_by_number: Dict[str, bool],
catalog_label: str = "Importación Definitiva",
) -> Optional[Dict[str, Any]]:
"""Si factura ya actualizada (Estatus AC) no se pueden hacer cambios. Mensaje DEF."""
"""Si factura ya actualizada (Estatus AC) no se pueden hacer cambios. Mensaje según catalog_label."""
if invoice_updated_by_number.get(invoice_number, False):
return {
"line": line_num,
"col": "NUMERO FACTURA",
"msg": (
f"Error: (Celda A{line_num}) La Factura de Importación: {invoice_number} "
f"Error: (Celda A{line_num}) La Factura de {catalog_label}: {invoice_number} "
"ya existe y está Actualizada, no se puede hacer cambios a las facturas actualizadas."
),
"solution": (
@@ -224,6 +226,7 @@ def validate_row_series_impo_def(
existing_series_keys: Set[Tuple[str, str, str]],
existing_series_data: Optional[Dict[Tuple[str, str, str], Dict[str, Any]]],
warnings: Optional[List[Dict[str, Any]]] = None,
catalog_label: str = "Importación Definitiva",
) -> Optional[Dict[str, Any]]:
"""
Punto de entrada: valida una fila de CSV de Series de Importación Definitiva.
@@ -243,11 +246,11 @@ def validate_row_series_impo_def(
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)
err = _check_factura_existe_def(invoice_number, line_num, invoice_id_by_number, catalog_label)
if err:
return err
err = _check_factura_no_actualizada_def(invoice_number, line_num, invoice_updated_by_number)
err = _check_factura_no_actualizada_def(invoice_number, line_num, invoice_updated_by_number, catalog_label)
if err:
return err