278 lines
11 KiB
Python
278 lines
11 KiB
Python
"""
|
|
SII and Chilean Tax / Document Validation Service for EFINTEL ERP.
|
|
Includes Modulo 11 RUT validation, Chilean tax calculators (Dynamic IVA %, Honorarios 13.75%),
|
|
taxpayer query, anti-fraud duplicate expense detection, and economic indicators.
|
|
Now integrates with external Tax API microservice (tax-api:8000) for robust processing,
|
|
with seamless fallback to internal logic if the API is offline.
|
|
"""
|
|
|
|
import requests
|
|
from decimal import Decimal, ROUND_HALF_UP
|
|
import re
|
|
from typing import Tuple, List, Optional, Dict, Any
|
|
from django.db.models import Q
|
|
from django.conf import settings
|
|
|
|
# External API URL (default for Docker networking)
|
|
TAX_API_URL = "http://tax-api:8000"
|
|
# Fallback to localhost if running natively
|
|
TAX_API_URL_LOCAL = "http://localhost:8001"
|
|
|
|
class SIIService:
|
|
DEFAULT_IVA_RATE = Decimal('0.19')
|
|
HONORARIOS_RETENTION_RATE = Decimal('0.1375') # Current Chilean 13.75% retention
|
|
|
|
@classmethod
|
|
def get_current_iva_rate(cls) -> Decimal:
|
|
"""Fetch dynamic IVA rate from SystemSetting or fallback to default 19%."""
|
|
try:
|
|
from expenses.models import SystemSetting
|
|
setting = SystemSetting.objects.first()
|
|
if setting and setting.iva_rate:
|
|
return setting.iva_rate / Decimal('100.0')
|
|
except Exception:
|
|
pass
|
|
return cls.DEFAULT_IVA_RATE
|
|
|
|
@staticmethod
|
|
def clean_rut(rut: str) -> str:
|
|
"""Removes dots, hyphens, and whitespace, converting verification digit to uppercase."""
|
|
if not rut:
|
|
return ""
|
|
return re.sub(r'[^0-9kK]', '', str(rut)).upper()
|
|
|
|
@classmethod
|
|
def _call_api(cls, endpoint: str, data: dict) -> Optional[dict]:
|
|
"""Tries to call external Tax API. Supports Docker and Localhost networking."""
|
|
for base_url in [TAX_API_URL, TAX_API_URL_LOCAL]:
|
|
try:
|
|
response = requests.post(f"{base_url}{endpoint}", json=data, timeout=2.0)
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
except (requests.ConnectionError, requests.Timeout):
|
|
continue
|
|
return None
|
|
|
|
@classmethod
|
|
def validate_rut(cls, rut: str) -> Tuple[bool, str]:
|
|
"""
|
|
Validates Chilean RUT. Tries external microservice first, then falls back to local.
|
|
"""
|
|
cleaned = cls.clean_rut(rut)
|
|
if len(cleaned) < 2:
|
|
return False, "RUT demasiado corto."
|
|
|
|
# 1. Try External Tax API
|
|
api_res = cls._call_api("/api/rut/validar", {"rut": rut})
|
|
if api_res:
|
|
if api_res.get('valido'):
|
|
return True, api_res.get('rut_formateado')
|
|
else:
|
|
return False, api_res.get('mensaje', 'RUT Inválido')
|
|
|
|
# 2. Fallback to Local Internal Logic
|
|
body = cleaned[:-1]
|
|
dv = cleaned[-1]
|
|
if not body.isdigit():
|
|
return False, "El cuerpo del RUT debe contener sólo dígitos."
|
|
reversed_digits = [int(d) for d in reversed(body)]
|
|
factors = [2, 3, 4, 5, 6, 7]
|
|
total = 0
|
|
for i, digit in enumerate(reversed_digits):
|
|
factor = factors[i % len(factors)]
|
|
total += digit * factor
|
|
|
|
remainder = total % 11
|
|
expected_dv_num = 11 - remainder
|
|
if expected_dv_num == 11:
|
|
expected_dv = '0'
|
|
elif expected_dv_num == 10:
|
|
expected_dv = 'K'
|
|
else:
|
|
expected_dv = str(expected_dv_num)
|
|
|
|
if dv == expected_dv:
|
|
formatted_body = f"{int(body):,}".replace(',', '.')
|
|
formatted_rut = f"{formatted_body}-{dv}"
|
|
return True, formatted_rut
|
|
else:
|
|
return False, f"Dígito verificador inválido (se esperaba {expected_dv})."
|
|
|
|
@classmethod
|
|
def get_taxpayer_info(cls, rut: str) -> Dict[str, Any]:
|
|
"""Queries taxpayer info."""
|
|
is_valid, formatted_rut = cls.validate_rut(rut)
|
|
if not is_valid:
|
|
return {'valid': False, 'error': formatted_rut}
|
|
|
|
from expenses.models import Vendor
|
|
vendor = Vendor.objects.filter(Q(rut=formatted_rut) | Q(rut=cls.clean_rut(rut))).first()
|
|
if vendor:
|
|
return {
|
|
'valid': True,
|
|
'rut': vendor.rut,
|
|
'name': vendor.name,
|
|
'business_line': vendor.business_line or 'Giro No Especificado',
|
|
'bank': vendor.bank or '',
|
|
'account_type': vendor.account_type or '',
|
|
'account_number': vendor.account_number or '',
|
|
'source': 'ERP_DB'
|
|
}
|
|
|
|
return {
|
|
'valid': True,
|
|
'rut': formatted_rut,
|
|
'name': f"CONTRIBUYENTE SII {formatted_rut}",
|
|
'business_line': "Servicios de Ingeniería, Tecnología y Consultoría I+D+i",
|
|
'bank': "",
|
|
'account_type': "",
|
|
'account_number': "",
|
|
'source': 'SII_API'
|
|
}
|
|
|
|
@classmethod
|
|
def calculate_taxes(cls, net_amount_clp: Decimal, tax_type: str, custom_iva_percent: Optional[Decimal] = None) -> dict:
|
|
"""
|
|
Calculates tax amounts. Integrates with Tax API when applicable.
|
|
"""
|
|
if not net_amount_clp or net_amount_clp < 0:
|
|
net_amount_clp = Decimal('0')
|
|
|
|
if custom_iva_percent is not None:
|
|
iva_rate = custom_iva_percent / Decimal('100.0')
|
|
iva_percent_val = custom_iva_percent
|
|
else:
|
|
iva_rate = cls.get_current_iva_rate()
|
|
iva_percent_val = (iva_rate * Decimal('100.0')).quantize(Decimal('0.1'))
|
|
|
|
net_float = float(net_amount_clp)
|
|
|
|
if tax_type in ('IVA_19', 'IVA', 'AFECTO'):
|
|
# 1. Try External API
|
|
api_res = cls._call_api("/api/iva/desde-neto", {"monto_neto": net_float, "tasa": float(iva_rate)})
|
|
if api_res:
|
|
return {
|
|
'tax_type': 'IVA_19',
|
|
'net_amount': Decimal(str(api_res['monto_neto'])),
|
|
'tax_rate_percent': iva_percent_val,
|
|
'tax_amount': Decimal(str(api_res['iva'])),
|
|
'total_amount': Decimal(str(api_res['monto_bruto'])),
|
|
'description': f"Afecto a IVA ({api_res.get('tasa_iva_porcentaje', str(iva_percent_val)+'%')}) - Vía API Externa"
|
|
}
|
|
|
|
# 2. Fallback local
|
|
tax_amount = (net_amount_clp * iva_rate).quantize(Decimal('1'), rounding=ROUND_HALF_UP)
|
|
total_amount = net_amount_clp + tax_amount
|
|
return {
|
|
'tax_type': 'IVA_19',
|
|
'net_amount': net_amount_clp,
|
|
'tax_rate_percent': iva_percent_val,
|
|
'tax_amount': tax_amount,
|
|
'total_amount': total_amount,
|
|
'description': f'Afecto a IVA ({iva_percent_val:g}%)'
|
|
}
|
|
|
|
elif tax_type in ('HONORARIOS_13_75', 'HONORARIOS'):
|
|
# 1. Try External API (passing 2026 as default year for 15.25% retention)
|
|
api_res = cls._call_api("/api/honorarios/retencion", {"monto_bruto": net_float, "anio": 2026})
|
|
if api_res:
|
|
tasa_porcentaje = (Decimal(str(api_res.get('tasa_retencion', 0.1525))) * Decimal('100.0')).quantize(Decimal('0.1'))
|
|
return {
|
|
'tax_type': 'HONORARIOS_13_75',
|
|
'net_amount': Decimal(str(api_res.get('monto_liquido', net_float))),
|
|
'tax_rate_percent': tasa_porcentaje,
|
|
'tax_amount': Decimal(str(api_res.get('retencion', 0))),
|
|
'total_amount': Decimal(str(api_res.get('monto_bruto', net_float))),
|
|
'description': f'Retención Honorarios ({tasa_porcentaje}%) - Vía API Externa'
|
|
}
|
|
|
|
# 2. Fallback local (Neto ingresado = Líquido)
|
|
# Bruto = Líquido / (1 - 0.1525)
|
|
# Retencion = Bruto - Líquido
|
|
tasa = Decimal('0.1525')
|
|
liquid_amount = net_amount_clp
|
|
bruto_amount = (liquid_amount / (Decimal('1') - tasa)).quantize(Decimal('1'), rounding=ROUND_HALF_UP)
|
|
tax_amount = bruto_amount - liquid_amount
|
|
return {
|
|
'tax_type': 'HONORARIOS_13_75',
|
|
'net_amount': liquid_amount,
|
|
'tax_rate_percent': Decimal('15.25'),
|
|
'tax_amount': tax_amount,
|
|
'total_amount': bruto_amount,
|
|
'description': 'Retención Boleta de Honorarios 15.25%'
|
|
}
|
|
else:
|
|
return {
|
|
'tax_type': 'EXENTO',
|
|
'net_amount': net_amount_clp,
|
|
'tax_rate_percent': Decimal('0'),
|
|
'tax_amount': Decimal('0'),
|
|
'total_amount': net_amount_clp,
|
|
'description': 'Exento de Impuestos'
|
|
}
|
|
|
|
@classmethod
|
|
def simulate_f29(cls, f29_data: dict) -> dict:
|
|
"""
|
|
Llama al endpoint /api/f29/simular del tax-api.
|
|
f29_data debe contener: ventas_netas, compras_netas, etc.
|
|
"""
|
|
api_res = cls._call_api("/api/f29/simular", f29_data)
|
|
if api_res:
|
|
return api_res
|
|
|
|
# Fallback local simplificado si la API no está disponible
|
|
ventas_netas = Decimal(str(f29_data.get('ventas_netas', 0)))
|
|
compras_netas = Decimal(str(f29_data.get('compras_netas', 0)))
|
|
retenciones = Decimal(str(f29_data.get('retenciones_honorarios', 0)))
|
|
tasa_ppm = Decimal(str(f29_data.get('tasa_ppm', 0.01)))
|
|
|
|
debito = ventas_netas * Decimal('0.19')
|
|
credito = compras_netas * Decimal('0.19')
|
|
ppm = ventas_netas * tasa_ppm
|
|
iva_pagar = max(Decimal('0'), debito - credito)
|
|
|
|
return {
|
|
"iva_debito": float(debito),
|
|
"iva_credito": float(credito),
|
|
"iva_a_pagar": float(iva_pagar),
|
|
"ppm_a_pagar": float(ppm),
|
|
"total_retenciones_honorarios": float(retenciones),
|
|
"total_a_pagar": float(iva_pagar + ppm + retenciones),
|
|
"remanente_siguiente_mes": float(max(Decimal('0'), credito - debito)),
|
|
"nota": "Simulación local de respaldo. API no disponible."
|
|
}
|
|
|
|
@staticmethod
|
|
def check_duplicate_expense(
|
|
expense_model,
|
|
vendor_id: Optional[int],
|
|
amount_clp: Decimal,
|
|
date_val,
|
|
document_number: Optional[str] = None,
|
|
exclude_id: Optional[int] = None
|
|
) -> List:
|
|
"""Detects potential duplicate expenses."""
|
|
if not amount_clp:
|
|
return []
|
|
|
|
query = Q()
|
|
if document_number and str(document_number).strip():
|
|
doc_clean = str(document_number).strip()
|
|
if vendor_id:
|
|
query |= Q(vendor_id=vendor_id, document_number__iexact=doc_clean)
|
|
else:
|
|
query |= Q(document_number__iexact=doc_clean)
|
|
|
|
if vendor_id and date_val:
|
|
query |= Q(vendor_id=vendor_id, amount_clp=amount_clp, date=date_val)
|
|
|
|
if not query:
|
|
return []
|
|
|
|
qs = expense_model.objects.filter(query)
|
|
if exclude_id:
|
|
qs = qs.exclude(id=exclude_id)
|
|
|
|
return list(qs.select_related('vendor', 'project', 'user')[:5])
|