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

feature/validaciones-clarion-facc-ame

Browse Source
This commit is contained in:
hreyes
2026-03-05 16:40:07 -07:00
parent 77bbcff223
commit cbfb2554cf
9 changed files with 334 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
backend/api/v1/modules/a76/layouts_csv/us_tariff_fractions/common/common_validators.py
View File

@@ -1,14 +1,18 @@
"""
Validadores reutilizables para import CSV de fracciones arancelarias americanas.
Paridad Clarion: Col A max 16, Col C en catálogo U.M., Col E PO/ME o vacío (default PO).
"""
from decimal import Decimal
from typing import Dict, Any, Optional
from typing import Dict, Any, Optional, Set
CODE_MAX = 16
PREFIX_MAX = 10
UNIT_MAX = 10
TYPE_MAX = 10
# Clarion: Tipo Advalorem PO (Porcentaje), ME (Costos Fijo Dlls) o vacío → PO
TIPO_ADVALOREM_VALIDOS = frozenset({"PO", "ME"})
def normalize_code(raw: Optional[str]) -> str:
"""Normalize fraction code: strip and remove dots/dashes, max 16 chars."""
@@ -60,3 +64,20 @@ def check_optional_decimal_min_zero(row: Dict[str, Any], col: str, line_num: int
except ValueError:
return {"line": line_num, "col": col, "msg": "Debe ser un número"}
return None
def check_tipo_advalorem(row: Dict[str, Any], col: str, line_num: int) -> Optional[Dict[str, Any]]:
"""Col E: solo PO, ME o vacío. Clarion mensaje."""
val = (row.get(col) or "").strip().upper()
if not val:
return None
if val in TIPO_ADVALOREM_VALIDOS:
return None
return {
"line": line_num,
"col": col,
"msg": (
f"Error: (Col. E) La opción de Tipo de Advalorem: {row.get(col) or ''} no es valido. "
"Capturar una opción valida: PO para Porcentaje, ME para Costos Fijo en Dlls o dejar el campo vacio y automaticamente se asigna Porcentaje."
),
}

52
backend/api/v1/modules/a76/layouts_csv/us_tariff_fractions/common/fk_loader.py Normal file
View File

@@ -0,0 +1,52 @@
"""
Carga de conjuntos FK para validación/mapeo de import CSV de Fracciones Americanas.
Clarion: U.M. (GUniMedida), códigos existentes de fracción americana (GFracAme).
"""
from typing import Set, Tuple
from core.database import CoreSessionLocal
def load_fa_fk_sets(
tenant_id: int,
company_id: int,
) -> Tuple[Set[str], Set[str]]:
"""
Carga conjuntos para validación CSV Fracciones Americanas (paridad Clarion).
Devuelve (valid_uom_codes, existing_fraction_codes).
- valid_uom_codes: códigos de Unidad de Medida (a76.units_of_measure, code UPPER, max 5 chars).
- existing_fraction_codes: códigos de USTariffFraction ya existentes por tenant/company.
"""
valid_uom_codes: Set[str] = set()
existing_fraction_codes: Set[str] = set()
try:
with CoreSessionLocal() as session:
from api.v1.modules.a76.general_catalogs.units_of_measure.models import UnitOfMeasure
from api.v1.modules.a76.general_catalogs.fractions.us_tariff_fractions.models import USTariffFraction
for row in (
session.query(UnitOfMeasure.code)
.filter(
UnitOfMeasure.tenant_id == tenant_id,
UnitOfMeasure.company_id == company_id,
)
.all()
):
if row[0]:
valid_uom_codes.add((row[0].strip() or "").upper()[:5])
for row in (
session.query(USTariffFraction.code)
.filter(
USTariffFraction.tenant_id == tenant_id,
USTariffFraction.company_id == company_id,
)
.all()
):
if row[0]:
existing_fraction_codes.add(row[0].strip())
except Exception as e:
import logging
logging.getLogger(__name__).warning("FA import: could not load FK sets: %s", e)
return valid_uom_codes, existing_fraction_codes

63
backend/api/v1/modules/a76/layouts_csv/us_tariff_fractions/common/mappers.py
View File

@@ -24,22 +24,67 @@ def _str_or_none(val: Any, max_len: Optional[int] = None) -> Optional[str]:
def row_to_us_tariff_fraction_data(
row_norm: Dict[str, Any], tenant_id: int, company_id: int
row_norm: Dict[str, Any], tenant_id: int, company_id: int, existing: Optional[Any] = None
) -> Dict[str, Any]:
"""Build dict for USTariffFraction model."""
"""
Build dict for USTariffFraction model.
Si existing está presente (modo Actualizar), campos vacíos en row_norm se rellenan desde existing (VALIDA_PARCIAL Clarion).
Col E vacío → type_code 'PO'.
"""
code = normalize_code(row_norm.get("FRACCION_ARANCELARIA"))
if not code:
return {}
fixed_cost_raw = parse_float_min_zero(row_norm.get("ADVALOREM_DLLS"))
fixed_cost = Decimal(str(round(fixed_cost_raw, 8))) if fixed_cost_raw is not None else None
def _prefijo() -> Optional[str]:
v = _str_or_none(row_norm.get("PREFIJO"), MAX_LEN["prefix"])
if v is not None:
return v
return getattr(existing, "prefix", None) if existing else None
def _unit() -> Optional[str]:
v = _str_or_none(row_norm.get("UNIDAD_DE_MEDIDA"), MAX_LEN["unit_of_measure"])
if v is not None:
return v
return getattr(existing, "unit_of_measure", None) if existing else None
def _desc() -> Optional[str]:
v = _str_or_none(row_norm.get("DESCRIPCION"))
if v is not None:
return v
return getattr(existing, "description", None) if existing else None
def _type_code() -> Optional[str]:
v = _str_or_none(row_norm.get("TIPO_DE_ADVALOREM"), MAX_LEN["type_code"])
if v is not None:
return v.upper() if v else "PO"
if existing is not None:
return getattr(existing, "type_code", None) or "PO"
return "PO"
def _adv() -> Optional[float]:
x = parse_float_min_zero(row_norm.get("ADVALOREM_PCT"))
if x is not None:
return x
return getattr(existing, "ad_valorem", None) if existing else None
def _fixed() -> Optional[Decimal]:
fixed_cost_raw = parse_float_min_zero(row_norm.get("ADVALOREM_DLLS"))
if fixed_cost_raw is not None:
return Decimal(str(round(fixed_cost_raw, 8)))
if existing is not None:
v = getattr(existing, "fixed_cost", None)
return Decimal(str(v)) if v is not None else None
return None
fixed_cost = _fixed()
return {
"tenant_id": tenant_id,
"company_id": company_id,
"code": code,
"prefix": _str_or_none(row_norm.get("PREFIJO"), MAX_LEN["prefix"]),
"type_code": _str_or_none(row_norm.get("TIPO_DE_ADVALOREM"), MAX_LEN["type_code"]),
"ad_valorem": parse_float_min_zero(row_norm.get("ADVALOREM_PCT")),
"prefix": _prefijo(),
"type_code": _type_code(),
"ad_valorem": _adv(),
"fixed_cost": fixed_cost,
"unit_of_measure": _str_or_none(row_norm.get("UNIDAD_DE_MEDIDA"), MAX_LEN["unit_of_measure"]),
"description": _str_or_none(row_norm.get("DESCRIPCION")),
"unit_of_measure": _unit(),
"description": _desc(),
}

3
backend/api/v1/modules/a76/layouts_csv/us_tariff_fractions/routes.py
View File

@@ -40,11 +40,13 @@ def _get_redis():
async def upload_import_file(
file: UploadFile = File(...),
company_id: int = Query(..., description="Company ID"),
actualizar: bool = Query(False, description="Modo Agregar/Actualizar (ACT); si False, Agregar/Reemplazar"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""
Fase 1: Subir CSV, guardar en Redis, encolar tarea de escaneo.
actualizar=True simula Clarion 'Agr./Actual.'; actualizar=False 'Agr./Reempl.'.
"""
try:
tenant_id = validate_access_to_resource(db, company_id, current_user)
@@ -63,6 +65,7 @@ async def upload_import_file(
"company_id": company_id,
"user_id": current_user.get("id"),
"template_id": "us_tariff_fractions",
"actualizar": actualizar,
}
try:

56
backend/api/v1/modules/a76/layouts_csv/us_tariff_fractions/tasks.py
View File

@@ -17,8 +17,9 @@ from ..common import meta as common_meta
from ..common import responses as common_responses
from ..common import csv_reader as common_csv_reader
from .template_config import row_from_template
from .validators import validate_row_us_tariff_fraction
from .validators import validate_row_us_tariff_fraction, validate_row_desfase_fa
from .common.mappers import row_to_us_tariff_fraction_data
from .common.fk_loader import load_fa_fk_sets
logger = logging.getLogger(__name__)
@@ -54,10 +55,15 @@ def _do_scan(job_id: str, progress_callback: Optional[Any] = None) -> Dict[str,
except ValueError as e:
return {"status": "failed", "error": str(e)}
meta_data = common_meta.load_meta(file_path)
actualizar = meta_data.get("actualizar", False)
valid_uom_codes, existing_fraction_codes = load_fa_fk_sets(tenant_id, company_id)
error_count = 0
processed_rows = 0
errors_detail: List[Dict[str, Any]] = []
error_lines_list: List[int] = []
warnings_detail: List[Dict[str, Any]] = []
try:
with open(error_path, "w", encoding="utf-8") as f_err:
@@ -65,8 +71,23 @@ def _do_scan(job_id: str, progress_callback: Optional[Any] = None) -> Dict[str,
if progress_callback and i % 500 == 0:
progress_callback(i, total_rows, error_count)
warn = validate_row_desfase_fa(row, i)
if warn and len(warnings_detail) < 500:
warnings_detail.append({
"line": warn["line"],
"col": warn.get("col", ""),
"msg": warn.get("msg", ""),
})
row_norm = _norm_row(row)
err = validate_row_us_tariff_fraction(row_norm, i)
err = validate_row_us_tariff_fraction(
row_norm,
i,
actualizar=actualizar,
existing_fraction_codes=existing_fraction_codes,
valid_uom_codes=valid_uom_codes,
raw_row=row,
)
if err:
error_count += 1
error_lines_list.append(err["line"])
@@ -85,7 +106,10 @@ def _do_scan(job_id: str, progress_callback: Optional[Any] = None) -> Dict[str,
logger.error("FA import scan failed: %s", e)
return {"status": "failed", "error": str(e)}
return common_responses.scan_result(job_id, processed_rows, error_count, errors_detail)
result = common_responses.scan_result(job_id, processed_rows, error_count, errors_detail)
if warnings_detail:
result["warnings"] = warnings_detail
return result
@celery_app.task(bind=True)
@@ -116,9 +140,14 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
except ValueError as e:
return {"status": "failed", "error": str(e)}
meta_data = common_meta.load_meta(file_path)
actualizar = meta_data.get("actualizar", False)
valid_uom_codes, existing_fraction_codes = load_fa_fk_sets(tenant_id, company_id)
from api.v1.modules.a76.general_catalogs.fractions.us_tariff_fractions.models import USTariffFraction
inserted_count = 0
updated_count = 0
skipped_invalid = 0
skipped_details: List[Dict[str, Any]] = []
meta_path = common_meta.get_meta_path(file_path)
@@ -130,7 +159,14 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
continue
row_norm = _norm_row(row)
err = validate_row_us_tariff_fraction(row_norm, i)
err = validate_row_us_tariff_fraction(
row_norm,
i,
actualizar=actualizar,
existing_fraction_codes=existing_fraction_codes,
valid_uom_codes=valid_uom_codes,
raw_row=row,
)
if err:
skipped_invalid += 1
skipped_details.append({
@@ -155,6 +191,9 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
.first()
)
if existing:
data = row_to_us_tariff_fraction_data(
row_norm, tenant_id, company_id, existing=existing
)
existing.prefix = data.get("prefix")
existing.type_code = data.get("type_code")
existing.ad_valorem = data.get("ad_valorem")
@@ -162,10 +201,11 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
existing.unit_of_measure = data.get("unit_of_measure")
existing.description = data.get("description")
session.add(existing)
updated_count += 1
else:
new_row = USTariffFraction(**data)
session.add(new_row)
inserted_count += 1
inserted_count += 1
try:
session.commit()
@@ -185,7 +225,7 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
meta_path=meta_path,
)
if inserted_count == 0 and skipped_invalid > 0:
if inserted_count == 0 and updated_count == 0 and skipped_invalid > 0:
return {
"status": "warning",
"inserted": 0,
@@ -196,7 +236,7 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
"skipped_details": skipped_details,
"message": f"No se insertaron registros. {skipped_invalid} rechazados.",
}
if inserted_count == 0:
if inserted_count == 0 and updated_count == 0:
return {
"status": "failed",
"error": "No hay registros válidos en el archivo CSV",
@@ -210,7 +250,7 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
return {
"status": "finished",
"inserted": inserted_count,
"updated": 0,
"updated": updated_count,
"skipped_invalid": skipped_invalid,
"skipped_duplicate": 0,
"skipped_missing_fk": 0,

5
backend/api/v1/modules/a76/layouts_csv/us_tariff_fractions/template_config.py
View File

@@ -14,9 +14,14 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
{"canonical": "TIPO_DE_ADVALOREM", "aliases": ["TIPO DE ADVALOREM", "TIPO"]},
{"canonical": "ADVALOREM_PCT", "aliases": ["ADVALOREM %", "ADVALOREM"]},
{"canonical": "ADVALOREM_DLLS", "aliases": ["ADVALOREM DLLS", "ADVALOREM DLL"]},
{"canonical": "COL_EXTRA", "aliases": ["DESFASE"]},
],
}
# Orden A..H para detectar desfase en 8ª columna (raw_row.values()[7])
TEMPLATE_ORDER = [item["canonical"] for item in TEMPLATE_COLUMNS["us_tariff_fractions"]]
DESFASE_COLUMN_INDEX = 7
def build_normalized_lookup(normalize_header_fn, template_id: str = "us_tariff_fractions") -> Dict[str, str]:
"""normalized_header -> canonical_name para plantilla us_tariff_fractions."""

4
backend/api/v1/modules/a76/layouts_csv/us_tariff_fractions/validators/__init__.py
View File

@@ -1,3 +1,3 @@
from .create import validate_row_us_tariff_fraction
from .create import validate_row_us_tariff_fraction, validate_row_desfase_fa
__all__ = ["validate_row_us_tariff_fraction"]
__all__ = ["validate_row_us_tariff_fraction", "validate_row_desfase_fa"]

151
backend/api/v1/modules/a76/layouts_csv/us_tariff_fractions/validators/common.py
View File

@@ -1,25 +1,113 @@
"""
Validaciones comunes de fila para import CSV de fracciones arancelarias americanas.
Paridad Clarion: desfase Col H, VALIDA_TODA_FRACCIONAME / VALIDA_PARCIAL_FRACCIONAME, VALIDACIONES_FRACCIONAME.
"""
from typing import Dict, Any, Optional
from typing import Dict, Any, Optional, Set
from ..template_config import DESFASE_COLUMN_INDEX
from ..common.common_validators import (
check_required_code,
normalize_code,
check_optional_max_length,
check_optional_decimal_min_zero,
check_tipo_advalorem,
PREFIX_MAX,
UNIT_MAX,
TYPE_MAX,
CODE_MAX,
)
from ..common.common_validators import TIPO_ADVALOREM_VALIDOS # noqa: F401 re-export
# Mensajes Clarion
MSG_COL_A_VACIO = (
"Error: (Col. A) La columna de Fracción Americana esta vacia y no se pueden hacer las validaciones. "
"Capturar en la Columna A una Fracción Americana nueva o una ya existente al cual desee actualizar campos"
)
MSG_COL_A_LONGITUD = (
"Capturar en la columna A una Fraccion Arancelaria de 16 caracteres como máximo."
)
MSG_COL_D_OBLIGATORIO = (
"Existen campos vacios que son obligatorios, es la (Col.D) Descripción. "
"Revisar la línea del archivo y capturar los campos con la información correcta."
)
MSG_FRACCION_NO_EXISTE = (
"Error: (Col. A) La Fraccion Americana no existe."
)
MSG_DESFASE = "Advertencia: Podría existir un desfase en esta línea."
MSG_DESFASE_SOLUCION = "Revisar esta línea del archivo CSV y verificar cada campo esté en la posición correcta."
def validate_row_us_tariff_fraction(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
def validate_row_desfase_fa(
raw_row: Optional[Dict[str, Any]], line_num: int
) -> Optional[Dict[str, Any]]:
"""
Valida una fila de CSV de fracción arancelaria americana.
FRACCION_ARANCELARIA requerida (max 16); PREFIJO, UNIDAD_DE_MEDIDA, TIPO_DE_ADVALOREM opcionales (max 10);
ADVALOREM_PCT y ADVALOREM_DLLS opcionales numéricos >= 0.
Si la fila tiene 8+ columnas y la 8ª (Col H) tiene valor → advertencia (no bloqueante).
Devuelve dict con severity=warning para que el caller pueda acumularlo en warnings_detail.
"""
err = check_required_code(row, "FRACCION_ARANCELARIA", line_num)
if not raw_row:
return None
values_ordered = list(raw_row.values())
if len(values_ordered) <= DESFASE_COLUMN_INDEX:
return None
if not (values_ordered[DESFASE_COLUMN_INDEX] or "").strip():
return None
return {
"line": line_num,
"col": "COL_EXTRA",
"msg": MSG_DESFASE,
"solution": MSG_DESFASE_SOLUCION,
"severity": "warning",
}
def _validate_col_a_required(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""Col A obligatoria; mensaje Clarion."""
code_raw = (row.get("FRACCION_ARANCELARIA") or "").strip()
if not code_raw:
return {"line": line_num, "col": "FRACCION_ARANCELARIA", "msg": MSG_COL_A_VACIO}
code_norm = normalize_code(code_raw)
if not code_norm:
return {"line": line_num, "col": "FRACCION_ARANCELARIA", "msg": MSG_COL_A_VACIO}
if len(code_norm) > CODE_MAX:
return {
"line": line_num,
"col": "FRACCION_ARANCELARIA",
"msg": f"Error: (Col. A) La Fraccion Americana: {code_raw} supera la longitud de caracteres. {MSG_COL_A_LONGITUD}",
}
return None
def _validate_col_d_required(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
"""Col D obligatoria en validación completa (VALIDA_TODA)."""
if (row.get("DESCRIPCION") or "").strip():
return None
return {"line": line_num, "col": "DESCRIPCION", "msg": MSG_COL_D_OBLIGATORIO}
def _validate_uom_catalog(
row: Dict[str, Any], line_num: int, valid_uom_codes: Optional[Set[str]]
) -> Optional[Dict[str, Any]]:
"""Col C: si no vacía, debe existir en catálogo U.M. (Clarion GUniMedida)."""
val = (row.get("UNIDAD_DE_MEDIDA") or "").strip().upper()
if not val or valid_uom_codes is None:
return None
if val in valid_uom_codes:
return None
return {
"line": line_num,
"col": "UNIDAD_DE_MEDIDA",
"msg": (
f"Error: (Col. C) La Unidad de Medida: {row.get('UNIDAD_DE_MEDIDA') or ''} no existe en el Catálogo de U.M. "
"Revisar esta Unidad de Medida en el archivo, en caso de ser correcta dar la de alta en el Catálogo de U.M."
),
}
def _validaciones_fraccioname(
row: Dict[str, Any],
line_num: int,
valid_uom_codes: Optional[Set[str]],
) -> Optional[Dict[str, Any]]:
"""VALIDACIONES_FRACCIONAME: Col A len ≤16, Col C en catálogo U.M., Col E PO/ME/vacío, F/G ≥0."""
err = _validate_col_a_required(row, line_num)
if err:
return err
err = check_optional_max_length(row, "PREFIJO", PREFIX_MAX, line_num)
@@ -28,7 +116,10 @@ def validate_row_us_tariff_fraction(row: Dict[str, Any], line_num: int) -> Optio
err = check_optional_max_length(row, "UNIDAD_DE_MEDIDA", UNIT_MAX, line_num)
if err:
return err
err = check_optional_max_length(row, "TIPO_DE_ADVALOREM", TYPE_MAX, line_num)
err = _validate_uom_catalog(row, line_num, valid_uom_codes)
if err:
return err
err = check_tipo_advalorem(row, "TIPO_DE_ADVALOREM", line_num)
if err:
return err
err = check_optional_decimal_min_zero(row, "ADVALOREM_PCT", line_num)
@@ -38,3 +129,45 @@ def validate_row_us_tariff_fraction(row: Dict[str, Any], line_num: int) -> Optio
if err:
return err
return None
def validate_row_us_tariff_fraction(
row: Dict[str, Any],
line_num: int,
*,
actualizar: bool = False,
existing_fraction_codes: Optional[Set[str]] = None,
valid_uom_codes: Optional[Set[str]] = None,
raw_row: Optional[Dict[str, Any]] = None,
) -> Optional[Dict[str, Any]]:
"""
Valida una fila de CSV de fracción arancelaria americana.
- raw_row: si se pasa y la 8ª columna tiene valor, se devuelve advertencia de desfase (no bloqueante; el caller decide si la trata como error).
- Col A vacía → error.
- Si actualizar y fracción no existe en existing_fraction_codes → error "La Fracción Americana no existe".
- Si actualizar y fracción existe → VALIDA_PARCIAL (solo VALIDACIONES_FRACCIONAME; Col D no obligatoria).
- Si no actualizar → VALIDA_TODA (Col D obligatoria + VALIDACIONES_FRACCIONAME).
"""
# Desfase: advertencia (no bloqueante por defecto; se devuelve como error para que scan la registre en errors_detail pero no bloquea commit si no está en error_lines)
# Plan: "advertencia no bloqueante" → no añadimos a error_lines; guardamos en warnings. Para simplificar, desfase lo devolvemos como error de severidad warning y en tasks no lo añadimos a error_lines (solo a warnings_detail). Mejor: desfase retornamos None (no error) y el caller puede llamar a validate_row_desfase_fa por separado y acumular warnings. Así no bloqueamos. Entonces en validate_row_us_tariff_fraction no llamamos desfase como error; en tasks llamamos primero validate_row_desfase_fa y si hay warning lo guardamos en warnings_detail, luego llamamos validate_row_us_tariff_fraction que puede devolver error. OK.
# So we don't return desfase from validate_row_us_tariff_fraction; tasks will call validate_row_desfase_fa and collect warnings. So no change here for desfase inside this function.
err = _validate_col_a_required(row, line_num)
if err:
return err
code_norm = normalize_code((row.get("FRACCION_ARANCELARIA") or "").strip())
fraction_exists = (
existing_fraction_codes is not None and code_norm in existing_fraction_codes
)
if actualizar and not fraction_exists:
return {"line": line_num, "col": "FRACCION_ARANCELARIA", "msg": MSG_FRACCION_NO_EXISTE}
use_full = not actualizar or not fraction_exists
if use_full:
err = _validate_col_d_required(row, line_num)
if err:
return err
return _validaciones_fraccioname(row, line_num, valid_uom_codes)

8
backend/api/v1/modules/a76/layouts_csv/us_tariff_fractions/validators/create.py
View File

@@ -1,6 +1,10 @@
"""
Punto de entrada de validación para import de una fila fracción arancelaria americana.
Firma: validate_row_us_tariff_fraction(row, line_num, *, actualizar=False, existing_fraction_codes=None, valid_uom_codes=None, raw_row=None).
"""
from .common import validate_row_us_tariff_fraction
from .common import (
validate_row_us_tariff_fraction,
validate_row_desfase_fa,
)
__all__ = ["validate_row_us_tariff_fraction"]
__all__ = ["validate_row_us_tariff_fraction", "validate_row_desfase_fa"]