feature/feature/clarion-validaciones-invoices-csv-series-def-impo
This commit is contained in:
hreyes
@@ -217,10 +217,181 @@ def scan_file(self, job_id: str, model_target: str, config: str = None):
|
||||
"imp_temp_header" if model_target == "invoice_header" else
|
||||
"imp_temp_details" if model_target == "invoice_details" else "imp_temp_series"
|
||||
)
|
||||
|
||||
inv_type_value = normalize_public_code(footer_config.get("invoice_type") or meta.get("invoice_type") or "TEM")
|
||||
if not inv_type_value:
|
||||
inv_type_value = "TEM"
|
||||
if model_target == "invoice_series" and inv_type_value in ("DEF", "MATDE", "EXDEF"):
|
||||
template_id = "imp_def_series"
|
||||
|
||||
# --- Series de Importación Definitiva: flujo específico (Clarion VALIDA_TODA_SERIES_IMPO_DEF / VALIDA_PARCIAL) ---
|
||||
DEF_SERIES_TEMPLATE_OR_TYPE = (
|
||||
model_target == "invoice_series"
|
||||
and (
|
||||
template_id == "imp_def_series"
|
||||
or (inv_type_value in ("DEF", "MATDE", "EXDEF"))
|
||||
)
|
||||
)
|
||||
if DEF_SERIES_TEMPLATE_OR_TYPE:
|
||||
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,
|
||||
)
|
||||
|
||||
DEF_INVOICE_TYPES = ("DEF", "MATDE", "EXDEF")
|
||||
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.in_(DEF_INVOICE_TYPES),
|
||||
)
|
||||
)
|
||||
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.in_(DEF_INVOICE_TYPES),
|
||||
)
|
||||
)
|
||||
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.in_(DEF_INVOICE_TYPES),
|
||||
)
|
||||
)
|
||||
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, "imp_def_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,
|
||||
)
|
||||
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 importación definitiva 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":
|
||||
@@ -2171,12 +2342,289 @@ def insert_valid_rows(self, job_id: str, model_target: str):
|
||||
error_path = common_storage.error_path_for_job(JOB_TYPE, job_id)
|
||||
error_lines = common_storage.get_error_lines(JOB_TYPE, job_id, error_path)
|
||||
|
||||
# Si el upload fue de series (template_id imp_temp_series), usar flujo series aunque model_target venga mal
|
||||
# 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") == "imp_temp_series"
|
||||
or meta.get("template_id") in ("imp_temp_series", "imp_def_series")
|
||||
)
|
||||
|
||||
_footer_for_series = parse_footer_config(meta.get("footer_config")) or {}
|
||||
_inv_type_series = normalize_public_code(_footer_for_series.get("invoice_type") or meta.get("invoice_type") or "")
|
||||
use_def_series_commit = (
|
||||
use_series_flow
|
||||
and (
|
||||
meta.get("template_id") == "imp_def_series"
|
||||
or _inv_type_series in ("DEF", "MATDE", "EXDEF")
|
||||
)
|
||||
)
|
||||
|
||||
# --- Series de Importación Definitiva: commit (INSERT/UPDATE item_line_series para facturas DEF) ---
|
||||
if use_def_series_commit:
|
||||
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,
|
||||
row_to_series_normalized_def,
|
||||
)
|
||||
|
||||
DEF_INVOICE_TYPES = ("DEF", "MATDE", "EXDEF")
|
||||
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.in_(DEF_INVOICE_TYPES),
|
||||
)
|
||||
)
|
||||
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 = {}
|
||||
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.in_(DEF_INVOICE_TYPES),
|
||||
)
|
||||
)
|
||||
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())
|
||||
partida_max_series[key] = int(qty) if qty else 0
|
||||
|
||||
existing_series_keys = 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.in_(DEF_INVOICE_TYPES),
|
||||
)
|
||||
)
|
||||
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] = {}
|
||||
inserted_count = 0
|
||||
updated_count = 0
|
||||
skipped_invalid = 0
|
||||
skipped_details: List[Dict[str, Any]] = []
|
||||
|
||||
with open(file_path, "r", encoding="utf-8-sig") as f:
|
||||
sample = f.read(2048)
|
||||
f.seek(0)
|
||||
try:
|
||||
dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
|
||||
except Exception:
|
||||
dialect = "excel"
|
||||
reader = csv.DictReader(f, dialect=dialect)
|
||||
|
||||
for i, row in enumerate(reader, start=1):
|
||||
if i in error_lines:
|
||||
continue
|
||||
row_norm = row_from_template(row, "imp_def_series", normalize_header)
|
||||
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=None,
|
||||
)
|
||||
if err:
|
||||
skipped_invalid += 1
|
||||
skipped_details.append({
|
||||
"line": i,
|
||||
"invoice": (row_norm.get("NUMERO FACTURA") or row_norm.get("NUM FACTURA") or "").strip(),
|
||||
"reason": err.get("msg", ""),
|
||||
})
|
||||
continue
|
||||
|
||||
data = row_to_series_normalized_def(row_norm)
|
||||
invoice_number = data["NUMERO FACTURA"]
|
||||
linea_factura = data["LINEA FACTURA"]
|
||||
linea_serie = data["LINEA SERIE"]
|
||||
|
||||
if not invoice_number or invoice_number not in invoice_id_by_number:
|
||||
skipped_invalid += 1
|
||||
continue
|
||||
invoice_id = invoice_id_by_number[invoice_number]
|
||||
line_number_val = parse_int(linea_factura)
|
||||
if line_number_val is None:
|
||||
skipped_invalid += 1
|
||||
skipped_details.append({"line": i, "invoice": invoice_number, "reason": "LINEA FACTURA debe ser numérico."})
|
||||
continue
|
||||
line_item = (
|
||||
session.query(LineItem)
|
||||
.filter(
|
||||
LineItem.invoice_id == invoice_id,
|
||||
LineItem.line_number == line_number_val,
|
||||
LineItem.tenant_id == tenant_id,
|
||||
LineItem.company_id == company_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if not line_item:
|
||||
line_numbers = [
|
||||
r[0] for r in
|
||||
session.query(LineItem.line_number)
|
||||
.filter(
|
||||
LineItem.invoice_id == invoice_id,
|
||||
LineItem.tenant_id == tenant_id,
|
||||
LineItem.company_id == company_id,
|
||||
)
|
||||
.order_by(LineItem.line_number)
|
||||
.all()
|
||||
]
|
||||
existing_str = ", ".join(str(n) for n in line_numbers) if line_numbers else "ninguna"
|
||||
skipped_invalid += 1
|
||||
skipped_details.append({
|
||||
"line": i,
|
||||
"invoice": invoice_number,
|
||||
"reason": f"Partida línea {linea_factura} no existe en la factura. Partidas existentes en la factura: {existing_str}.",
|
||||
})
|
||||
continue
|
||||
|
||||
if autonumerar:
|
||||
max_row = (
|
||||
session.query(Serie.row)
|
||||
.filter(Serie.line_item_id == line_item.id)
|
||||
.order_by(Serie.row.desc())
|
||||
.limit(1)
|
||||
.scalar()
|
||||
)
|
||||
row_num = (max_row or 0) + 1
|
||||
else:
|
||||
row_num = parse_int(linea_serie)
|
||||
if row_num is None:
|
||||
skipped_invalid += 1
|
||||
skipped_details.append({"line": i, "invoice": invoice_number, "reason": "LINEA SERIE debe ser numérico."})
|
||||
continue
|
||||
|
||||
existing_serie = (
|
||||
session.query(Serie)
|
||||
.filter(
|
||||
Serie.line_item_id == line_item.id,
|
||||
Serie.row == row_num,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
if existing_serie:
|
||||
if actualizar:
|
||||
existing_serie.serial_numbers = data["SERIE"] or existing_serie.serial_numbers
|
||||
existing_serie.model = data["MODELO"] or existing_serie.model
|
||||
existing_serie.sub_model = data["SUB MODELO"] or existing_serie.sub_model
|
||||
existing_serie.number_id = data["NUMERO ID"] or existing_serie.number_id
|
||||
session.add(existing_serie)
|
||||
updated_count += 1
|
||||
else:
|
||||
skipped_invalid += 1
|
||||
skipped_details.append({"line": i, "invoice": invoice_number, "reason": "Serie ya existe (use actualizar)."})
|
||||
else:
|
||||
new_serie = Serie(
|
||||
tenant_id=tenant_id,
|
||||
company_id=company_id,
|
||||
line_item_id=line_item.id,
|
||||
row=row_num,
|
||||
serial_numbers=data["SERIE"] or None,
|
||||
model=data["MODELO"] or None,
|
||||
sub_model=data["SUB MODELO"] or None,
|
||||
number_id=data["NUMERO ID"] or None,
|
||||
)
|
||||
session.add(new_serie)
|
||||
inserted_count += 1
|
||||
|
||||
key_csv = (invoice_number.strip(), (linea_factura or "").strip())
|
||||
if key_csv[0] and key_csv[1]:
|
||||
csv_series_count_so_far[key_csv] = csv_series_count_so_far.get(key_csv, 0) + 1
|
||||
|
||||
session.commit()
|
||||
|
||||
common_storage.cleanup_import_job(JOB_TYPE, job_id, file_path=file_path, error_path=error_path, meta_path=meta_path)
|
||||
status = "finished" if (inserted_count + updated_count) > 0 else ("warning" if skipped_invalid else "failed")
|
||||
out = {
|
||||
"status": status,
|
||||
"inserted": inserted_count,
|
||||
"updated": updated_count,
|
||||
"skipped_invalid": skipped_invalid,
|
||||
"skipped_missing_fk": 0,
|
||||
"skipped_duplicate": 0,
|
||||
"skipped_details": skipped_details,
|
||||
}
|
||||
if status == "failed":
|
||||
out["error"] = "No hay registros válidos en el archivo CSV."
|
||||
elif status == "warning" and skipped_invalid:
|
||||
out["message"] = f"No se insertaron registros. {skipped_invalid} fueron rechazados."
|
||||
return out
|
||||
except Exception as e:
|
||||
logger.exception("Series importación definitiva commit failed: %s", e)
|
||||
return {"status": "failed", "error": str(e)}
|
||||
|
||||
# --- Series de Importación Temporal: commit (INSERT/UPDATE item_line_series) ---
|
||||
if use_series_flow:
|
||||
try:
|
||||
|
||||
@@ -136,6 +136,18 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
|
||||
{"canonical": "NUMERO ID", "aliases": ["NUMEROID"]},
|
||||
{"canonical": "COL_EXTRA"}, # Optional; if has value → desfase warning (Clarion)
|
||||
],
|
||||
# --- Series de Importación Definitiva (misma estructura que TEM; cabeceras imagen: NUMERO/LINEA FAC, LINEA SER, etc.) ---
|
||||
"imp_def_series": [
|
||||
{"canonical": "NUMERO FACTURA", "aliases": ["NUM FACTURA", "FACTURA", "NUMERO / LINEA FAC"]},
|
||||
{"canonical": "LINEA FACTURA", "aliases": ["LINEA", "PARTIDA"]},
|
||||
{"canonical": "LINEA SERIE", "aliases": ["RENGLON", "LINEA SER"]},
|
||||
{"canonical": "SERIE", "aliases": ["NUMERO SERIE"]},
|
||||
{"canonical": "MODELO"},
|
||||
{"canonical": "NUM PARTE", "aliases": ["NUMPARTE", "NUMERO PARTE", "NUM PART"]},
|
||||
{"canonical": "SUB MODELO", "aliases": ["SUBMODELO", "SUB MODE"]},
|
||||
{"canonical": "NUMERO ID", "aliases": ["NUMEROID"]},
|
||||
{"canonical": "COL_EXTRA"},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -10,11 +10,17 @@ from .encabezados_impo_def import (
|
||||
parse_pedimento_col_a_impo_def,
|
||||
)
|
||||
from .partidas_impo_def import validate_row_partidas_impo_def
|
||||
from .series_impo_def import (
|
||||
validate_row_series_impo_def,
|
||||
row_to_series_normalized_def,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"validate_row_encabezados_impo_temp",
|
||||
"validate_row_encabezados_impo_def",
|
||||
"validate_row_partidas_impo_def",
|
||||
"validate_row_series_impo_def",
|
||||
"row_to_series_normalized_def",
|
||||
"row_to_transport_type_clarion",
|
||||
"parse_pedimento_col_a",
|
||||
"parse_pedimento_col_a_impo_def",
|
||||
|
||||
@@ -0,0 +1,335 @@
|
||||
"""
|
||||
Validaciones CSV para Series de Importación Definitiva.
|
||||
Paridad Clarion: VALIDA_TODA_SERIES_IMPO_DEF, VALIDA_PARCIAL_SERIES_IMPO_DEF, LLENA_SERIES_IMPO_DEF.
|
||||
Estructura: NUMERO FACTURA, LINEA FACTURA, LINEA SERIE, SERIE, MODELO, NUM PARTE, SUB MODELO, NUMERO ID.
|
||||
Reutiliza helpers de series_impo_temp; añade reglas DEF: factura en catálogo DEF, cantidad series vs partida.
|
||||
"""
|
||||
from typing import Dict, Any, Optional, Set, Tuple, List
|
||||
|
||||
from .series_impo_temp import (
|
||||
_clip,
|
||||
normalize_sacarcomasenters,
|
||||
_check_desfase,
|
||||
_check_factura_vacia,
|
||||
_check_linea_factura_vacia,
|
||||
_check_linea_serie_si_no_autonumerar,
|
||||
_warn_apostrofes,
|
||||
_check_max_length,
|
||||
MAX_LEN,
|
||||
)
|
||||
|
||||
|
||||
def _check_factura_existe_def(
|
||||
invoice_number: str,
|
||||
line_num: int,
|
||||
invoice_id_by_number: Dict[str, int],
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""Factura debe existir en BD (Importación Definitiva: DEF/MATDE/EXDEF)."""
|
||||
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. "
|
||||
),
|
||||
"solution": (
|
||||
f"Capturar en la Celda A{line_num} un número de Factura existente "
|
||||
"al cual desee agregar o actualizar series"
|
||||
),
|
||||
"identifier": "FAC_IMPO_DEF",
|
||||
"fields": invoice_number,
|
||||
}
|
||||
return None
|
||||
|
||||
|
||||
def _check_factura_no_actualizada_def(
|
||||
invoice_number: str,
|
||||
line_num: int,
|
||||
invoice_updated_by_number: Dict[str, bool],
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""Si factura ya actualizada (Estatus AC) no se pueden hacer cambios. Mensaje DEF."""
|
||||
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} "
|
||||
"ya existe y está Actualizada, no se puede hacer cambios a las facturas actualizadas."
|
||||
),
|
||||
"solution": (
|
||||
"Capturar otro número de Factura de Importación Temporal o Desactualizar la factura."
|
||||
),
|
||||
"identifier": "FAC_IMPO_DEF",
|
||||
}
|
||||
return None
|
||||
|
||||
|
||||
def _check_partida_existe_en_factura(
|
||||
invoice_number: str,
|
||||
linea_factura: str,
|
||||
line_num: int,
|
||||
partida_max_series: Dict[Tuple[str, str], int],
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Clarion: la partida (línea de factura) debe existir en la factura.
|
||||
partida_max_series tiene como claves (invoice_number, line_number) de las partidas existentes.
|
||||
"""
|
||||
if not linea_factura:
|
||||
return None
|
||||
key = (invoice_number.strip(), _clip(linea_factura))
|
||||
if key in partida_max_series:
|
||||
return None
|
||||
return {
|
||||
"line": line_num,
|
||||
"col": "LINEA FACTURA",
|
||||
"msg": f"Partida línea {linea_factura} no existe en la factura.",
|
||||
"solution": "Usar un número de línea de partida que exista en la factura (capturar partidas antes de cargar series).",
|
||||
"identifier": "ARCHIVO CSV",
|
||||
}
|
||||
|
||||
|
||||
def _check_cantidad_series_vs_partida(
|
||||
invoice_number: str,
|
||||
linea_factura: str,
|
||||
line_num: int,
|
||||
partida_max_series: Dict[Tuple[str, str], int],
|
||||
csv_series_count_so_far: Dict[Tuple[str, str], int],
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Clarion: si cantidad de series en CSV para (factura, partida) excede CantImpoDef → error.
|
||||
partida_max_series[(inv, line)] = máximo permitido (desde LineQuantity.quantity).
|
||||
csv_series_count_so_far = conteo actual de filas válidas ya procesadas por (inv, line).
|
||||
"""
|
||||
key = (invoice_number.strip(), _clip(linea_factura))
|
||||
max_allowed = partida_max_series.get(key)
|
||||
if max_allowed is None or max_allowed <= 0:
|
||||
return None
|
||||
current = csv_series_count_so_far.get(key, 0)
|
||||
if current + 1 > max_allowed:
|
||||
return {
|
||||
"line": line_num,
|
||||
"col": "LINEA FACTURA",
|
||||
"msg": (
|
||||
f"Error: La cantidad de Series de la partida {_clip(linea_factura)} en la factura "
|
||||
f"{invoice_number} es menor al número de Series en el archivo. "
|
||||
),
|
||||
"solution": "Nivelar la cantidad de la Partida o el número de Series.",
|
||||
"identifier": "ARCHIVO CSV",
|
||||
}
|
||||
return None
|
||||
|
||||
|
||||
def _get_val_def(row: Dict[str, Any], k: str) -> str:
|
||||
"""Obtener valor normalizado por clave (aliases DEF/TEM)."""
|
||||
if k == "NUM PARTE":
|
||||
return _clip(row.get("NUM PARTE") or row.get("NUMPARTE") or row.get("NUMERO PARTE"))
|
||||
if k == "SUB MODELO":
|
||||
return _clip(row.get("SUB MODELO") or row.get("SUBMODELO"))
|
||||
if k == "NUMERO ID":
|
||||
return _clip(row.get("NUMERO ID") or row.get("NUMEROID"))
|
||||
return _clip(row.get(k))
|
||||
|
||||
|
||||
def valida_toda_series_impo_def(
|
||||
row: Dict[str, Any],
|
||||
line_num: int,
|
||||
validar_series_exception: bool,
|
||||
warnings: Optional[List[Dict[str, Any]]],
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
VALIDA_TODA_SERIES_IMPO_DEF: cuando no existe la serie o autonumerar=SI.
|
||||
Si D+E+F están todos vacíos y no aplica excepción ValidarSeries → error obligatorios.
|
||||
Valida longitudes máximas.
|
||||
"""
|
||||
d = _get_val_def(row, "SERIE")
|
||||
e = _get_val_def(row, "MODELO")
|
||||
f = _get_val_def(row, "NUM PARTE")
|
||||
campos = d + e + f
|
||||
|
||||
if not campos and not validar_series_exception:
|
||||
obligatorios = []
|
||||
if not d:
|
||||
obligatorios.append("(Col.D) Serie")
|
||||
if not e:
|
||||
obligatorios.append("(Col.E) Modelo")
|
||||
if not f:
|
||||
obligatorios.append("(Col.F) Num. Parte")
|
||||
if obligatorios:
|
||||
return {
|
||||
"line": line_num,
|
||||
"col": "SERIE",
|
||||
"msg": (
|
||||
f"Existen campos vacíos que son obligatorios al no tener ningun campo, "
|
||||
f"es la {', '.join(obligatorios)}."
|
||||
),
|
||||
"solution": "Revisar la línea del archivo y capturar los campos con la información correcta.",
|
||||
"identifier": "ARCHIVO CSV",
|
||||
}
|
||||
|
||||
for col, key, max_len in [
|
||||
("SERIE", "SERIE", MAX_LEN["serial_numbers"]),
|
||||
("MODELO", "MODELO", MAX_LEN["model"]),
|
||||
("NUM PARTE", "NUM PARTE", 50),
|
||||
("SUB MODELO", "SUB MODELO", MAX_LEN["sub_model"]),
|
||||
("NUMERO ID", "NUMERO ID", MAX_LEN["number_id"]),
|
||||
]:
|
||||
val = _get_val_def(row, key)
|
||||
if val:
|
||||
err = _check_max_length(col, val, line_num, max_len)
|
||||
if err:
|
||||
return err
|
||||
return None
|
||||
|
||||
|
||||
def valida_parcial_series_impo_def(
|
||||
row: Dict[str, Any],
|
||||
line_num: int,
|
||||
existing_series_data: Dict[str, Any],
|
||||
warnings: Optional[List[Dict[str, Any]]],
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
VALIDA_PARCIAL_SERIES_IMPO_DEF: actualizar serie existente; campos vacíos se rellenan con existente.
|
||||
Solo validar longitudes en campos no vacíos.
|
||||
"""
|
||||
for col, key, max_len in [
|
||||
("SERIE", "SERIE", MAX_LEN["serial_numbers"]),
|
||||
("MODELO", "MODELO", MAX_LEN["model"]),
|
||||
("NUM PARTE", "NUM PARTE", 50),
|
||||
("SUB MODELO", "SUB MODELO", MAX_LEN["sub_model"]),
|
||||
("NUMERO ID", "NUMERO ID", MAX_LEN["number_id"]),
|
||||
]:
|
||||
val = _get_val_def(row, key)
|
||||
if val:
|
||||
err = _check_max_length(col, val, line_num, max_len)
|
||||
if err:
|
||||
return err
|
||||
return None
|
||||
|
||||
|
||||
def _series_key(invoice_number: str, linea_factura: str, linea_serie: str) -> Tuple[str, str, str]:
|
||||
return (invoice_number.strip(), _clip(linea_factura), _clip(linea_serie))
|
||||
|
||||
|
||||
def validate_row_series_impo_def(
|
||||
row: Dict[str, Any],
|
||||
line_num: int,
|
||||
actualizar: bool,
|
||||
autonumerar: bool,
|
||||
validar_series_exception: bool,
|
||||
invoice_id_by_number: Dict[str, int],
|
||||
invoice_updated_by_number: Dict[str, bool],
|
||||
partida_max_series: Dict[Tuple[str, str], int],
|
||||
csv_series_count_so_far: Dict[Tuple[str, str], int],
|
||||
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,
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Punto de entrada: valida una fila de CSV de Series de Importación Definitiva.
|
||||
Clarion: decisión VALIDA_TODA vs VALIDA_PARCIAL según autonumerar, actualizar y si la serie existe.
|
||||
partida_max_series: máximo de series por (invoice_number, linea_factura); 0 o ausente = no validar.
|
||||
csv_series_count_so_far: conteo de filas ya aceptadas por (invoice_number, linea_factura); el caller debe incrementar al aceptar.
|
||||
"""
|
||||
desfase = _check_desfase(row, line_num)
|
||||
if desfase and warnings is not None:
|
||||
warnings.append(desfase)
|
||||
|
||||
err = _check_factura_vacia(row, line_num)
|
||||
if err:
|
||||
return err
|
||||
|
||||
invoice_number = _clip(row.get("NUMERO FACTURA") or row.get("NUM FACTURA") or row.get("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_def(invoice_number, line_num, invoice_updated_by_number)
|
||||
if err:
|
||||
return err
|
||||
|
||||
err = _check_linea_factura_vacia(row, line_num)
|
||||
if err:
|
||||
return err
|
||||
|
||||
linea_factura = _clip(row.get("LINEA FACTURA") or row.get("LINEA") or row.get("PARTIDA"))
|
||||
err = _check_partida_existe_en_factura(
|
||||
invoice_number,
|
||||
linea_factura,
|
||||
line_num,
|
||||
partida_max_series,
|
||||
)
|
||||
if err:
|
||||
return err
|
||||
|
||||
err = _check_cantidad_series_vs_partida(
|
||||
invoice_number,
|
||||
linea_factura,
|
||||
line_num,
|
||||
partida_max_series,
|
||||
csv_series_count_so_far,
|
||||
)
|
||||
if err:
|
||||
return err
|
||||
|
||||
err = _check_linea_serie_si_no_autonumerar(row, line_num, autonumerar)
|
||||
if err:
|
||||
return err
|
||||
|
||||
_warn_apostrofes(row, line_num, warnings)
|
||||
|
||||
linea_serie = _clip(row.get("LINEA SERIE") or row.get("RENGLON"))
|
||||
key = _series_key(invoice_number, linea_factura, linea_serie)
|
||||
|
||||
use_partial = (
|
||||
actualizar
|
||||
and not autonumerar
|
||||
and bool(linea_serie)
|
||||
and key in (existing_series_keys or set())
|
||||
)
|
||||
|
||||
if use_partial and existing_series_data and key in existing_series_data:
|
||||
return valida_parcial_series_impo_def(
|
||||
row, line_num, existing_series_data[key], warnings
|
||||
)
|
||||
return valida_toda_series_impo_def(
|
||||
row, line_num, validar_series_exception, warnings
|
||||
)
|
||||
|
||||
|
||||
def row_to_series_normalized_def(row: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""
|
||||
Normaliza fila para guardar: SACARCOMASENTERS en D, E, F, G, H.
|
||||
Clarion LLENA_SERIES_IMPO_DEF: asigna QueCSV a SerDef.
|
||||
Incluye NUM PARTE para validación; el modelo Serie actual no tiene campo parte (documentar si se añade).
|
||||
"""
|
||||
def clip(col: str, alt: Optional[List[str]] = None) -> str:
|
||||
v = row.get(col)
|
||||
if alt:
|
||||
for k in alt:
|
||||
if v is None or (isinstance(v, str) and not v.strip()):
|
||||
v = row.get(k)
|
||||
return _clip(v) if v is not None else ""
|
||||
|
||||
def norm(col: str, alt: Optional[List[str]] = None, max_len: int = 50) -> str:
|
||||
v = row.get(col)
|
||||
if alt:
|
||||
for k in alt:
|
||||
if v is None or (isinstance(v, str) and not v.strip()):
|
||||
v = row.get(k)
|
||||
s = normalize_sacarcomasenters(v) if v is not None else ""
|
||||
return s[:max_len] if s else ""
|
||||
|
||||
return {
|
||||
"NUMERO FACTURA": clip("NUMERO FACTURA", ["NUM FACTURA", "FACTURA"]),
|
||||
"LINEA FACTURA": clip("LINEA FACTURA", ["LINEA", "PARTIDA"]),
|
||||
"LINEA SERIE": clip("LINEA SERIE", ["RENGLON"]),
|
||||
"SERIE": norm("SERIE", max_len=MAX_LEN["serial_numbers"]),
|
||||
"MODELO": norm("MODELO", max_len=MAX_LEN["model"]),
|
||||
"NUM PARTE": norm("NUM PARTE", ["NUMPARTE", "NUMERO PARTE"]),
|
||||
"SUB MODELO": norm("SUB MODELO", ["SUBMODELO"], MAX_LEN["sub_model"]),
|
||||
"NUMERO ID": norm("NUMERO ID", ["NUMEROID"], MAX_LEN["number_id"]),
|
||||
}
|
||||
Reference in New Issue
Block a user