122 lines
4.8 KiB
Python
122 lines
4.8 KiB
Python
import logging
|
|
from config import load_config
|
|
import textwrap
|
|
from typing import Dict, Any
|
|
from decimal import Decimal, ROUND_HALF_UP
|
|
from utils import limpiar_cadena, normalizar_telefono_con_plus, corregir_y_validar_email, normalizar_url_para_insert, map_tax_code, cents, money_round, tax_fraction_from_code
|
|
from striprtf.striprtf import rtf_to_text
|
|
|
|
|
|
def rtf_a_texto_plano(rtf: str) -> str:
|
|
"""
|
|
Convierte RTF a texto plano usando striprtf.
|
|
"""
|
|
return rtf_to_text(rtf).strip()
|
|
|
|
|
|
def normalize_customer_fields(fd: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Normaliza campos del registro de origen 'fd' (factura).
|
|
"""
|
|
config = load_config()
|
|
|
|
# >>> aquí usa tus helpers reales:
|
|
def clean_phone(v): return normalizar_telefono_con_plus(v)
|
|
def clean_web(v): return normalizar_url_para_insert(v)
|
|
def clean_tin(v): return limpiar_cadena(v)
|
|
|
|
email1_ok, email1 = corregir_y_validar_email(fd.get("EMAIL_1"))
|
|
email2_ok, email2 = corregir_y_validar_email(fd.get("EMAIL_2"))
|
|
|
|
return {
|
|
"is_company": config["CTE_IS_COMPANY"],
|
|
"tin": clean_tin(str(fd.get("NIF_CIF"))),
|
|
"name": str(fd.get("NOMBRE")),
|
|
"street": str(fd.get("CALLE")),
|
|
"city": str(fd.get("POBLACION")),
|
|
"province": str(fd.get("PROVINCIA")),
|
|
"postal_code": str(fd.get("CODIGO_POSTAL")),
|
|
"country": config['CTE_COUNTRY_CODE'],
|
|
"language_code": config['CTE_LANGUAGE_CODE'],
|
|
"phone_primary": clean_phone(fd.get("TELEFONO_1")),
|
|
"phone_secondary": clean_phone(fd.get("TELEFONO_2")),
|
|
"mobile_primary": clean_phone(fd.get("MOVIL_1")),
|
|
"mobile_secondary": clean_phone(fd.get("MOVIL_2")),
|
|
"email_primary": email1 if email1_ok else None,
|
|
"email_secondary": email2 if email2_ok else None,
|
|
"website": clean_web(str(fd.get("PAGINA_WEB")))
|
|
}
|
|
|
|
|
|
def normalize_header_invoice_fields(fd: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Normaliza campos del registro de origen 'fd' (factura).
|
|
|
|
# campos pendiente de revisar en un futuro
|
|
# xxxxxxx = str(factura_detalle['ID_EMPRESA'])
|
|
# xxxxxxx = str(factura_detalle['OBSERVACIONES'])
|
|
"""
|
|
config = load_config()
|
|
|
|
return {
|
|
"company_id": config['CTE_COMPANY_ID'],
|
|
"is_proforma": config["CTE_IS_PROFORMA"],
|
|
"status": config["CTE_STATUS_INVOICE"],
|
|
"series": config['CTE_SERIE'],
|
|
|
|
"factuges_id": int(fd['ID_FACTURA']),
|
|
"reference": str(fd['REFERENCIA']),
|
|
"invoice_date": str(fd['FECHA_FACTURA']),
|
|
"operation_date": str(fd['FECHA_FACTURA']),
|
|
"description": textwrap.shorten(
|
|
f"{str(fd['REFERENCIA'])} - {str(fd.get('NOMBRE')) or ''}", width=50, placeholder="…"),
|
|
|
|
# siempre tendrán 2 decimales
|
|
'subtotal_amount_value': cents(fd.get('IMPORTE_NETO')),
|
|
'discount_amount_value': cents(fd.get('IMPORTE_DESCUENTO')),
|
|
'discount_percentage_val': int((Decimal(str(fd.get('DESCUENTO') or 0)) * 100).to_integral_value())
|
|
if fd.get('DESCUENTO') is not None else 0,
|
|
'taxable_amount_value': cents(fd.get('BASE_IMPONIBLE')),
|
|
'tax_code': map_tax_code(str(fd.get("DES_TIPO_IVA"))),
|
|
'taxes_amount_value': cents((fd.get('IMPORTE_IVA') or 0) + (fd.get('IMPORTE_RE') or 0)),
|
|
'total_amount_value': cents(fd.get('IMPORTE_TOTAL')),
|
|
|
|
'base': cents(fd.get('BASE_IMPONIBLE')),
|
|
'iva': cents(fd.get('IMPORTE_IVA')),
|
|
're': cents(fd.get('IMPORTE_RE'))
|
|
|
|
}
|
|
|
|
|
|
def normalize_details_invoice_fields(fd: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Normaliza campos del registro de origen 'fd' (factura).
|
|
"""
|
|
config = load_config()
|
|
|
|
tax_code = map_tax_code(str(fd.get("DES_TIPO_IVA")))
|
|
disc_pct = fd.get('DESCUENTO_DET')
|
|
|
|
# Calcular cuota de IVA de la línea
|
|
frac = tax_fraction_from_code(tax_code) # 0.21, 0.10, 0…
|
|
# Ttotal_det ya incluye descuento
|
|
line_base = Decimal(str(fd.get('IMPORTE_TOTAL_DET') or 0))
|
|
tax_amount = int(
|
|
(money_round(line_base * frac, 2)*100).to_integral_value())
|
|
|
|
# Se calcula en el objeto de negocio de nuevo, comprobar si coincide
|
|
# item_discount_amount = (
|
|
# (factura_detalle['IMPORTE_UNIDAD'] or 0)*((factura_detalle['DESCUENTO'] or 0)/100))*100
|
|
|
|
return {
|
|
'tax_code': tax_code,
|
|
'position': int(fd['POSICION']),
|
|
'description': rtf_a_texto_plano(str(fd['CONCEPTO'])),
|
|
'quantity_value': None if fd['CANTIDAD'] is None else int(Decimal(str(fd['CANTIDAD'] or 0)) * 100),
|
|
'unit_value': None if fd['IMPORTE_UNIDAD'] is None else int(Decimal(str(fd['IMPORTE_UNIDAD'] or 0)) * 10000),
|
|
'disc_pct': disc_pct,
|
|
'discount_percentage_value': None if disc_pct in (None, 0) else int(Decimal(str(disc_pct)) * 100),
|
|
'total_value': cents(fd.get('IMPORTE_TOTAL_DET')),
|
|
'tax_amount': tax_amount
|
|
}
|