Funciones y logia de exportacion, como mejoras CRUD en las partes de exportacion
This commit is contained in:
@@ -1,4 +1,5 @@
|
||||
from typing import List, Optional
|
||||
from datetime import datetime
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||||
from sqlalchemy.orm import Session
|
||||
from core.database import get_core_db
|
||||
@@ -34,6 +35,46 @@ def get_historical_fractions(
|
||||
"pages": (total + page_size - 1) // page_size if page_size > 0 else 1
|
||||
}
|
||||
|
||||
@router.get("/rate")
|
||||
async def get_rate(
|
||||
company_id: int = Query(..., description="Company ID"),
|
||||
historical_fraction: str = Query(..., description="8-character fraction code"),
|
||||
nico: str = Query(..., description="2-character NICO code"),
|
||||
direction: str = Query("export", description="Movement direction: 'import' or 'export'"),
|
||||
tariff_type: str = Query("GENERAL", description="Tariff regimen: 'GENERAL', 'PROSEC', etc."),
|
||||
invoice_date: str = Query(..., description="ISO Date (YYYY-MM-DD)"),
|
||||
is_regime_change: bool = Query(False, description="Whether the invoice is a regime change (Cambio de Régimen)"),
|
||||
db: Session = Depends(get_core_db),
|
||||
current_user = Depends(get_current_user)
|
||||
):
|
||||
"""
|
||||
Get the historical tariff rate for a specific fraction, nico, and date.
|
||||
"""
|
||||
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
||||
|
||||
# Parse date
|
||||
try:
|
||||
parsed_date = datetime.fromisoformat(invoice_date.split('T')[0])
|
||||
except ValueError:
|
||||
raise HTTPException(status_code=400, detail="Invalid date format. Use YYYY-MM-DD")
|
||||
|
||||
service = HistoricalTariffFractionService(db)
|
||||
rate = await service.get_historical_rate(
|
||||
tenant_id=int(tenant_id),
|
||||
company_id=company_id,
|
||||
historical_fraction=historical_fraction,
|
||||
nico=nico,
|
||||
direction=direction,
|
||||
tariff_type=tariff_type,
|
||||
invoice_date=parsed_date,
|
||||
is_regime_change=is_regime_change
|
||||
)
|
||||
|
||||
if rate is None:
|
||||
return {"found": False, "rate": 0}
|
||||
|
||||
return {"found": True, "rate": float(rate)}
|
||||
|
||||
@router.get("/{id}", response_model=HistoricalTariffFractionResponse)
|
||||
def get_historical_fraction(
|
||||
id: int,
|
||||
|
||||
@@ -1,9 +1,11 @@
|
||||
|
||||
from datetime import datetime
|
||||
from decimal import Decimal
|
||||
from typing import Optional, List, Tuple
|
||||
from sqlalchemy import select, or_, func
|
||||
from sqlalchemy.orm import Session
|
||||
from .models import HistoricalTariffFraction
|
||||
from .schemas import HistoricalTariffFractionCreate, HistoricalTariffFractionUpdate
|
||||
from .schemas import HistoricalTariffFractionResponse, HistoricalTariffFractionCreate, HistoricalTariffFractionUpdate
|
||||
from api.v1.modules.sitar.fracciones.service import FraccionesService
|
||||
|
||||
class HistoricalTariffFractionService:
|
||||
def __init__(self, db: Session):
|
||||
@@ -67,3 +69,109 @@ class HistoricalTariffFractionService:
|
||||
self.db.delete(obj)
|
||||
self.db.commit()
|
||||
return obj
|
||||
async def get_historical_rate(
|
||||
self,
|
||||
tenant_id: int,
|
||||
company_id: int,
|
||||
historical_fraction: str,
|
||||
nico: str,
|
||||
direction: str,
|
||||
tariff_type: str,
|
||||
invoice_date: datetime,
|
||||
is_regime_change: bool = False
|
||||
) -> Optional[Decimal]:
|
||||
"""
|
||||
Gets the historical tax rate based on the fraction, nico, and date.
|
||||
Equivalent to Clarion BUSCA_FRACCION_HISTORICA.
|
||||
"""
|
||||
# Normalize input fraction to handle both 8-digit and unpadded (7-digit) versions
|
||||
fraction_variants = [historical_fraction]
|
||||
unpadded = historical_fraction.lstrip('0')
|
||||
if unpadded and unpadded != historical_fraction:
|
||||
fraction_variants.append(unpadded)
|
||||
|
||||
query = select(HistoricalTariffFraction).where(
|
||||
HistoricalTariffFraction.tenant_id == tenant_id,
|
||||
HistoricalTariffFraction.company_id == company_id,
|
||||
HistoricalTariffFraction.historical_fraction.in_(fraction_variants),
|
||||
HistoricalTariffFraction.fraction_type.ilike(tariff_type), # Match GENERAL, PROSEC, etc.
|
||||
or_(
|
||||
HistoricalTariffFraction.nico == nico,
|
||||
HistoricalTariffFraction.nico.is_(None),
|
||||
HistoricalTariffFraction.nico == ''
|
||||
),
|
||||
HistoricalTariffFraction.publication_date <= invoice_date
|
||||
).order_by(HistoricalTariffFraction.publication_date.desc())
|
||||
|
||||
result = self.db.execute(query).scalars().first()
|
||||
|
||||
if result:
|
||||
# Special rule: If it's a regime change, always return the import rate (TasaImNum)
|
||||
# as per Clarion logic ASIGNA_FRACCION_HISTORICA
|
||||
if is_regime_change:
|
||||
return result.import_tax_rate
|
||||
|
||||
# Otherwise return based on direction
|
||||
if direction.lower() == 'import':
|
||||
return result.import_tax_rate
|
||||
else:
|
||||
return result.export_tax_rate
|
||||
|
||||
# 2. Priority 2: Try SITAR API (Modern source of truth)
|
||||
try:
|
||||
sitar_service = FraccionesService.get_instance()
|
||||
# Search by 8-digit fraction and 2-digit NICO
|
||||
sitar_data = await sitar_service.search(
|
||||
fraccion=historical_fraction,
|
||||
nico=nico,
|
||||
limit=1
|
||||
)
|
||||
|
||||
if sitar_data:
|
||||
first_record = sitar_data[0]
|
||||
# Special rule: If it's a regime change, always return the import rate (TasaImNum)
|
||||
if is_regime_change:
|
||||
return first_record.ADVIMPONUM
|
||||
|
||||
# Otherwise return based on direction
|
||||
if direction.lower() == 'import':
|
||||
return first_record.ADVIMPONUM
|
||||
else:
|
||||
return first_record.ADVEXPONUM
|
||||
except Exception as e:
|
||||
# Log error but continue to fallback
|
||||
print(f"Error fetching data from SITAR API: {e}")
|
||||
|
||||
# 3. Priority 3: Fallback to main TariffFraction catalog if not found in historical or SITAR
|
||||
from ..tariff_fractions.models import TariffFraction
|
||||
|
||||
# In the main catalog, fractions might be stored with dots (e.g. 0101.90.99)
|
||||
# or as code (e.g. 01019099)
|
||||
# We search primarily by fraction code (8 digits) and take the first one found.
|
||||
# This handles cases where NICO doesn't match perfectly.
|
||||
fallback_query = select(TariffFraction).where(
|
||||
or_(
|
||||
TariffFraction.code == historical_fraction,
|
||||
TariffFraction.fraction == f"{historical_fraction[:4]}.{historical_fraction[4:6]}.{historical_fraction[6:8]}"
|
||||
)
|
||||
).order_by(TariffFraction.nico) # Order so we get a consistent result if multiple NICOs exist
|
||||
|
||||
main_result = self.db.execute(fallback_query).scalars().first()
|
||||
|
||||
if main_result:
|
||||
# Apply same regime change rule to fallback if found
|
||||
if is_regime_change:
|
||||
rate_str = main_result.adv_impo
|
||||
else:
|
||||
rate_str = main_result.adv_impo if direction.lower() == 'import' else main_result.adv_expo
|
||||
|
||||
if rate_str:
|
||||
try:
|
||||
# Remove non-numeric characters (like % or text)
|
||||
import re
|
||||
clean_rate = re.sub(r'[^\d.]', '', rate_str)
|
||||
return Decimal(clean_rate) if clean_rate else Decimal(0)
|
||||
except:
|
||||
return Decimal(0)
|
||||
|
||||
return None
|
||||
|
||||
Reference in New Issue
Block a user