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

Merge pull request 'development' (#144) from development into feature/item-calculations

Browse Source 
Reviewed-on: ADUANASOFT/anexo76#144
This commit is contained in:
acazares
2026-02-16 18:10:12 +00:00
parent 9e37340dd5 89ac720324
commit 56a781f2be
15 changed files with 611 additions and 644 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
backend/api/v1/modules/a76/general_catalogs/fractions/tariff_fractions/dto.py
View File

@@ -48,6 +48,9 @@ class TariffFractionResponseDTO(BaseModel):
umt: Optional[str] = None
adv_impo: Optional[str] = None
adv_expo: Optional[str] = None
dof: Optional[str] = None
aplica_ieps: Optional[str] = None
um_code: Optional[str] = None
model_config = ConfigDict(from_attributes=True)

60
backend/api/v1/modules/a76/general_catalogs/fractions/tariff_fractions/routes.py
View File

@@ -38,7 +38,11 @@ async def list_tariff_fractions(
if search:
filters["search"] = search
items, total = TariffFractionService.get_all(
# Updated to async call with Sitar integration
# WARNING: Using async def with blocking DB dependency (Session) run in threadpool by FastAPI.
# Service.get_all calls Sitar (async) or DB (sync).
# This should be fine.
items, total = await TariffFractionService.get_all(
db, skip, page_size, filters
)
@@ -55,7 +59,7 @@ async def list_tariff_fractions(
"/{tariff_fraction_id}",
response_model=TariffFractionResponseDTO,
summary="Get Tariff Fraction by ID",
description="Get a specific tariff fraction by ID",
description="Get a specific tariff fraction by ID (Lookups in Local DB for legacy compatibility)",
)
async def get_tariff_fraction(
tariff_fraction_id: int,
@@ -68,55 +72,3 @@ async def get_tariff_fraction(
raise HTTPException(status_code=404, detail="Tariff fraction not found")
return TariffFractionResponseDTO.model_validate(item)
@router.post(
"/",
response_model=TariffFractionResponseDTO,
summary="Create Tariff Fraction",
description="Create a new tariff fraction (admin only)",
status_code=201,
)
async def create_tariff_fraction(
data: TariffFractionCreateDTO,
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
item = TariffFractionService.create(db, data)
return TariffFractionResponseDTO.model_validate(item)
@router.put(
"/{tariff_fraction_id}",
response_model=TariffFractionResponseDTO,
summary="Update Tariff Fraction",
description="Update an existing tariff fraction (admin only)",
)
async def update_tariff_fraction(
tariff_fraction_id: int,
data: TariffFractionUpdateDTO,
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
item = TariffFractionService.update(db, tariff_fraction_id, data)
if not item:
from fastapi import HTTPException
raise HTTPException(status_code=404, detail="Tariff fraction not found")
return TariffFractionResponseDTO.model_validate(item)
@router.delete(
"/{tariff_fraction_id}",
summary="Delete Tariff Fraction",
description="Delete a tariff fraction (admin only)",
status_code=204,
)
async def delete_tariff_fraction(
tariff_fraction_id: int,
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
success = TariffFractionService.delete(db, tariff_fraction_id)
if not success:
from fastapi import HTTPException
raise HTTPException(status_code=404, detail="Tariff fraction not found")

160
backend/api/v1/modules/a76/general_catalogs/fractions/tariff_fractions/service.py
View File

@@ -7,26 +7,165 @@ from typing import List, Optional, Tuple, Dict, Any
from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError
from fastapi import HTTPException
import zlib
import logging
from .models import TariffFraction
from .dto import TariffFractionCreateDTO, TariffFractionUpdateDTO
from api.v1.modules.sitar.fracciones.service import FraccionesService
from api.v1.modules.sitar.fracciones.schemas import FraccionesResponse
logger = logging.getLogger(__name__)
class TariffFractionMapper:
"""Helper to map Sitar responses to Local domain objects"""
@staticmethod
def to_domain(fraccion: FraccionesResponse) -> TariffFraction:
# Generate ID: Use SYSID if available, else composite hash of code + nico
if fraccion.SYSID:
fake_id = fraccion.SYSID
else:
# Composite key for uniqueness if SYSID missing
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
description_val = fraccion.DESCRIPCION if fraccion.DESCRIPCION else "(Sin descripción)"
tf = TariffFraction(
id=fake_id,
code=code_val,
fraction=fraction_val,
description=description_val,
nico=fraccion.NICO,
# MAP CHANGE: UMT now maps to Abbreviation (e.g., Pza, Kg)
umt=fraccion.UMABREVIACION,
adv_impo=fraccion.ADVIMPOTXT,
adv_expo=fraccion.ADVEXPOTXT
)
# Dynamically attach non-model attributes for DTO
tf.dof = fraccion.DOF
tf.aplica_ieps = fraccion.APLICAIEPS
# MAP CHANGE: New field for the numeric code (e.g., 01, 06)
tf.um_code = fraccion.UMCLAVE
return tf
class TariffFractionService:
"""Service para gestionar fracciones arancelarias (catálogo global)"""
@staticmethod
def get_all(
async def get_all(
db: Session,
skip: int = 0,
limit: int = 100,
filters: Optional[Dict[str, Any]] = None,
) -> Tuple[List[TariffFraction], int]:
"""Obtiene todas las fracciones arancelarias con filtros opcionales"""
"""
Obtiene fracciones arancelarias.
Estrategia: Sitar API -> Fallback Local DB
"""
# 1. Try Sitar API
try:
sitar_service = FraccionesService.get_instance()
# Map filters
sitar_fraccion = None
sitar_nico = None
has_filters = False
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"
clean_term = term.replace(".", "")
if clean_term and clean_term[0].isdigit():
sitar_fraccion = clean_term
has_filters = True
else:
# Attempt description search via API first
logger.info(f"Search term '{term}' identified as text. Attempting API description search.")
pass
if filters.get("code"):
sitar_fraccion = filters["code"]
has_filters = True
if filters.get("fraction"):
sitar_fraccion = filters["fraction"]
has_filters = True
if filters.get("nico"):
sitar_nico = filters["nico"]
has_filters = True
# 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"]
has_filters = True
# Note: Sitar search might not return total count.
# We fetch page items. Pagination might be tricky if Sitar doesn't return total.
# Assuming Sitar returns a list.
sitar_items = await sitar_service.search(
fraccion=sitar_fraccion,
nico=sitar_nico,
description=sitar_description,
nivel=5, # User requested filtering by level 5
skip=skip,
limit=limit
)
# STRICT API USAGE:
# We do NOT fallback to local DB on empty list, as user requested strict API consumption.
# We also do NOT attempt enrichment as codes mismatch (API uses '010191A' vs Local '01012101').
# Map items
items = [TariffFractionMapper.to_domain(item) for item in sitar_items]
# Estimate total (Sitar service doesn't return total currently)
# If we got full limit, assume there are more.
total = len(items) + skip
if len(items) == limit:
total += 1 # Indicate more pages
return items, total
except Exception as e:
logger.error(f"Error fetching from Sitar API: {e}")
# STRICT API USAGE: Propagate error, do NOT fallback to local DB.
raise e
@staticmethod
async def _get_all_local_async(
db: Session,
skip: int = 0,
limit: int = 100,
filters: Optional[Dict[str, Any]] = None,
) -> Tuple[List[TariffFraction], int]:
"""Lógica original de consulta local"""
query = db.query(TariffFraction)
# Aplicar filtros
@@ -64,21 +203,24 @@ class TariffFractionService:
db: Session,
tariff_fraction_id: int,
) -> Optional[TariffFraction]:
"""Obtiene una fracción arancelaria por ID"""
"""
Obtiene por ID.
Como Sitar no usa estos IDs, consultamos Local DB directamente para compatibilidad legacy.
Si se necesitara obtener detalle de Sitar, se requeriría otro identificador (Code).
"""
return (
db.query(TariffFraction)
.filter(TariffFraction.id == tariff_fraction_id)
.first()
)
# WRITE OPERATIONS - DEPRECATED / LOCAL ONLY (Optional: Remove or Keep for Fallback Maintenance)
@staticmethod
def get_by_code(
db: Session,
code: str,
) -> Optional[TariffFraction]:
"""Obtiene una fracción arancelaria por código"""
return (
db.query(TariffFraction)
.filter(TariffFraction.code == code)
@@ -90,8 +232,6 @@ class TariffFractionService:
db: Session,
tariff_fraction_data: TariffFractionCreateDTO,
) -> TariffFraction:
"""Crea una nueva fracción arancelaria"""
try:
tariff_fraction = TariffFraction(
**tariff_fraction_data.model_dump(),
@@ -114,8 +254,6 @@ class TariffFractionService:
tariff_fraction_id: int,
tariff_fraction_data: TariffFractionUpdateDTO,
) -> Optional[TariffFraction]:
"""Actualiza una fracción arancelaria existente"""
tariff_fraction = TariffFractionService.get_by_id(
db, tariff_fraction_id
)
@@ -144,8 +282,6 @@ class TariffFractionService:
db: Session,
tariff_fraction_id: int,
) -> bool:
"""Elimina una fracción arancelaria"""
tariff_fraction = TariffFractionService.get_by_id(
db, tariff_fraction_id
)

42
backend/api/v1/modules/a76/general_catalogs/fractions/us_tariff_fractions/routes.py
View File

@@ -17,21 +17,8 @@ from .dto import (
)
from .service import USTariffFractionService
# Create base router with generic CRUD routes (disabled list because we'll create a custom one)
base_router = TenantCRUDRoutes(
service=USTariffFractionService,
create_schema=USTariffFractionCreateDTO,
update_schema=USTariffFractionUpdateDTO,
response_schema=USTariffFractionResponseDTO,
prefix="/us-tariff-fractions",
tags=["a76 / general catalogs / us tariff fractions"],
resource_name="USTariffFraction",
id_name="us_tariff_fraction_id",
enable_list=False, # Disable default list, we'll add custom one
enable_filters=False,
default_page_size=50,
max_page_size=10000,
)
# Create base router with generic CRUD routes - REMOVED strictly read-only from Sitar
# Writes are disabled at API level, but Service still supports fallback writes if needed internally
router = APIRouter(prefix="/us-tariff-fractions", tags=["a76 / general catalogs / us tariff fractions"])
@@ -57,7 +44,8 @@ async def list_us_tariff_fractions(
if search:
filters["search"] = search
items, total = USTariffFractionService.get_all(
# Updated to async call with Sitar integration
items, total = await USTariffFractionService.get_all(
db, tenant_id, company_id, skip, page_size, filters
)
@@ -69,5 +57,23 @@ async def list_us_tariff_fractions(
"pages": (total + page_size - 1) // page_size,
}
# Include other CRUD routes from base router
router.include_router(base_router.router)
@router.get(
"/{us_tariff_fraction_id}",
response_model=USTariffFractionResponseDTO,
summary="Get US Tariff Fraction by ID",
description="Get a specific US tariff fraction by ID (Lookups in Local DB for legacy compatibility)",
)
async def get_us_tariff_fraction(
us_tariff_fraction_id: int,
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
tenant_id = validate_access_to_resource(db, company_id, current_user)
item = USTariffFractionService.get_by_id(db, tenant_id, company_id, us_tariff_fraction_id)
if not item:
from fastapi import HTTPException
raise HTTPException(status_code=404, detail="US Tariff fraction not found")
return USTariffFractionResponseDTO.model_validate(item)

115
backend/api/v1/modules/a76/general_catalogs/fractions/us_tariff_fractions/service.py
View File

@@ -6,19 +6,68 @@ from typing import List, Optional, Tuple, Dict, Any
from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError
from fastapi import HTTPException
import zlib
import logging
import re
from decimal import Decimal
from .models import USTariffFraction
from .dto import USTariffFractionCreateDTO, USTariffFractionUpdateDTO
from api.v1.modules.sitar.fracciones_usa.service import FraccionesUSAService
from api.v1.modules.sitar.fracciones_usa.schemas import FraccionesUSAResponse
logger = logging.getLogger(__name__)
class USTariffFractionMapper:
"""Helper to map Sitar USA responses to Local domain objects"""
@staticmethod
def to_domain(fraccion: FraccionesUSAResponse, tenant_id: int, company_id: int) -> USTariffFraction:
# Generate a deterministic numeric ID based on the unique code
# We use CRC32 to get a consistent integer implementation-independent
fake_id = zlib.crc32((fraccion.FRACCION_SIN_PUNTO or "").encode('utf-8'))
# Parse numeric values safely
ad_valorem = None
if fraccion.TARIFA1:
try:
# Extract numbers from string like "5.2%" or similar if present
# Assuming TARIFA1 might be clean number or percentage string
clean_val = re.sub(r'[^\d.]', '', str(fraccion.TARIFA1))
if clean_val:
ad_valorem = Decimal(clean_val)
except:
pass
fixed_cost = None
if fraccion.ESPECIFICO:
try:
clean_val = re.sub(r'[^\d.]', '', str(fraccion.ESPECIFICO))
if clean_val:
fixed_cost = Decimal(clean_val)
except:
pass
return USTariffFraction(
id=fake_id, # Updated to use fake_id instead of sitar consecutive if needed, or consistent hash
tenant_id=tenant_id,
company_id=company_id,
code=fraccion.FRACCION_SIN_PUNTO or "",
prefix=None, # Not mapped from Sitar response currently
type_code=None,
ad_valorem=ad_valorem,
fixed_cost=fixed_cost,
unit_of_measure=fraccion.UNIDADCANTIDAD,
description=fraccion.DESCRIPCION
)
class USTariffFractionService:
"""Service para gestionar fracciones arancelarias americanas"""
@staticmethod
def get_all(
async def get_all(
db: Session,
tenant_id: int,
company_id: int,
@@ -26,8 +75,60 @@ class USTariffFractionService:
limit: int = 100,
filters: Optional[Dict[str, Any]] = None,
) -> Tuple[List[USTariffFraction], int]:
"""Obtiene todas las fracciones arancelarias americanas con filtros opcionales"""
"""
Obtiene todas las fracciones arancelarias americanas con filtros opcionales.
Estrategia: Sitar API -> Fallback Local DB
"""
# 1. Try Sitar API
try:
sitar_service = FraccionesUSAService.get_instance()
sitar_fraccion = None
has_filters = False
if filters and filters.get("search"):
term = filters["search"]
# Sitar only filters by fraction code
if term.replace(".", "").isdigit():
sitar_fraccion = term
has_filters = True
sitar_items = await sitar_service.search(
fraccion=sitar_fraccion,
skip=skip,
limit=limit
)
# If Sitar returns empty list AND we didn't have specific filters, attempt fallback
if not sitar_items and not has_filters:
logger.warning("Sitar return empty list for USA broad query. Attempting fallback to local DB.")
return USTariffFractionService._get_all_local(db, tenant_id, company_id, skip, limit, filters)
# Map items
items = [USTariffFractionMapper.to_domain(item, tenant_id, company_id) for item in sitar_items]
# Estimate total
total = len(items) + skip
if len(items) == limit:
total += 1
return items, total
except Exception as e:
logger.error(f"Error fetching USA Fractions from Sitar API, falling back to local DB: {e}")
return USTariffFractionService._get_all_local(db, tenant_id, company_id, skip, limit, filters)
@staticmethod
def _get_all_local(
db: Session,
tenant_id: int,
company_id: int,
skip: int = 0,
limit: int = 100,
filters: Optional[Dict[str, Any]] = None,
) -> Tuple[List[USTariffFraction], int]:
"""Lógica original de consulta local"""
query = db.query(USTariffFraction).filter(
USTariffFraction.tenant_id == tenant_id,
USTariffFraction.company_id == company_id,
@@ -53,7 +154,10 @@ class USTariffFractionService:
def get_by_id(
db: Session, tenant_id: int, company_id: int, fraction_id: int
) -> Optional[USTariffFraction]:
"""Obtiene una fracción arancelaria americana por ID"""
"""
Obtiene por ID.
Legacy: Consulta Local DB.
"""
return (
db.query(USTariffFraction)
.filter(
@@ -64,6 +168,8 @@ class USTariffFractionService:
.first()
)
# WRITE OPERATIONS - DEPRECATED / LOCAL ONLY
@staticmethod
def create(
db: Session,
@@ -71,7 +177,6 @@ class USTariffFractionService:
company_id: int,
fraction_data: USTariffFractionCreateDTO,
) -> USTariffFraction:
"""Crea una nueva fracción arancelaria americana"""
try:
db_fraction = USTariffFraction(
tenant_id=tenant_id,
@@ -98,7 +203,6 @@ class USTariffFractionService:
fraction_id: int,
fraction_data: USTariffFractionUpdateDTO,
) -> Optional[USTariffFraction]:
"""Actualiza una fracción arancelaria americana existente"""
db_fraction = USTariffFractionService.get_by_id(
db, tenant_id, company_id, fraction_id
)
@@ -117,7 +221,6 @@ class USTariffFractionService:
def delete(
db: Session, tenant_id: int, company_id: int, fraction_id: int
) -> bool:
"""Elimina una fracción arancelaria americana"""
db_fraction = USTariffFractionService.get_by_id(
db, tenant_id, company_id, fraction_id
)

18
backend/api/v1/modules/sitar/common/base_service.py
View File

@@ -22,7 +22,7 @@ class SitarAPIBaseService:
self.base_url = os.getenv("SITAR_API_URL")
self.username = os.getenv("SITAR_API_USER")
self.password = os.getenv("SITAR_API_PASSWORD")
self.timeout = 10.0
self.timeout = 30.0
if not all([self.base_url, self.username, self.password]):
raise ValueError(
@@ -101,4 +101,20 @@ class SitarAPIBaseService:
headers=headers,
)
response.raise_for_status()
# DEBUG LOGGING for SITAR inspection
if "fracciones" in url:
import logging
logger = logging.getLogger(__name__)
logger.info(f"SITAR API Response Headers for {url}: {dict(response.headers)}")
try:
data = response.json()
if isinstance(data, dict):
logger.info(f"SITAR API Response Body Keys: {list(data.keys())}")
elif isinstance(data, list) and len(data) > 0:
logger.info(f"SITAR API Response List Item Keys: {list(data[0].keys())}")
return data
except Exception:
pass
return response.json()

1
backend/api/v1/modules/sitar/fracciones/schemas.py
View File

@@ -33,6 +33,7 @@ class FraccionesResponse(BaseModel):
APLICAIEPS: Optional[str] = Field(None, max_length=1)
NIVEL: Optional[int] = None
NICO: Optional[str] = Field(None, max_length=14)
SYSID: Optional[int] = None
class Config:
from_attributes = True

6
backend/api/v1/modules/sitar/fracciones/service.py
View File

@@ -21,6 +21,8 @@ class FraccionesService(SitarAPIBaseService):
self,
fraccion: Optional[str] = None,
nico: Optional[str] = None,
description: Optional[str] = None,
nivel: Optional[int] = None,
skip: int = 0,
limit: int = 100,
) -> List[FraccionesResponse]:
@@ -30,6 +32,10 @@ class FraccionesService(SitarAPIBaseService):
params["fraccion"] = fraccion
if nico:
params["nico"] = nico
if description:
params["descripcion"] = description
if nivel is not None:
params["nivel"] = nivel
data = await self._make_request("GET", "/api/v1/fracciones/", params=params)
return [FraccionesResponse(**item) for item in data]