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

feature/series-clarion-csv-validaciones-expo

Browse Source
This commit is contained in:
hreyes
2026-03-10 17:03:24 -06:00
parent 6d56448f9f
commit 3692fd890d
9 changed files with 706 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
backend/api/v1/modules/a76/layouts_csv/exportacion/routes.py
View File

@@ -33,7 +33,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),
template_id: Optional[str] = Form(None),
@@ -56,13 +56,18 @@ async def upload_import_file(
contents = await file.read()
file_key, meta_key, _ = common_storage.storage_keys(JOB_TYPE, job_id)
default_template = (
"exp_def_header" if model_target == "invoice_header"
else "exp_def_series" if model_target == "invoice_series"
else "exp_def_partidas"
)
meta_data = {
"tenant_id": tenant_id,
"company_id": company_id,
"user_id": current_user.get("id"),
"footer_config": footer_config,
"operation_type": operation_type or "exp",
"template_id": template_id or ("exp_def_header" if model_target == "invoice_header" else "exp_def_partidas"),
"template_id": template_id or default_template,
}
try:

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

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

8
backend/api/v1/modules/a76/layouts_csv/exportacion/tasks.py
View File

@@ -50,6 +50,10 @@ def scan_file(self, job_id: str, model_target: str, config: str = None):
from api.v1.modules.a76.layouts_csv.facturas.tasks import _do_scan_file
return _do_scan_file(job_id, "invoice_details", config, job_type_override="exp")
if model_target == "invoice_series":
from api.v1.modules.a76.layouts_csv.facturas.tasks import _do_scan_file
return _do_scan_file(job_id, "invoice_series", config, job_type_override="exp")
# Fallback (e.g. unknown model_target)
file_path = _ensure_file(job_id)
if not file_path:
@@ -96,6 +100,10 @@ def insert_valid_rows(self, job_id: str, model_target: str):
from api.v1.modules.a76.layouts_csv.facturas.tasks import _do_insert_valid_rows
return _do_insert_valid_rows(job_id, "invoice_details", job_type_override="exp")
if model_target == "invoice_series":
from api.v1.modules.a76.layouts_csv.facturas.tasks import _do_insert_valid_rows
return _do_insert_valid_rows(job_id, "invoice_series", job_type_override="exp")
# Fallback: stub sin inserción
file_path = _ensure_file(job_id)
if not file_path:

12
backend/api/v1/modules/a76/layouts_csv/exportacion/template_config.py
View File

@@ -87,6 +87,18 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
{"canonical": "NUM. PARTE", "aliases": ["NUMPARTE", "NUMERO PARTE", "NUM PARTE"]},
{"canonical": "ORDEN DE COMPRA", "aliases": ["ORDENCOMPRA", "ORDEN DE VENTA"]},
],
# Series de exportación definitiva (misma estructura que imp_def_series; Clarion SERIES EXPO)
"exp_def_series": [
{"canonical": "NUMERO FACTURA", "aliases": ["NUM FACTURA", "FACTURA", "FACTURA EXPO"]},
{"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"},
],
}

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

