Merge branch 'feature/sitar-fractions-upgrade' into feature/visible-clicks-csv

This commit is contained in:
2026-04-29 08:24:39 -06:00
16 changed files with 736 additions and 63 deletions

View File

@@ -9,6 +9,7 @@ from sqlalchemy.exc import IntegrityError
from fastapi import HTTPException
import zlib
import logging
import re
from .models import TariffFraction
from .dto import TariffFractionCreateDTO, TariffFractionUpdateDTO
@@ -22,7 +23,42 @@ logger = logging.getLogger(__name__)
class TariffFractionMapper:
"""Helper to map Sitar responses to Local domain objects"""
@staticmethod
def _digits_only(value: Optional[str]) -> str:
return re.sub(r"\D", "", (value or "").strip())
@staticmethod
def _format_mx_fraction(code: str) -> str:
if code.isdigit() and len(code) == 8:
return f"{code[:2]}.{code[2:4]}.{code[4:6]}.{code[6:]}"
if code.isdigit() and len(code) == 6:
return f"{code[:2]}.{code[2:4]}.{code[4:]}"
return code
@staticmethod
def _format_usa_fraction(code: str) -> str:
if code.isdigit() and len(code) == 10:
return f"{code[:4]}.{code[4:6]}.{code[6:8]}.{code[8:]}"
if code.isdigit() and len(code) == 8:
return f"{code[:4]}.{code[4:6]}.{code[6:]}"
return code
@staticmethod
def _normalized_pair(raw_code: Optional[str], raw_fraction: Optional[str], formatter) -> Tuple[str, str]:
"""Return (code_without_separators, formatted_fraction)."""
code = TariffFractionMapper._digits_only(raw_code)
fraction = (raw_fraction or "").strip()
if not code:
code = TariffFractionMapper._digits_only(fraction)
if not fraction:
fraction = formatter(code)
elif "." not in fraction and "-" not in fraction:
fraction = formatter(TariffFractionMapper._digits_only(fraction))
if not fraction:
fraction = formatter(code)
return code, fraction
@staticmethod
def to_domain(fraccion: FraccionesResponse) -> TariffFraction:
# Generate ID: Use SYSID if available, else composite hash of code + nico
@@ -33,22 +69,9 @@ class TariffFractionMapper:
unique_str = f"{fraccion.FRACCION}-{fraccion.NICO}"
fake_id = zlib.crc32(unique_str.encode('utf-8'))
# UX Enhauncement: Sitar API returns empty strings for some fields.
# We fill them with fallbacks so the frontend table isn't 90% empty.
code_val = fraccion.FRACCION
# Formatting Logic: if FRACCIONPUNTO is empty, try to format code_val
formatted_fraction = code_val
if fraccion.FRACCIONPUNTO:
formatted_fraction = fraccion.FRACCIONPUNTO
elif code_val and code_val.isdigit() and len(code_val) == 8:
# Standard 8 digit format: XX.XX.XX.XX
formatted_fraction = f"{code_val[:2]}.{code_val[2:4]}.{code_val[4:6]}.{code_val[6:]}"
elif code_val and code_val.isdigit() and len(code_val) == 6:
# 6 digit (subheading): XX.XX.XX
formatted_fraction = f"{code_val[:2]}.{code_val[2:4]}.{code_val[4:]}"
fraction_val = formatted_fraction
code_val, fraction_val = TariffFractionMapper._normalized_pair(
fraccion.FRACCION, fraccion.FRACCIONPUNTO, TariffFractionMapper._format_mx_fraction
)
description_val = fraccion.DESCRIPCION if fraccion.DESCRIPCION else "(Sin descripción)"
tf = TariffFraction(
@@ -73,10 +96,15 @@ class TariffFractionMapper:
@staticmethod
def to_domain_usa(item: FraccionesUSAResponse) -> TariffFraction:
"""Map US Fraction to Domain"""
code_val, fraction_val = TariffFractionMapper._normalized_pair(
item.FRACCION_SIN_PUNTO,
item.FRACCION_CON_PUNTO or item.FRACCION_MOSTRAR,
TariffFractionMapper._format_usa_fraction,
)
return TariffFraction(
id=item.CONSECUTIVO,
code=item.FRACCION_SIN_PUNTO or "",
fraction=item.FRACCION_CON_PUNTO or "",
code=code_val,
fraction=fraction_val,
description=item.DESCRIPCION or "(Sin descripción)",
nico=None, # Not applicable
umt=item.UNIDADCANTIDAD,
@@ -183,28 +211,27 @@ class TariffFractionService:
# Map filters
sitar_fraccion = None
sitar_nico = None
sitar_description = None
# Default level logic
level_filter = 5 # Default legacy
# Legacy parity: base query is always Nivel = 5 unless caller explicitly requests another level.
level_filter = 5
if filters and filters.get("level") is not None:
level_filter = filters["level"]
# Allow disabling level filter explicitly
# UI compatibility: level -1 means "sin filtro de nivel".
if level_filter == -1:
level_filter = None
if filters:
if filters.get("search"):
term = filters["search"]
# Heuristic: if search starts with digit (after removing dots), treat as code/fraccion/nico
# This covers "0101", "01.01", "020691A"
term = str(filters["search"]).strip()
# Legacy-like behavior:
# - Numeric search targets fracción first.
# - Text search targets descripción.
clean_term = term.replace(".", "")
if clean_term and clean_term[0].isdigit():
if clean_term.isdigit():
sitar_fraccion = clean_term
else:
# Attempt description search via API first
logger.info(f"Search term '{term}' identified as text. Attempting API description search.")
pass
sitar_description = term
if filters.get("code"):
sitar_fraccion = filters["code"]
@@ -212,14 +239,8 @@ class TariffFractionService:
sitar_fraccion = filters["fraction"]
if filters.get("nico"):
sitar_nico = filters["nico"]
# Determine description filter
sitar_description = None
# Only use description if we didn't use it as code above
if filters and filters.get("search"):
clean_term = filters["search"].replace(".", "")
if not (clean_term and clean_term[0].isdigit()):
sitar_description = filters["search"]
if sitar_fraccion is not None:
sitar_fraccion = str(sitar_fraccion).replace(".", "").strip()
# Note: Sitar search might not return total count.
# We fetch page items. Pagination might be tricky if Sitar doesn't return total.
@@ -239,6 +260,8 @@ class TariffFractionService:
# Map items
items = [TariffFractionMapper.to_domain(item) for item in sitar_items]
# Legacy browse behavior: keep table in ascending fracción order.
items = sorted(items, key=lambda row: ((row.code or ""), (row.nico or "")))
# Estimate total (Sitar service doesn't return total currently)
# If we got full limit, assume there are more.
@@ -289,8 +312,6 @@ class TariffFractionService:
query = query.filter(TariffFraction.umt.ilike(f"%{filters['umt']}%"))
total = query.count()
# Add deterministic sort order
query = query.order_by(TariffFraction.fraction)
items = query.offset(skip).limit(limit).all()
return items, total

View File

@@ -6,6 +6,7 @@ from datetime import datetime
from typing import Optional, Any
from pydantic import BaseModel, Field, ConfigDict, model_validator
import re
class USTariffFractionCreateDTO(BaseModel):
@@ -62,10 +63,20 @@ class USTariffFractionResponseDTO(BaseModel):
if raw_code:
code_str = str(raw_code)
# fraction keeps the original formatted string
fraction = code_str
# code strips dots and hyphens
code = code_str.replace(".", "").replace("-", "")
fraction_raw = ""
if isinstance(data, dict):
fraction_raw = str(data.get("fraction") or "")
else:
fraction_raw = str(getattr(data, "fraction", "") or "")
code = re.sub(r"[.\s-]", "", code_str)
fraction = fraction_raw.strip() or code_str
if "." not in fraction and "-" not in fraction:
only_digits = re.sub(r"[.\s-]", "", fraction)
if len(only_digits) == 10:
fraction = f"{only_digits[:4]}.{only_digits[4:6]}.{only_digits[6:8]}.{only_digits[8:]}"
elif len(only_digits) == 8:
fraction = f"{only_digits[:4]}.{only_digits[4:6]}.{only_digits[6:]}"
if isinstance(data, dict):
data["code"] = code

View File

@@ -28,6 +28,7 @@ def _sitar_row_to_us_response_payload(item: FraccionesUSAResponse) -> dict:
return {
"id": item.CONSECUTIVO,
"code": canon,
"fraction": item.FRACCION_CON_PUNTO or item.FRACCION_MOSTRAR or canon,
"prefix": item.FRACCION_SIN_PUNTO,
"type_code": str(item.NIVEL) if item.NIVEL is not None else None,
"ad_valorem": american_fraction_ad_valorem_from_row(item),

View File

@@ -128,7 +128,7 @@ class USTariffFractionService:
)
total = query.count()
items = query.order_by(USTariffFraction.code).offset(skip).limit(limit).all()
items = query.offset(skip).limit(limit).all()
return items, total