Merge branch 'feature/sitar-fractions-upgrade' into feature/scaf-partes-default
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user