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

feature/series-invoices-clarion-validations-csv

Browse Source
This commit is contained in:
hreyes
2026-03-09 07:28:05 -06:00
parent 2ac07b55b6
commit 4dc7fbd8ed
8 changed files with 745 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -35,7 +35,7 @@ def _get_redis():
@router.post("/upload/{model_target}", response_model=ImportJobResponse)
async def upload_import_file(
model_target: Literal["invoice_header", "invoice_details"],
model_target: Literal["invoice_header", "invoice_details", "invoice_series"],
file: UploadFile = File(...),
footer_config: Optional[str] = Form(None), # JSON string with settings
template_id: Optional[str] = Form(None), # id de la plantilla (ej. imp_temp_header) para respetar columnas

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

@@ -7,7 +7,7 @@ class ImportJobResponse(BaseModel):
message: str
class CommitRequest(BaseModel):
model_target: Literal["invoice_header", "invoice_details"]
model_target: Literal["invoice_header", "invoice_details", "invoice_series"]
class ImportJobStatus(BaseModel):
status: str

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

@@ -6,7 +6,7 @@ import json
import logging
import re
import unicodedata
from typing import Dict, Any, Optional, List
from typing import Dict, Any, Optional, List, Set, Tuple
from core.celery_app import celery_app
from core.database import CoreSessionLocal
@@ -165,12 +165,142 @@ def scan_file(self, job_id: str, model_target: str, config: str = None):
meta = common_meta.load_meta(file_path)
template_id = meta.get("template_id") or (
"imp_temp_header" if model_target == "invoice_header" else "imp_temp_details"
"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"
# --- Series de Importación Temporal: flujo específico (Clarion VALIDA_TODA / VALIDA_PARCIAL) ---
if model_target == "invoice_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 .validators.series_impo_temp import (
validate_row_series_impo_temp,
)
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:
# Invoice lookup: imp + TEM
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 == "TEM",
)
)
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)
# Existing series keys: (invoice_number, linea_factura, linea_serie)
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 == "TEM",
)
)
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 "",
})
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: List[Dict[str, Any]] = []
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_temp_series", normalize_header)
warnings_list: List[Dict[str, Any]] = []
err = validate_row_series_impo_temp(
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,
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", "")})
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 import scan failed: %s", e)
return {"status": "failed", "error": str(e)}
try:
from api.v1.modules.a76.invoices.models import InvoiceHeader
@@ -711,6 +841,237 @@ 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
use_series_flow = (
model_target == "invoice_series"
or meta.get("template_id") == "imp_temp_series"
)
# --- Series de Importación Temporal: commit (INSERT/UPDATE item_line_series) ---
if use_series_flow:
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 .validators.series_impo_temp import (
validate_row_series_impo_temp,
row_to_series_normalized,
)
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 == "TEM",
)
)
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)
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 == "TEM",
)
)
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 "",
})
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_temp_series", normalize_header)
err = validate_row_series_impo_temp(
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,
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(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:
skipped_invalid += 1
skipped_details.append({
"line": i,
"invoice": invoice_number,
"reason": f"Partida línea {linea_factura} no existe en la factura.",
})
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
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 import commit failed: %s", e)
return {"status": "failed", "error": str(e)}
try:
from api.v1.modules.a76.invoices.models import (
InvoiceHeader,

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

@@ -69,6 +69,19 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
# --- Partidas: Impo Def y Expo - misma estructura ---
"imp_def_details": None,
"exp_def_details": None,
# --- Series de Importación Temporal (EstructuraSeriesFacImpoTemp.xls) ---
# Clarion: NUMERO FACTURA, LINEA FACTURA, LINEA SERIE, SERIE, MODELO, NUM PARTE, SUB MODELO, NUMERO ID
"imp_temp_series": [
{"canonical": "NUMERO FACTURA", "aliases": ["NUM FACTURA", "FACTURA"]},
{"canonical": "LINEA FACTURA", "aliases": ["LINEA", "PARTIDA"]},
{"canonical": "LINEA SERIE", "aliases": ["RENGLON"]},
{"canonical": "SERIE", "aliases": ["NUMERO SERIE"]},
{"canonical": "MODELO"},
{"canonical": "NUM PARTE", "aliases": ["NUMPARTE", "NUMERO PARTE"]},
{"canonical": "SUB MODELO", "aliases": ["SUBMODELO"]},
{"canonical": "NUMERO ID", "aliases": ["NUMEROID"]},
{"canonical": "COL_EXTRA"}, # Optional; if has value → desfase warning (Clarion)
],
}

1
backend/api/v1/modules/a76/layouts_csv/facturas/validators/__init__.py Normal file
View File

@@ -0,0 +1 @@
# Validators for invoice CSV imports (header, details, series).

362
backend/api/v1/modules/a76/layouts_csv/facturas/validators/series_impo_temp.py Normal file
View File

@@ -0,0 +1,362 @@
"""
Validaciones CSV para Series de Importación Temporal.
Paridad Clarion: VALIDA_TODA_SERIES_IMPO_TEM, VALIDA_PARCIAL_SERIES_IMPO_TEM, LLENA_SERIES_IMPO_TEM.
Estructura: NUMERO FACTURA, LINEA FACTURA, LINEA SERIE, SERIE, MODELO, NUM PARTE, SUB MODELO, NUMERO ID.
"""
from typing import Dict, Any, Optional, Set, Tuple, List
# Longitudes máximas según modelo Serie (item_line_series)
MAX_LEN = {
"serial_numbers": 50,
"model": 50,
"sub_model": 50,
"number_id": 25,
}
APOSTROFE = "'"
def _clip(val: Any) -> str:
if val is None:
return ""
return str(val).strip()
def normalize_sacarcomasenters(val: Any) -> str:
"""Clarion SACARCOMASENTERS: quitar comas y saltos de línea."""
s = _clip(val)
s = s.replace(",", " ").replace("\n", " ").replace("\r", " ")
return " ".join(s.split())
def _check_desfase(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""Si COL_EXTRA tiene valor → advertencia desfase (no bloqueante)."""
val = _clip(row.get("COL_EXTRA"))
if not val:
return None
return {
"line": line_num,
"col": "COL_EXTRA",
"msg": "Advertencia: Podría existir un desfase en esta línea.",
"solution": "Revisar esta línea del archivo CSV y verificar cada campo esté en la posición correcta.",
"warning": True,
}
def _check_factura_vacia(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""NUMERO FACTURA (A) vacío → error bloqueante."""
val = _clip(row.get("NUMERO FACTURA") or row.get("NUM FACTURA") or row.get("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.",
"solution": "Capturar en la Celda A un número de Factura existente al cual desee agregar o actualizar series.",
}
return None
def _check_factura_existe(
invoice_number: str,
line_num: int,
invoice_id_by_number: Dict[str, int],
) -> Optional[Dict[str, Any]]:
"""Factura debe existir en BD (imp + TEM)."""
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.",
"solution": "Capturar en la Celda A un número de Factura existente al cual desee agregar o actualizar partidas.",
"identifier": "FAC_IMPO_TEM",
"fields": invoice_number,
}
return None
def _check_factura_no_actualizada(
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."""
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.",
"solution": "Capturar otro número de Factura de Importación Temporal o Desactualizar la factura.",
"identifier": "FAC_IMPO_TEM",
}
return None
def _check_linea_factura_vacia(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""LINEA FACTURA (B) vacía → error."""
val = _clip(row.get("LINEA FACTURA") or row.get("LINEA") or row.get("PARTIDA"))
if not val:
return {
"line": line_num,
"col": "LINEA FACTURA",
"msg": "Error: (Celda B) El campo de la línea de la partida está vacío y no se pueden hacer las validaciones.",
"solution": "Capturar en la Celda B la línea de la partida al cual desee agregar o actualizar información.",
}
return None
def _check_linea_serie_si_no_autonumerar(
row: Dict[str, Any],
line_num: int,
autonumerar: bool,
) -> Optional[Dict[str, Any]]:
"""Si Autonumerar = NO, LINEA SERIE (C) es obligatoria."""
if autonumerar:
return None
val = _clip(row.get("LINEA SERIE") or row.get("RENGLON"))
if not val:
return {
"line": line_num,
"col": "LINEA SERIE",
"msg": "Error: (Celda C) El campo de Renglón está vacío y no se pueden hacer las validaciones, ya que se tiene la opción de Autonumerar como NO.",
"solution": "Capturar en la Celda C el renglón de la serie la cual desee agregar o actualizar información.",
}
return None
def _warn_apostrofes(
row: Dict[str, Any],
line_num: int,
warnings: Optional[List[Dict[str, Any]]],
) -> None:
"""Advertencias si SERIE, MODELO o NUM PARTE contienen apostrofe (no bloqueante)."""
if warnings is None:
return
checks = [
("SERIE", "Número de Serie", row.get("SERIE")),
("MODELO", "Número de Modelo", row.get("MODELO")),
("NUM PARTE", "Número de Parte", row.get("NUM PARTE") or row.get("NUMPARTE") or row.get("NUMERO PARTE")),
]
for col, label, val in checks:
v = _clip(val) if val is not None else ""
if v and APOSTROFE in v:
warnings.append({
"line": line_num,
"col": col,
"msg": f"Advertencia: El {label}: {v} Contiene Apostrofes.",
"solution": "Se Omitirá el Apostrofe para Subir.",
"warning": True,
})
def _check_max_length(col: str, val: str, line_num: int, max_len: int) -> Optional[Dict[str, Any]]:
if not val or len(val) <= max_len:
return None
return {
"line": line_num,
"col": col,
"msg": f"Máximo {max_len} caracteres",
}
def valida_toda_series_impo_tem(
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_TEM: cuando no existe la partida/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 = _clip(row.get("SERIE"))
e = _clip(row.get("MODELO"))
f = _clip(row.get("NUM PARTE") or row.get("NUMPARTE") or row.get("NUMERO 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: {', '.join(obligatorios)}.",
"solution": "Revisar la línea del archivo y capturar los campos con la información correcta.",
}
# Longitudes (solo si hay valor)
def get_val(k: str) -> str:
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))
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(key)
if val:
err = _check_max_length(col, val, line_num, max_len)
if err:
return err
return None
def valida_parcial_series_impo_tem(
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_TEM: actualizar serie existente; campos vacíos se rellenan con existente.
Solo validar longitudes en campos no vacíos.
"""
def get_val(k: str) -> str:
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))
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(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_temp(
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],
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 Temporal.
Clarion: decisión VALIDA_TODA vs VALIDA_PARCIAL según autonumerar, actualizar y si la serie existe.
"""
# Desfase (solo advertencia)
desfase = _check_desfase(row, line_num)
if desfase and warnings is not None:
warnings.append(desfase)
# Factura vacía
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"}
# Factura existe
err = _check_factura_existe(invoice_number, line_num, invoice_id_by_number)
if err:
return err
# Factura no actualizada
err = _check_factura_no_actualizada(invoice_number, line_num, invoice_updated_by_number)
if err:
return err
# LINEA FACTURA vacía
err = _check_linea_factura_vacia(row, line_num)
if err:
return err
# Autonumerar NO y LINEA SERIE vacía
err = _check_linea_serie_si_no_autonumerar(row, line_num, autonumerar)
if err:
return err
# Advertencias apostrofes (no bloqueante)
_warn_apostrofes(row, line_num, warnings)
# Decisión VALIDA_TODA vs VALIDA_PARCIAL
linea_factura = _clip(row.get("LINEA FACTURA") or row.get("LINEA") or row.get("PARTIDA"))
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_tem(
row, line_num, existing_series_data[key], warnings
)
return valida_toda_series_impo_tem(
row, line_num, validar_series_exception, warnings
)
def row_to_series_normalized(row: Dict[str, Any]) -> Dict[str, Any]:
"""
Normaliza fila para guardar: SACARCOMASENTERS en D, E, F, G, H.
Clarion LLENA_SERIES_IMPO_TEM: asigna QueCSV a SerImp.
"""
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"]),
}

5
frontend/src/lib/config/csv-upload.ts
View File

@@ -355,8 +355,9 @@ export const importacionConfig: CsvUploadItem[] = [
title: 'Series',
icon: Hash,
group: 'Impo. Temp.',
modelTarget: 'InvoiceSeries',
disabled: true,
modelTarget: 'invoice_series',
templateId: 'imp_temp_series',
layoutModule: 'layouts_csv/facturas'
},
// Impo Def
{

2
frontend/src/routes/dashboard/csv-upload/+page.svelte
View File

@@ -431,7 +431,7 @@
footerConfig,
companyId,
opType,
config.id
config.templateId || config.id
);
if (res.data?.job_id) {
currentJobId = res.data.job_id;