@@ -227,6 +227,8 @@ def _do_scan_file(job_id: str, model_target: str, config: Optional[str] = None,
template_id = "exp_def_header"
if job_type_override == "exp" and model_target == "invoice_details":
template_id = "exp_def_partidas"
if job_type_override == "exp" and model_target == "invoice_series":
template_id = "exp_def_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"
@@ -409,6 +411,183 @@ def _do_scan_file(job_id: str, model_target: str, config: Optional[str] = None,
logger.exception("Series importación definitiva scan failed: %s", e)
return {"status": "failed", "error": str(e)}
# --- Series de Exportación Definitiva: flujo exp_def_series (Clarion VALIDA_TODA_SERIES_EXPO / VALIDA_PARCIAL) ---
if model_target == "invoice_series" and template_id == "exp_def_series":
logger.info("Series expo scan: running validation for job %s", job_id)
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 api.v1.modules.a76.general_catalogs.company.models import Company
from .validators.series_expo import validate_row_series_expo
_fc = parse_footer_config(meta.get("footer_config"))
autonumerar = meta.get("autonumerar", True)
actualizar = meta.get("actualizar", False)
validar_series_exception = meta.get("validar_series", False)
if _fc:
if "autonumerar" in _fc:
autonumerar = bool(_fc["autonumerar"])
elif _fc.get("autonumber_series", "true") is not None:
autonumerar = str(_fc.get("autonumber_series", "true")).lower() in ("true", "1", "si", "sí", "yes")
if "actualizar" in _fc:
actualizar = bool(_fc["actualizar"])
if "validar_series" in _fc:
validar_series_exception = bool(_fc["validar_series"])
RFC_EXCEPTION_EGM = {"EGM0303257J1"}
with CoreSessionLocal() as session:
company = session.query(Company).filter(Company.id == company_id).first()
company_rfc = (company.rfc or "").strip().upper() if company else ""
q_inv_expo = (
session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.is_updated)
.filter(
InvoiceHeader.tenant_id == tenant_id,
InvoiceHeader.company_id == company_id,
InvoiceHeader.operation_type == "exp",
)
)
invoice_id_by_number: Dict[str, int] = {}
invoice_updated_by_number: Dict[str, bool] = {}
for num, iid, is_upd in q_inv_expo.all():
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 == "exp",
)
)
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 == "exp",
)
)
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] = {}
invoice_numbers_from_csv: 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)
for row in reader:
row_norm = row_from_template(row, "exp_def_series", normalize_header)
inv = (row_norm.get("NUMERO FACTURA") or row_norm.get("NUM FACTURA") or row_norm.get("FACTURA EXPO") or "").strip()
if inv:
invoice_numbers_from_csv.add(inv)
if company_rfc in RFC_EXCEPTION_EGM:
rfc_exception_updated: Set[str] = invoice_numbers_from_csv
else:
rfc_exception_updated = set()
with open(file_path, "r", encoding="utf-8-sig") as f_in, open(error_path, "w", encoding="utf-8") as f_err:
f_in.seek(0)
try:
dialect = csv.Sniffer().sniff(f_in.read(2048), delimiters=",;\t")
except Exception:
dialect = "excel"
f_in.seek(0)
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, "exp_def_series", normalize_header)
warnings_list: List[Dict[str, Any]] = []
err = validate_row_series_expo(
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,
rfc_exception_updated=rfc_exception_updated,
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 row_norm.get("FACTURA EXPO") 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
common_storage.store_error_lines(effective_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 exportació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:
@@ -3590,7 +3769,7 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
# 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", "cmex_series")
or meta.get("template_id") in ("imp_temp_series", "imp_def_series", "cmex_series", "exp_def_series")
)
_footer_for_series = parse_footer_config(meta.get("footer_config")) or {}
@@ -3603,6 +3782,182 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
or _inv_type_series in ("DEF", "MATDE", "EXDEF")
)
)
use_expo_series_commit = use_series_flow and meta.get("template_id") == "exp_def_series"
# --- Series de Exportación Definitiva: commit (INSERT/UPDATE item_line_series para facturas exp) ---
if use_expo_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 .validators.series_expo import row_to_series_normalized_expo
actualizar = meta.get("actualizar", False)
autonumerar = meta.get("autonumerar", True)
_fc = parse_footer_config(meta.get("footer_config"))
if _fc:
if "actualizar" in _fc:
actualizar = bool(_fc["actualizar"])
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")
with CoreSessionLocal() as session:
q_inv = (
session.query(InvoiceHeader.invoice_number, InvoiceHeader.id)
.filter(
InvoiceHeader.tenant_id == tenant_id,
InvoiceHeader.company_id == company_id,
InvoiceHeader.operation_type == "exp",
)
)
invoice_id_by_number: Dict[str, int] = {}
for num, iid in q_inv.all():
if num:
invoice_id_by_number[str(num).strip()] = iid
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:
skipped_invalid += 1
continue
row_norm = row_from_template(row, "exp_def_series", normalize_header)
data = row_to_series_normalized_expo(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
skipped_details.append({
"line": i,
"invoice": invoice_number or "(vacío)",
"reason": "Factura de exportación no encontrada.",
})
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: {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
session.commit()
common_storage.cleanup_import_job(effective_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 exportación definitiva commit failed: %s", e)
return {"status": "failed", "error": str(e)}
# --- Series de Importación Definitiva: commit (INSERT/UPDATE item_line_series para facturas DEF o MEX) ---
if use_def_series_commit:
@@ -4158,6 +4513,8 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
)
if job_type_override == "exp" and model_target == "invoice_details":
_template_id_insert = "exp_def_partidas"
if job_type_override == "exp" and model_target == "invoice_series":
_template_id_insert = "exp_def_series"
if model_target == "invoice_header" and _template_id_insert == "imp_def_header":
inv_type_value = "DEF"
if model_target == "invoice_header" and _template_id_insert == "cmex_header":

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

@@ -243,6 +243,19 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
{"canonical": "NUMERO ID", "aliases": ["NUMEROID"]},
{"canonical": "COL_EXTRA"},
],
# --- Series de Exportación Definitiva (Clarion VALIDA_TODA_SERIES_EXPO / VALIDA_PARCIAL_SERIES_EXPO) ---
# Misma estructura: NUMERO FACTURA, LINEA FACTURA, LINEA SERIE, SERIE, MODELO, NUM PARTE, SUB MODELO, NUMERO ID
"exp_def_series": [
{"canonical": "NUMERO FACTURA", "aliases": ["NUM FACTURA", "FACTURA", "FACTURA EXPO"]},
{"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"},
],
}
@@ -260,6 +273,8 @@ def _resolve_template_columns(template_id: str) -> Optional[List[Dict[str, Any]]
return TEMPLATE_COLUMNS.get("imp_temp_details")
if template_id == "cmex_series":
return TEMPLATE_COLUMNS.get("imp_def_series")
if template_id == "exp_def_series":
return TEMPLATE_COLUMNS.get("exp_def_series")
return None

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

@@ -17,6 +17,10 @@ from .series_impo_def import (
validate_row_series_impo_def,
row_to_series_normalized_def,
)
from .series_expo import (
validate_row_series_expo,
row_to_series_normalized_expo,
)
__all__ = [
"validate_row_encabezados_impo_temp",
@@ -27,6 +31,8 @@ __all__ = [
"validate_row_partidas_impo_def",
"validate_row_series_impo_def",
"row_to_series_normalized_def",
"validate_row_series_expo",
"row_to_series_normalized_expo",
"row_to_transport_type_clarion",
"parse_pedimento_col_a",
"parse_pedimento_col_a_impo_def",

296
backend/api/v1/modules/a76/layouts_csv/facturas/validators/series_expo.py Normal file
View File

@@ -0,0 +1,296 @@
"""
Validaciones CSV para Series de Exportación Definitiva.
Paridad Clarion: VALIDA_TODA_SERIES_EXPO, VALIDA_PARCIAL_SERIES_EXPO, LLENA_SERIES_EXPO.
Estructura: NUMERO FACTURA, LINEA FACTURA, LINEA SERIE, SERIE, MODELO, NUM PARTE, SUB MODELO, NUMERO ID.
Reutiliza helpers de series_impo_temp y series_impo_def; factura = exportación (FAC_EXPO), rfc_exception_updated.
"""
from typing import Dict, Any, Optional, Set, Tuple, List
from .series_impo_temp import (
_clip,
normalize_sacarcomasenters,
_check_desfase,
_check_linea_factura_vacia,
_check_linea_serie_si_no_autonumerar,
_warn_apostrofes,
_check_max_length,
MAX_LEN,
)
from .series_impo_def import (
_check_partida_existe_en_factura,
_check_cantidad_series_vs_partida,
_get_val_def,
)
def _check_factura_vacia_expo(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""NUMERO FACTURA (A) vacío → error. Mensaje Factura de Exportación."""
val = _clip(row.get("NUMERO FACTURA") or row.get("NUM FACTURA") or row.get("FACTURA") or row.get("FACTURA EXPO"))
if not val:
return {
"line": line_num,
"col": "NUMERO FACTURA",
"msg": (
f"Error: (Celda A{line_num}) La Factura de Exportación está vacía y no se pueden hacer las validaciones. "
),
"solution": (
f"Capturar en la Celda A{line_num} un número de Factura existente "
f"al cual desee agregar o actualizar series"
),
}
return None
def _check_factura_existe_expo(
invoice_number: str,
line_num: int,
invoice_id_by_number: Dict[str, int],
) -> Optional[Dict[str, Any]]:
"""Factura debe existir en BD (catálogo exportación)."""
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 Exportación {invoice_number} "
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 "
f"al cual desee agregar o actualizar series"
),
"identifier": "FAC_EXPO",
"fields": invoice_number,
}
return None
def _check_factura_no_actualizada_expo(
invoice_number: str,
line_num: int,
invoice_updated_by_number: Dict[str, bool],
rfc_exception_updated: Set[str],
) -> Optional[Dict[str, Any]]:
"""Si factura ya actualizada (Estatus AC) no se pueden hacer cambios; excepción por RFC (ej. EGM0303257J1)."""
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{line_num}) La Factura de Exportació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 Exportación o Desactualizar la factura.",
"identifier": "FAC_EXPO",
}
return None
def valida_toda_series_expo(
row: Dict[str, Any],
line_num: int,
validar_series_exception: bool,
warnings: Optional[List[Dict[str, Any]]],
) -> Optional[Dict[str, Any]]:
"""
VALIDA_TODA_SERIES_EXPO: 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_expo(
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_EXPO: 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_expo(
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],
rfc_exception_updated: Set[str],
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 Exportación Definitiva.
Clarion: decisión VALIDA_TODA vs VALIDA_PARCIAL según autonumerar, actualizar y si la serie existe.
rfc_exception_updated: set de números de factura que se consideran no actualizadas (ej. EGM0303257J1).
"""
desfase = _check_desfase(row, line_num)
if desfase and warnings is not None:
warnings.append(desfase)
err = _check_factura_vacia_expo(row, line_num)
if err:
return err
invoice_number = _clip(
row.get("NUMERO FACTURA") or row.get("NUM FACTURA") or row.get("FACTURA") or row.get("FACTURA EXPO")
)
if not invoice_number:
return {"line": line_num, "col": "NUMERO FACTURA", "msg": "Requerido"}
err = _check_factura_existe_expo(invoice_number, line_num, invoice_id_by_number)
if err:
return err
err = _check_factura_no_actualizada_expo(
invoice_number, line_num, invoice_updated_by_number, rfc_exception_updated
)
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_expo(
row, line_num, existing_series_data[key], warnings
)
return valida_toda_series_expo(
row, line_num, validar_series_exception, warnings
)
def row_to_series_normalized_expo(row: Dict[str, Any]) -> Dict[str, Any]:
"""
Normaliza fila para guardar: SACARCOMASENTERS en D, E, F, G, H.
Clarion LLENA_SERIES_EXPO: asigna QueCSV a SerExpo.
NUM PARTE se valida pero el modelo Serie no tiene campo parte.
"""
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", "FACTURA EXPO"]),
"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

@@ -444,8 +444,9 @@ export const exportacionConfig: CsvUploadItem[] = [
title: 'Series',
icon: Hash,
group: 'Expo. Def./Cam. Reg.',
modelTarget: 'InvoiceSeries',
disabled: true,
modelTarget: 'invoice_series',
templateId: 'exp_def_series',
layoutModule: 'layouts_csv/exportacion'
},
{
id: 'exp_def_nodes',