Uecko/Uecko_ERP_FactuGES_sync
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: Uecko:931f9a6825e21f9ed9c2431175b730b07860b62b
Branches Tags
Uecko:main
...
compare: Uecko:0e3d9d4fcf2e69460f01b0a4d8d2691a1522b2cb
Branches Tags
Uecko:main

2 Commits
931f9a6825 ... 0e3d9d4fcf

Author SHA1 Message Date
david
0e3d9d4fcf funcionando con acana 2025-11-05 18:43:40 +01:00
david
405a6317ed version sinc con verifactu test 2025-10-29 17:08:14 +01:00
20 changed files with 953 additions and 1209 deletions
Show Stats Expand all files Collapse all files

20
.env.development
View File

@ -18,11 +18,17 @@ FACTUGES_CONTRATO_TIPO_DETALLE = "Concepto"
FACTUGES_NOMBRE_TARIFA = TARIFA 2024
FACTUGES_PRECIO_PUNTO = 3.31
UECKO_MYSQL_HOST = 192.168.0.104
UECKO_MYSQL_PORT = 3306
UECKO_MYSQL_DATABASE = uecko_erp_sync
UECKO_MYSQL_USER = rodax
UECKO_MYSQL_PASSWORD = rodax
PRO_UECKO_MYSQL_HOST = 192.168.0.250
PRO_UECKO_MYSQL_PORT = 3306
PRO_UECKO_MYSQL_DATABASE = factuges_db
PRO_UECKO_MYSQL_USER = root
PRO_UECKO_MYSQL_PASSWORD = rootpass
DEV_UECKO_MYSQL_HOST = 192.168.0.104
DEV_UECKO_MYSQL_PORT = 3306
DEV_UECKO_MYSQL_DATABASE = uecko_erp_sync
DEV_UECKO_MYSQL_USER = rodax
DEV_UECKO_MYSQL_PASSWORD = rodax
UECKO_DEFAULT_IVA = 2100
UECKO_DEFAULT_CURRENCY_CODE = EUR
@ -36,3 +42,7 @@ BREVO_EMAIL_TEMPLATE = 1
MAIL_FROM = 'no-reply@presupuestos.uecko.com'
MAIL_TO = 'soporte@rodax-software.com'
VERIFACTU_BASE_URL = https://api.verifacti.com/
VERIFACTU_API_KEY = vf_test_kY9FoI86dH+g1a5hmEnb/0YcLTlMFlu+tpp9iMZp020=
VERIFACTU_NIFS_API_KEY = vfn_osYpNdqSzAdTAHpazXG2anz4F3o0gfbSb5FFrCBZcno=

41
.env.production
View File

@ -8,13 +8,26 @@ FACTUGES_DATABASE = D:\RODAX\FACTUGES\BD\FACTUGES_FABRICA.FDB
FACTUGES_USER = sysdba
FACTUGES_PASSWORD = abeto2010
FACTUGES_ID_EMPRESA = 1
FACTUGES_CONTRATO_ID_TIENDA = 1
FACTUGES_CONTRATO_SITUACION = "PENDIENTE"
FACTUGES_CONTRATO_ENVIADA_REVISADA = 10
FACTUGES_CONTRATO_TIPO_DETALLE = "Concepto"
FACTUGES_NOMBRE_TARIFA = TARIFA 2024
FACTUGES_PRECIO_PUNTO = 3.31
# DESARROLLO ALISO FACTUGES FIREBIRD
#FACTUGES_HOST = 192.168.0.105
#FACTUGES_PORT = 3050
#FACTUGES_DATABASE = C:\Codigo Acana\Output\Debug\Database\FACTUGES.FDB
#FACTUGES_USER = sysdba
#FACTUGES_PASSWORD = masterkey
# PRODUCCION RODAX FACTUGES FIREBIRD
#FACTUGES_HOST = 192.168.0.101
#FACTUGES_PORT = 3050
#FACTUGES_DATABASE = C:\FactuGES\FACTUGES.FDB
#FACTUGES_USER = sysdba
#FACTUGES_PASSWORD = masterkey
# PRODUCCION RODAX MYSQL
#UECKO_MYSQL_HOST = 192.168.0.250
#UECKO_MYSQL_PORT = 3306
#UECKO_MYSQL_DATABASE = factuges_db
#UECKO_MYSQL_USER = root
#UECKO_MYSQL_PASSWORD = rootpass
UECKO_MYSQL_HOST = mariadb2
UECKO_MYSQL_PORT = 3306
@ -34,3 +47,17 @@ BREVO_EMAIL_TEMPLATE = 1
MAIL_FROM = 'no-reply@presupuestos.uecko.com'
MAIL_TO = 'pedidos@uecko.com'
VERIFACTU_BASE_URL = https://api.verifacti.com/
VERIFACTU_API_KEY = vf_test_kY9FoI86dH+g1a5hmEnb/0YcLTlMFlu+tpp9iMZp020=
VERIFACTU_NIFS_API_KEY = vfn_osYpNdqSzAdTAHpazXG2anz4F3o0gfbSb5FFrCBZcno=
#VERIFACTU_RODAX_TEST_API_KEY = vf_test_C03HL2F0X5OXSDRunjNFoMxD4IrRfK3kCC8PfcvCENI=
#VERIFACTU_RODAX_PROD_API_KEY = vf_prod_yfjonNPv2E4Fij+5J0hct0zCgUeFYT2dZzb23UZlM+Q=
#VERIFACTU_ALISO_TEST_API_KEY = vf_test_ei8WYAvEq5dhSdEyQVjgCS8NZaNpEK2BljSHSUXf+Y0=

2
app/__version__.py
View File

@ -1 +1 @@
__version__ = "1.0.8"
__version__ = "1.0.0"

4
app/config/settings.py
View File

@ -43,4 +43,8 @@ def load_config():
'BREVO_EMAIL_TEMPLATE': os.getenv("BREVO_EMAIL_TEMPLATE"),
'MAIL_FROM': os.getenv('MAIL_FROM'),
'MAIL_TO': os.getenv('MAIL_TO'),
'VERIFACTU_BASE_URL': os.getenv('VERIFACTU_BASE_URL'),
'VERIFACTU_API_KEY': os.getenv('VERIFACTU_API_KEY'),
'VERIFACTU_NIFS_API_KEY': os.getenv('VERIFACTU_NIFS_API_KEY'),
}

3
app/db/__init__.py
View File

@ -1,7 +1,4 @@
from .db_connection import get_factuges_connection
from .db_connection import get_mysql_connection
from .sync_catalog import sync_catalog
from .sync_dealers import sync_dealers
from .sync_orders import sync_orders
from .sync_invoices import sync_invoices
from .sync_invoices_verifactu import sync_invoices_verifactu

74
app/db/sql_sentences.py Normal file
View File

@ -0,0 +1,74 @@
# =========================
# SQL (constantes)
# =========================
SELECT_CUSTOMER_BY_FACTUGES = (
"SELECT customers.id FROM customers WHERE customers.factuges_id=%s"
)
INSERT_CUSTOMER = (
"INSERT INTO customers (id, name, tin, street, city, province, postal_code, country, "
"phone_primary, phone_secondary, mobile_primary, mobile_secondary, "
"email_primary, email_secondary, website, factuges_id, company_id, is_company, "
"language_code, currency_code, status, created_at, updated_at) "
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,1,'es','EUR','active',CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)"
)
UPDATE_CUSTOMER = (
"UPDATE customers SET name=%s, tin=%s, street=%s, city=%s, province=%s, postal_code=%s, country=%s, "
"phone_primary=%s, phone_secondary=%s, mobile_primary=%s, mobile_secondary=%s, "
"email_primary=%s, email_secondary=%s, website=%s, updated_at=CURRENT_TIMESTAMP "
"WHERE (id=%s)"
)
SELECT_PAYMENT_METHOD_BY_FACTUGES = (
"SELECT payment_methods.id FROM payment_methods WHERE payment_methods.factuges_id=%s"
)
INSERT_PAYMENT_METHOD = (
"INSERT INTO payment_methods (id, description, factuges_id, created_at, updated_at) "
"VALUES (%s,%s,%s,CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)"
)
INSERT_INVOICE = (
"INSERT INTO customer_invoices (id, company_id, invoice_number, status, series, reference, invoice_date, operation_date, description, "
"subtotal_amount_value, discount_amount_value, discount_percentage_value, taxable_amount_value, taxes_amount_value, total_amount_value, "
"customer_id, customer_tin, customer_name, customer_street, customer_city, customer_province, customer_postal_code, customer_country, "
"payment_method_id, payment_method_description, "
"subtotal_amount_scale, discount_amount_scale, discount_percentage_scale, taxable_amount_scale, taxes_amount_scale, total_amount_scale, "
"language_code, currency_code, created_at, updated_at) "
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,2,2,2,2,2,2,'es','EUR',CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)"
)
INSERT_INVOICE_ITEM = (
"INSERT INTO customer_invoice_items "
"(item_id, invoice_id, position, description, quantity_value, unit_amount_value, "
"discount_percentage_value, discount_amount_value, total_amount_value, "
"quantity_scale, unit_amount_scale, discount_amount_scale, total_amount_scale, discount_percentage_scale, created_at, updated_at) "
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,2,4,2,4,2,CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)"
)
INSERT_INVOICE_TAX = (
"INSERT INTO customer_invoice_taxes (tax_id, invoice_id, tax_code, taxable_amount_value, taxes_amount_value, "
"taxable_amount_scale, taxes_amount_scale, created_at, updated_at) "
"VALUES (%s,%s,%s,%s,%s,2,2,CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)"
)
INSERT_INVOICE_ITEM_TAX = (
"INSERT INTO customer_invoice_item_taxes "
"(tax_id, item_id, tax_code, taxable_amount_value, taxes_amount_value, taxable_amount_scale, taxes_amount_scale, created_at, updated_at) "
"VALUES (%s,%s,%s,%s,%s,4,2,CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)"
)
UPDATE_FACTUGES_LINK = (
"UPDATE FACTURAS_CLIENTE "
"SET ID_VERIFACTU=? "
"WHERE ID=?"
)
LIMPIAR_FACTUGES_LINK = (
"UPDATE FACTURAS_CLIENTE "
"SET ID_VERIFACTU = NULL, "
"VERIFACTU = 0 "
"WHERE (ID_VERIFACTU = ?)"
)

219
app/db/sync_catalog.py
View File

@ -1,219 +0,0 @@
import logging
from uuid import uuid4
from config import load_config
def sync_catalog(conn_factuges, conn_mysql, last_execution_date):
config = load_config()
logging.info(f"Tarifa: {config['FACTUGES_NOMBRE_TARIFA']}")
logging.info(f"Precio punto: {config['FACTUGES_PRECIO_PUNTO']}")
# Construir la consulta SQL con la condición de fecha de modificación
consulta_sql_art_modificados = (
f"SELECT art.id || '' AS id, art.tarifa as tarifa, COALESCE(art.referencia,'') AS referencia, "
f"TRIM(COALESCE(art.familia, '') || ' ' || COALESCE(art.referencia_prov, '') || ' ' || COALESCE(art.descripcion, '')) as descripcion_es, "
f"TRIM(COALESCE(art_idioma_en.descripcion, '')) AS descripcion_en, "
f"TRUNC(art.precio_coste * 100) || '' AS puntos, "
f"TRUNC(ROUND(art.precio_coste * {
config['FACTUGES_PRECIO_PUNTO']}, 2) * 100) || '' AS pvp "
f"FROM articulos AS art "
f"LEFT JOIN ARTICULOS_IDIOMAS AS art_idioma_en ON art.id = art_idioma_en.id_articulo AND art_idioma_en.id_idioma = 2 "
f"WHERE "
f"(art.eliminado = 0) AND "
f"(art.tarifa = '{config['FACTUGES_NOMBRE_TARIFA']}') "
f"AND (art.FECHA_MODIFICACION > '{last_execution_date}')"
)
consulta_sql_all_tarifa = (
f"SELECT art.id || '' AS id, art.tarifa as tarifa "
f"FROM articulos AS art "
f"WHERE "
f"(art.eliminado = 0) AND "
f"(art.tarifa = '{config['FACTUGES_NOMBRE_TARIFA']}') "
)
# Crear un cursor para ejecutar consultas SQL
cursor_FactuGES = None
try:
cursor_FactuGES = conn_factuges.cursor()
# Ejecutar la consulta de articulos modificados
cursor_FactuGES.execute(consulta_sql_art_modificados)
filas = cursor_FactuGES.fetchall()
except Exception as e:
if cursor_FactuGES is not None:
cursor_FactuGES.close()
logging.error(f"(ERROR) Failed to fetch from database:{
config['FACTUGES_DATABASE']} - using user:{config['FACTUGES_USER']}")
logging.error(e)
raise e
# Obtener los nombres de las columnas
columnas = [desc[0] for desc in cursor_FactuGES.description]
cursor_FactuGES.close()
# Convertir las filas en diccionarios con nombres de columnas como claves
tuplas_seleccionadas = []
for fila in filas:
tupla = dict(zip(columnas, fila))
tuplas_seleccionadas.append(tupla)
logging.info(f"Catalog rows to be processed: {len(tuplas_seleccionadas)}")
# Verificar si hay filas en el resultado
if tuplas_seleccionadas:
insertar_datos(conn_mysql, tuplas_seleccionadas, config)
else:
logging.info(
"There are no new or modified catalog rows since the last run.")
# Verificamos que en el catálogo de mysql solo hay los artículos del catálogo de FactuGES
# es decir, que si un artículo lo asignan a otra tarifa debe desaparecer del catálogo mysql
try:
cursor_FactuGES.execute(consulta_sql_all_tarifa)
filas = cursor_FactuGES.fetchall()
cursor_FactuGES.close()
# Crear un conjunto con los IDs [0] de los artículos en FactuGES para una búsqueda rápida
ids_factuges = {str(fila[0]) for fila in filas}
logging.info(f"{config['FACTUGES_NOMBRE_TARIFA']} rows to be processed: {
len(ids_factuges)}")
# Verificar si hay filas en el resultado
if ids_factuges:
eliminar_datos(conn_mysql, ids_factuges, config)
else:
logging.info(f"There are no rows in the {
config['FACTUGES_NOMBRE_TARIFA']}.")
except Exception as e:
if cursor_FactuGES is not None:
cursor_FactuGES.close()
logging.error(f"(ERROR) Failed to fetch from database:{
config['FACTUGES_DATABASE']} - using user:{config['FACTUGES_USER']}")
logging.error(e)
raise e
def eliminar_datos(conn_mysql, ids_factuges, config):
# Recorrer todos los articulos del catálogo web para ver si estan en filas, si no están se eliminan
select_all_catalog_query = (
"SELECT catalog.id, catalog.id_article FROM catalog"
)
delete_catalog_query = (
"DELETE FROM catalog WHERE catalog.id_article = %s"
)
cursorMySQL = None
try:
cursorMySQL = conn_mysql.cursor()
cursorMySQL.execute(select_all_catalog_query)
catalog_rows = cursorMySQL.fetchall()
logging.info(
f">>>>Comprobar que todos los artículos del catálogo existen en FactuGES")
ids_a_eliminar = [
catalog_row[1] # id_article
for catalog_row in catalog_rows
if str(catalog_row[1]) not in ids_factuges
]
if ids_a_eliminar:
logging.info(f"Deleting articles: {ids_a_eliminar}")
cursorMySQL.executemany(delete_catalog_query, [(
id_article,) for id_article in ids_a_eliminar])
else:
logging.info("No articles to delete.")
except Exception as e:
# Escribir el error en el archivo de errores
logging.error(str(e))
raise e # Re-lanzar la excepción para detener el procesamiento
finally:
# Cerrar la conexión
if cursorMySQL is not None:
cursorMySQL.close()
def insertar_datos(conn_mysql, filas, config):
insert_catalog_query = (
"INSERT INTO catalog (id, catalog_name, id_article, points, retail_price, created_at, updated_at) "
"VALUES (%s, %s, %s, %s, %s, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)"
)
insert_translation_query = (
"INSERT INTO catalog_translations (id, lang_code, catalog_id, description) "
"VALUES (%s, %s, %s, %s)"
)
update_catalog_query = (
"UPDATE catalog set "
"points = %s, "
"retail_price = %s, "
"updated_at = Now() "
"WHERE id_article=%s"
)
update_translation_query = (
"UPDATE catalog_translations SET "
"description = %s "
"WHERE lang_code = %s AND "
"catalog_id IN (SELECT catalog.id FROM catalog WHERE catalog.id_article = %s)"
)
select_catalog_query = (
"SELECT count(catalog.id) FROM catalog WHERE catalog.id_article = %s"
)
cursorMySQL = None
try:
cursorMySQL = conn_mysql.cursor()
# Insertar datos en la tabla 'catalog'
for articulo in filas:
# Generar un ID único para la tabla catalog
id_catalog = str(uuid4())
id_article = int(articulo['ID'])
points = int(articulo['PUNTOS'])
retail_price = int(articulo['PVP'])
tarifa = config['FACTUGES_NOMBRE_TARIFA']
cursorMySQL.execute(select_catalog_query, (id_article, ))
row_count = cursorMySQL.fetchone()
is_new = row_count[0] < 1
if is_new:
logging.info(f"Inserting article {id_article} {tarifa}")
cursorMySQL.execute(
insert_catalog_query, (id_catalog, tarifa, id_article, points, retail_price))
else:
logging.info(f"Updating article {id_article} {tarifa}")
cursorMySQL.execute(update_catalog_query,
(points, retail_price, id_article))
# Insertar traducciones en la tabla 'catalog_translations'
for lang_code, desc_key in [('es', 'DESCRIPCION_ES'), ('en', 'DESCRIPCION_EN')]:
descripcion_traducida = articulo.get(desc_key, '')
if descripcion_traducida:
if (is_new):
logging.info(f"Inserting translation {
lang_code} {descripcion_traducida}")
# Generar un ID único para cada traducción
id_translation = str(uuid4())
cursorMySQL.execute(
insert_translation_query, (id_translation, lang_code, id_catalog, descripcion_traducida))
else:
logging.info(f"Updating translation {
lang_code} {descripcion_traducida}")
cursorMySQL.execute(
update_translation_query, (descripcion_traducida, lang_code, id_article))
except Exception as e:
# Escribir el error en el archivo de errores
logging.error(str(e))
raise e # Re-lanzar la excepción para detener el procesamiento
finally:
# Cerrar la conexión
if cursorMySQL is not None:
cursorMySQL.close()

320
app/db/sync_dealers.py
View File

@ -1,320 +0,0 @@
import logging
from uuid import uuid4
from config import load_config
from utils import hashPassword
def sync_dealers(conn_factuges, conn_mysql, last_execution_date):
config = load_config()
consulta_factuges = (
f"select V_CONTACTOS.ID as ID, V_CONTACTOS.NOMBRE, V_CONTACTOS.IDIOMA_ISO, "
f"CLIENTES_DATOS.DIST_EMAIL, CLIENTES_DATOS.DIST_PASSWORD, CLIENTES_DATOS.BLOQUEADO "
f"from V_CONTACTOS "
f"left OUTER JOIN CLIENTES_DATOS on (V_CONTACTOS.ID = CLIENTES_DATOS.ID_CLIENTE) "
f"where (V_CONTACTOS.ID_CATEGORIA = 1) "
f"and (V_CONTACTOS.ID_EMPRESA = '{config['FACTUGES_ID_EMPRESA']}') "
f"and (CLIENTES_DATOS.TIENDA_WEB = 1) "
f"and (V_CONTACTOS.FECHA_MODIFICACION is not null) "
f"and (V_CONTACTOS.FECHA_MODIFICACION > '{last_execution_date}')"
)
consulta_dealer_uecko = (
"SELECT dealers.id, dealers.id_contact, dealers.user_id, dealers.status, dealers.updated_at "
"FROM dealers "
"WHERE dealers.id_contact = %s"
)
cursor_FactuGES = None
try:
cursor_FactuGES = conn_factuges.cursor()
# Ejecutar la consulta
cursor_FactuGES.execute(consulta_factuges)
contactos = cursor_FactuGES.fetchall()
except Exception as e:
if cursor_FactuGES is not None:
cursor_FactuGES.close()
logging.error(f"(ERROR) Failed to fetch from database:{
config['FACTUGES_DATABASE']} - using user:{config['FACTUGES_USER']}")
logging.error(e)
raise e
columnas_contacto = [desc[0] for desc in cursor_FactuGES.description]
cursor_FactuGES.close()
contactos_seleccionados = []
for contacto in contactos:
tupla = dict(zip(columnas_contacto, contacto))
contactos_seleccionados.append(tupla)
logging.info(f"Contacts rows to be processed: {
len(contactos_seleccionados)}")
if contactos_seleccionados:
for contacto in contactos_seleccionados:
cursor_MySQL = None
try:
cursor_MySQL = conn_mysql.cursor()
cursor_MySQL.execute(consulta_dealer_uecko, (contacto['ID'],))
dealer = cursor_MySQL.fetchone()
if (dealer is None):
user_id = insert_user(conn_mysql, contacto, config)
insert_dealer(conn_mysql, user_id, contacto, config)
logging.info(f"Inserted user and dealer from contact {
contacto['ID']} {contacto['NOMBRE']}")
else:
# 0 => 'ID'
# 2 => 'USER_ID'
# Casos:
# - Cambio en el nombre del distribuidor
# - Distribuidor bloqueado / desbloqueado
# - Usuario con baja lógica
id = dealer[0]
user_id = dealer[2]
update_dealer(conn_mysql, id, contacto, config)
update_user(conn_mysql, user_id, contacto, config)
except Exception as e:
# Escribir el error en el archivo de errores
logging.error(str(e))
raise e # Re-lanzar la excepción para detener el procesamiento
finally:
# Cerrar la conexión
if cursor_MySQL is not None:
cursor_MySQL.close()
else:
logging.info(
"There are no new or modified contacts rows since the last run.")
# Revisar todos los distribuidores dados de alta y
# comprobar si en FactuGES siguen estando activos (tienda_web = 1)
#
# - USUARIO DISABLED
dealers = fetch_all_dealers(conn_mysql, config)
for dealer in dealers:
# dealer[1] => id_contact
if (dealer[1] is not None) and (not is_valid_dealer(conn_factuges, dealer[1], config)):
user_id = dealer[7] # 7 => user_id
# Desactivar el distribuidor
disable_dealer(conn_mysql, dealer[0], config) # 0 => id
# Baja lógica del usuario del dealer
soft_delete_user(conn_mysql, user_id, config)
logging.info(f"Deleted dealer and user from contact {dealer[1]}")
def fetch_all_dealers(conn_mysql, config):
consulta = (
f"SELECT dealers.id, dealers.id_contact, dealers.default_payment_method, dealers.default_notes, "
f"dealers.default_legal_terms, dealers.default_quote_validity, dealers.status, dealers.user_id "
f"FROM dealers "
)
cursor_MySQL = None
try:
cursor_MySQL = conn_mysql.cursor()
cursor_MySQL.execute(consulta)
return cursor_MySQL.fetchall()
except Exception as e:
# Escribir el error en el archivo de errores
logging.error(str(e))
raise e # Re-lanzar la excepción para detener el procesamiento
finally:
if cursor_MySQL is not None:
cursor_MySQL.close()
def is_valid_dealer(conn_factuges, id_contact, config):
consulta = (
f"select CLIENTES_DATOS.ID_CLIENTE from CLIENTES_DATOS where CLIENTES_DATOS.ID_CLIENTE = {
id_contact} and CLIENTES_DATOS.TIENDA_WEB = 1"
)
cursor_FactuGES = None
try:
cursor_FactuGES = conn_factuges.cursor()
cursor_FactuGES.execute(consulta)
exists = cursor_FactuGES.fetchone()
return exists is not None
except Exception as e:
# Escribir el error en el archivo de errores
logging.error(str(e))
raise e # Re-lanzar la excepción para detener el procesamiento
finally:
if cursor_FactuGES is not None:
cursor_FactuGES.close()
def insert_user(conn_mysql, data, config):
cursor_MySQL = None
try:
cursor_MySQL = conn_mysql.cursor()
id = str(uuid4())
name = str(data['NOMBRE'])
email = str(data['DIST_EMAIL'])
password = hashPassword(str(data['DIST_PASSWORD']))
lang_code = str(data['IDIOMA_ISO'])
insert_data = (
"INSERT INTO users (id, name, email, password, lang_code, roles, created_at, updated_at) VALUES ("
"%s, %s, %s, %s, %s, 'ROLE_USER', Now(), Now()"
")"
)
cursor_MySQL.execute(
insert_data, (id, name, email, password, lang_code))
return id
except Exception as e:
# Escribir el error en el archivo de errores
logging.error(str(e))
raise e # Re-lanzar la excepción para detener el procesamiento
finally:
if cursor_MySQL is not None:
cursor_MySQL.close()
def update_user(conn_mysql, user_id, data, config):
cursor_MySQL = None
try:
cursor_MySQL = conn_mysql.cursor()
name = str(data['NOMBRE'])
email = str(data['DIST_EMAIL'])
password = hashPassword(str(data['DIST_PASSWORD']))
lang_code = str(data['IDIOMA_ISO'])
update_data = (
"UPDATE users set "
"name = %s, "
"email = %s, "
"password = %s, "
"lang_code = %s, "
"updated_at = Now(), "
"deleted_at = NULL "
"WHERE id = %s"
)
cursor_MySQL.execute(
update_data, (name, email, password, lang_code, user_id))
logging.info(f"Updated user from contact {data['ID']} {name}")
except Exception as e:
# Escribir el error en el archivo de errores
logging.error(str(e))
raise e # Re-lanzar la excepción para detener el procesamiento
finally:
if cursor_MySQL is not None:
cursor_MySQL.close()
def insert_dealer(conn_mysql, user_id, data, config):
cursor_MySQL = None
try:
cursor_MySQL = conn_mysql.cursor()
id = str(uuid4())
id_contact = str(data['ID'])
name = str(data['NOMBRE'])
default_payment_method = str(config['UECKO_DEFAULT_FORMA_PAGO'])
default_notes = str(config['UECKO_DEFAULT_NOTAS'])
default_legal_terms = str(config['UECKO_DEFAULT_LOPD'])
default_quote_validity = str(config['UECKO_DEFAULT_VALIDEZ'])
default_tax = str(config["UECKO_DEFAULT_IVA"])
lang_code = str(data['IDIOMA_ISO'])
currency_code = str(config["UECKO_DEFAULT_CURRENCY_CODE"])
insert_data = (
"INSERT INTO dealers (id, id_contact, name, default_payment_method, default_notes, "
"default_legal_terms, default_quote_validity, default_tax, lang_code, "
"currency_code, user_id, status, created_at, updated_at ) values ("
"%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'actived', Now(), Now()"
")"
)
cursor_MySQL.execute(insert_data, (id, id_contact, name, default_payment_method, default_notes,
default_legal_terms, default_quote_validity, default_tax, lang_code, currency_code, user_id))
return id
except Exception as e:
# Escribir el error en el archivo de errores
logging.error(str(e))
raise e # Re-lanzar la excepción para detener el procesamiento
finally:
if cursor_MySQL is not None:
cursor_MySQL.close()
def update_dealer(conn_mysql, dealer_id, data, config):
cursor_MySQL = None
try:
cursor_MySQL = conn_mysql.cursor()
name = str(data['NOMBRE'])
status = 'disabled' if data['BLOQUEADO'] == 1 else 'actived'
insert_data = (
"UPDATE dealers SET name = %s, status = %s WHERE dealers.id = %s"
)
cursor_MySQL.execute(insert_data, (name, status, dealer_id))
logging.info(f"Dealer with id = {dealer_id} name = {name} updated")
except Exception as e:
# Escribir el error en el archivo de errores
logging.error(str(e))
raise e # Re-lanzar la excepción para detener el procesamiento
finally:
if cursor_MySQL is not None:
cursor_MySQL.close()
def soft_delete_user(conn_mysql, user_id, config):
consulta_sql = "UPDATE users SET users.deleted_at = NOW() WHERE users.id = %s AND users.roles = 'ROLE_USER'"
cursor_MySQL = None
try:
cursor_MySQL = conn_mysql.cursor()
cursor_MySQL.execute(consulta_sql, (user_id, ))
logging.info(f"User with id = {id} soft delete")
except Exception as e:
# Escribir el error en el archivo de errores
logging.error(str(e))
raise e # Re-lanzar la excepción para detener el procesamiento
finally:
if cursor_MySQL is not None:
cursor_MySQL.close()
def active_dealer(conn_mysql, id, config):
consulta_sql = "UPDATE dealers SET status = 'actived' WHERE dealers.id = %s"
cursor_MySQL = None
try:
cursor_MySQL = conn_mysql.cursor()
cursor_MySQL.execute(consulta_sql, (id, ))
logging.info(f"Dealer with id = {id} actived")
except Exception as e:
# Escribir el error en el archivo de errores
logging.error(str(e))
raise e # Re-lanzar la excepción para detener el procesamiento
finally:
if cursor_MySQL is not None:
cursor_MySQL.close()
def disable_dealer(conn_mysql, id, config):
consulta_sql = "UPDATE dealers SET status = 'disabled' WHERE dealers.id = %s"
cursor_MySQL = None
try:
cursor_MySQL = conn_mysql.cursor()
cursor_MySQL.execute(consulta_sql, (id, ))
logging.info(f"Dealer with id = {id} disabled")
except Exception as e:
# Escribir el error en el archivo de errores
logging.error(str(e))
raise e # Re-lanzar la excepción para detener el procesamiento
finally:
if cursor_MySQL is not None:
cursor_MySQL.close()

208
app/db/sync_invoices.py
View File

@ -1,7 +1,10 @@
import logging
from uuid import uuid4
import textwrap
from . import sql_sentences as SQL
from uuid6 import uuid7
from config import load_config
from decimal import Decimal, ROUND_HALF_UP
from utils import limpiar_cadena, normalizar_telefono_con_plus, corregir_y_validar_email, normalizar_url_para_insert
def sync_invoices(conn_factuges, conn_mysql, last_execution_date):
@ -36,16 +39,17 @@ def sync_invoices(conn_factuges, conn_mysql, last_execution_date):
f"ORDER BY (fac.ID)"
)
# LIMPIAMOS LAS FACTURAS DE FACTUGES QUE HAYAN SIDO ELIMINADAS DEL PROGRAMA NUEVO DE FACTURACION, PARA QUE PUEDAN SER MODIFICADAS
# Crear un cursor para ejecutar consultas SQL
cursor_mysql = None
try:
cursor_mysql = conn_mysql.cursor()
# Ejecutar la consulta de FACTURAS_CLIENTE
# Ejecutar la consulta de FACTURAS_CLIENTE
cursor_mysql.execute(consulta_sql_customer_invoices_deleted)
filas = cursor_mysql.fetchall()
cursor_mysql.close()
# Crear un conjunto con los IDs [0] de los customer_inovices que debo liberar en FactuGES
# Crear un conjunto con los IDs [0] de los customer_inovices que debo liberar en FactuGES, porque han sido eliminadas en programa de facturación nuevo
ids_verifactu_deleted = {str(fila[0]) for fila in filas}
logging.info(f"Customer invoices rows to be deleted: {
len(ids_verifactu_deleted)}")
@ -64,6 +68,7 @@ def sync_invoices(conn_factuges, conn_mysql, last_execution_date):
logging.error(e)
raise e
# BUSCAMOS FACTURAS ENVIADAS A VERIFACTU EN FACTUGES, PARA SUBIRLAS AL NUEVO PROGRAMA DE FACTURACIÓN
# Crear un cursor para ejecutar consultas SQL
cursor_FactuGES = None
try:
@ -99,52 +104,16 @@ def sync_invoices(conn_factuges, conn_mysql, last_execution_date):
logging.info(
"There are no new FACTURAS rows since the last run.")
# Verificamos que en customer_invoice de mysql no se ha eliminado ninguna factura,
# si se ha eliminado alguna factura, procedemos quitar la asociación en factuges para que se pueda modificar
# en un futuro se modificará solo en el programa nuevo y tendrá que sincronizarse con factuges los importes totales de la factura
# ya pasamos de los detalles ya que no se van a ver las facturas en FactuGES, pero la relación que tengan las facturas con otros módulos
# deben verse, ejemplo contratos-facturas
# try:
# cursor_FactuGES.execute(consulta_sql_FACTURAS_CLIENTE)
# filas = cursor_FactuGES.fetchall()
# cursor_FactuGES.close()
# Crear un conjunto con los IDs [0] de los artículos en FactuGES para una búsqueda rápida
# ids_factuges = {str(fila[0]) for fila in filas}
# logging.info(f"customer_invoice rows to be processed: {
# len(ids_factuges)}")
# Verificar si hay filas en el resultado
# if ids_factuges:
# eliminar_datos(conn_mysql, ids_factuges, config)
# else:
# logging.info(f"There are no rows in the {
# config['FACTUGES_NOMBRE_TARIFA']}.")
# except Exception as e:
# if cursor_FactuGES is not None:
# cursor_FactuGES.close()
# logging.error(f"(ERROR) Failed to fetch from database:{
# config['FACTUGES_DATABASE']} - using user:{config['FACTUGES_USER']}")
# logging.error(e)
# raise e
def eliminar_datos(conn_factuges, ids_verifactu_deleted, config):
# Recorrer todos los articulos del catálogo web para ver si estan en filas, si no están se eliminan
update_facturas_cliente_query = (
f"UPDATE FACTURAS_CLIENTE "
f"SET ID_VERIFACTU = NULL, "
f"VERIFACTU = 0 "
f"WHERE (ID_VERIFACTU = ?)"
)
# Eliminamos todos los IDs de verifacti que han sido eliminados así liberaremos la factura borrador y podermos modificarla de nuevo, para volverla a subir una vez hechos los cambios.
cursor_FactuGES = None
try:
cursor_FactuGES = conn_factuges.cursor()
if ids_verifactu_deleted:
logging.info(f"Liberate factures: {ids_verifactu_deleted}")
cursor_FactuGES.executemany(update_facturas_cliente_query, [(
cursor_FactuGES.executemany(SQL.LIMPIAR_FACTUGES_LINK, [(
id_verifactu,) for id_verifactu in ids_verifactu_deleted])
else:
logging.info("No articles to delete.")
@ -160,60 +129,10 @@ def eliminar_datos(conn_factuges, ids_verifactu_deleted, config):
def insertar_datos(conn_mysql, filas, conn_factuges, config):
# Insertaremos cada factura existente en las filas a la nueva estructura de tablas del programa nuevo de facturacion.
# Compañia RODAX
cte_company_id = '5e4dc5b3-96b9-4968-9490-14bd032fec5f'
insert_customer_query = (
"INSERT INTO customers (id, name, tin, street, city, province, postal_code, country, phone_primary, phone_secondary, mobile_primary, mobile_secondary, "
"email_primary, email_secondary, website, factuges_id, company_id, is_company, language_code, currency_code, status, created_at, updated_at ) "
"VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 1, 'es', 'EUR', 'active', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) "
)
insert_payment_methods_query = (
"INSERT INTO payment_methods (id, description, factuges_id, created_at, updated_at ) "
"VALUES (%s, %s, %s, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) "
)
insert_customer_invoices_query = (
"INSERT INTO customer_invoices (id, company_id, status, series, reference, invoice_date, operation_date, "
"subtotal_amount_value, discount_amount_value, discount_percentage_value, taxable_amount_value, taxes_amount_value, total_amount_value, "
"customer_id, customer_tin, customer_name, customer_street, customer_city, customer_province, customer_postal_code, customer_country, "
"payment_method_id, payment_method_description, "
"subtotal_amount_scale, discount_amount_scale, discount_percentage_scale, taxable_amount_scale, taxes_amount_scale, total_amount_scale, "
"language_code, currency_code, created_at, updated_at) "
"VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 2, 2, 2, 2, 2, 2, 'es', 'EUR', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)"
)
insert_customer_invoices_taxes_query = (
"INSERT INTO customer_invoice_taxes (tax_id, invoice_id, tax_code, taxable_amount_value, taxes_amount_value, taxable_amount_scale, taxes_amount_scale, "
"created_at, updated_at) "
"VALUES (%s, %s, %s, %s, %s, 2, 2, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)"
)
insert_customer_invoice_items_query = (
"INSERT INTO customer_invoice_items (item_id, invoice_id, position, description, quantity_value, unit_amount_value, discount_percentage_value, discount_amount_value, total_amount_value, "
"quantity_scale, unit_amount_scale, discount_amount_scale, total_amount_scale, discount_percentage_scale, created_at, updated_at) "
"VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, 2, 4, 2, 4, 2, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)"
)
insert_customer_invoice_item_taxes_query = (
"INSERT INTO customer_invoice_item_taxes (tax_id, item_id, tax_code, taxable_amount_value, taxes_amount_value, taxable_amount_scale, taxes_amount_scale, "
"created_at, updated_at) "
"VALUES (%s, %s, %s, %s, %s, 4, 2, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)"
)
update_FACTURAS_CLIENTE_query = (
"UPDATE FACTURAS_CLIENTE set ID_VERIFACTU = ? WHERE ID = ?")
select_customer_query = (
"SELECT customers.id "
"FROM customers "
"WHERE customers.factuges_id = %s"
)
select_payment_method_query = (
"SELECT payment_methods.id "
"FROM payment_methods "
"WHERE payment_methods.factuges_id = %s"
)
cursorMySQL = None
cursor_FactuGES = None
@ -223,13 +142,14 @@ def insertar_datos(conn_mysql, filas, conn_factuges, config):
try:
cursorMySQL = conn_mysql.cursor()
cursor_FactuGES = conn_factuges.cursor()
contador_serie = 0
# Insertar datos en la tabla 'customer_invoices'
for factura_detalle in filas:
factuges_id = int(factura_detalle['ID_FACTURA'])
invoice_status = str('draft')
invoice_series = str('A')
invoice_number = str(factura_detalle['REFERENCIA'])
invoice_series = str('F25/')
reference = str(factura_detalle['REFERENCIA'])
invoice_date = str(factura_detalle['FECHA_FACTURA'])
operation_date = str(factura_detalle['FECHA_FACTURA'])
# siempre tendrán 2 decimales
@ -241,14 +161,18 @@ def insertar_datos(conn_mysql, filas, conn_factuges, config):
taxable_amount_value = (factura_detalle['BASE_IMPONIBLE'] or 0)*100
# Preparamos el tipo de IVA, en FactuGES es único
tax_code = str(factura_detalle['DES_TIPO_IVA'])
if tax_code == 'IVA21':
if (tax_code == 'IVA21') or (tax_code == 'IVA 21'):
tax_code = 'iva_21'
elif tax_code == 'IVA18':
elif tax_code == 'IVA18' or (tax_code == 'IVA 18'):
tax_code = 'iva_18'
elif tax_code == 'IVA16':
elif tax_code == 'IVA16' or (tax_code == 'IVA 16'):
tax_code = 'iva_16'
elif tax_code == 'IVA10':
elif tax_code == 'IVA10' or (tax_code == 'IVA 10'):
tax_code = 'iva_10'
elif tax_code == 'IVA4' or (tax_code == 'IVA 4'):
tax_code = 'iva_4'
elif tax_code == 'EXENTO':
tax_code = 'iva_exenta'
else:
tax_code = ''
# La cuota de impuestos es el IVA + RE
@ -257,13 +181,13 @@ def insertar_datos(conn_mysql, filas, conn_factuges, config):
total_amount_value = (factura_detalle['IMPORTE_TOTAL'] or 0)*100
payment_method_id = str(uuid4())
payment_method_id = str(uuid7())
factuges_payment_method_id = str(factura_detalle['ID_FORMA_PAGO'])
payment_method_description = str(factura_detalle['DES_FORMA_PAGO'])
customer_id = str(uuid4())
customer_id = str(uuid7())
factuges_customer_id = str(factura_detalle['ID_CLIENTE'])
customer_tin = str(factura_detalle['NIF_CIF'])
customer_tin = limpiar_cadena(str(factura_detalle['NIF_CIF']))
customer_name = str(factura_detalle['NOMBRE'])
customer_street = str(factura_detalle['CALLE'])
customer_city = str(factura_detalle['POBLACION'])
@ -277,6 +201,8 @@ def insertar_datos(conn_mysql, filas, conn_factuges, config):
customer_email_secondary = factura_detalle['EMAIL_2']
customer_webside = str(factura_detalle['PAGINA_WEB'])
customer_country = 'es'
description = textwrap.shorten(
f"{reference or ''} - {customer_name or ''}", width=50, placeholder="…")
item_position = int(factura_detalle['POSICION'])
item_description = str(factura_detalle['CONCEPTO'])
@ -303,26 +229,63 @@ def insertar_datos(conn_mysql, filas, conn_factuges, config):
if factuges_id_anterior is None or factuges_id_anterior != factuges_id:
# Comprobamos si existe el cliente del primer item de la factura
cursorMySQL.execute(select_customer_query,
cursorMySQL.execute(SQL.SELECT_CUSTOMER_BY_FACTUGES,
(factuges_customer_id, ))
row = cursorMySQL.fetchone()
is_new = (row is None) or (row[0] is None)
#Validamos los campos que pueden dar conflicto
customer_phone_primary_tratado = normalizar_telefono_con_plus(customer_phone_primary)
customer_phone_secondary_tratado = normalizar_telefono_con_plus(customer_phone_secondary)
customer_mobile_primary_tratado = normalizar_telefono_con_plus(customer_mobile_primary)
customer_mobile_secondary_tratado = normalizar_telefono_con_plus(customer_mobile_secondary)
customer_webside_tratado = normalizar_url_para_insert(customer_webside)
customer_email_primary_ok, customer_email_primary_tratado = corregir_y_validar_email(customer_email_primary)
customer_email_secondary_ok, customer_email_secondary_tratado = corregir_y_validar_email(customer_email_secondary)
logging.info(f"La ficha de cliente {customer_tin} se modifican los siguientes campos:")
if (customer_phone_primary_tratado != customer_phone_primary):
logging.info(f"phone_primary ({customer_phone_primary}) se cambia por ({customer_phone_primary_tratado})")
if (customer_phone_secondary_tratado != customer_phone_secondary):
logging.info(f"phone_secondary ({customer_phone_secondary}) se cambia por ({customer_phone_secondary_tratado})")
if (customer_mobile_primary_tratado != customer_mobile_primary):
logging.info(f"mobile_primary ({customer_mobile_primary}) se cambia por ({customer_mobile_primary_tratado})")
if (customer_mobile_secondary_tratado != customer_mobile_secondary):
logging.info(f"mobile_secondary ({customer_mobile_secondary}) se cambia por ({customer_mobile_secondary_tratado})")
if (customer_webside_tratado != customer_webside):
logging.info(f"customer_webside ({customer_webside}) se cambia por ({customer_webside_tratado})")
if customer_email_primary_ok:
if (customer_email_primary_tratado != customer_email_primary):
logging.info(f"email_primary ({customer_email_primary}) se cambia por ({customer_email_primary_tratado})")
else:
logging.info(f"Omitimos email_primary invalido({customer_email_primary})")
if customer_email_secondary_ok:
if (customer_email_secondary_tratado != customer_email_secondary):
logging.info(f"email_secondary ({customer_email_secondary}) se cambia por ({customer_email_secondary_tratado})")
else:
logging.info(f"Omitimos email_secondary invalido({customer_email_secondary})")
if is_new:
logging.info(
f"Inserting customer {factuges_customer_id} {customer_tin} {customer_name}")
cursorMySQL.execute(insert_customer_query, (customer_id, customer_name, customer_tin, customer_street, customer_city, customer_province,
customer_postal_code, customer_country, customer_phone_primary, customer_phone_secondary, customer_mobile_primary,
customer_mobile_secondary, customer_email_primary, customer_email_secondary, customer_webside, factuges_customer_id, cte_company_id))
cursorMySQL.execute(SQL.INSERT_CUSTOMER, (customer_id, customer_name, customer_tin, customer_street, customer_city, customer_province,
customer_postal_code, customer_country, customer_phone_primary_tratado, customer_phone_secondary_tratado, customer_mobile_primary_tratado,
customer_mobile_secondary_tratado, customer_email_primary_tratado, customer_email_secondary_tratado, customer_webside_tratado, factuges_customer_id, cte_company_id))
else:
# Si ya exite ponemos el id del customer correspondiente
customer_id = str(row[0])
logging.info(
f"Updating customer {factuges_customer_id} {customer_id}")
# cursorMySQL.execute(update_customer_query, .....)
cursorMySQL.execute(SQL.UPDATE_CUSTOMER, (customer_name, customer_tin, customer_street, customer_city, customer_province, customer_postal_code, customer_country,
customer_phone_primary_tratado, customer_phone_secondary_tratado, customer_mobile_primary_tratado, customer_mobile_secondary_tratado,
customer_email_primary_tratado, customer_email_secondary_tratado, customer_webside_tratado, customer_id))
# Comprobamos si existe la forma de pago del primer item de la factura
cursorMySQL.execute(select_payment_method_query,
cursorMySQL.execute(SQL.SELECT_PAYMENT_METHOD_BY_FACTUGES,
(factuges_payment_method_id, ))
row = cursorMySQL.fetchone()
is_new = (row is None) or (row[0] is None)
@ -330,7 +293,7 @@ def insertar_datos(conn_mysql, filas, conn_factuges, config):
if is_new:
logging.info(
f"Inserting cuspayment method {factuges_payment_method_id} {payment_method_id} {payment_method_description}")
cursorMySQL.execute(insert_payment_methods_query, (
cursorMySQL.execute(SQL.INSERT_PAYMENT_METHOD, (
payment_method_id, payment_method_description, factuges_payment_method_id))
else:
# Si ya exite ponemos el id del customer correspondiente
@ -341,20 +304,22 @@ def insertar_datos(conn_mysql, filas, conn_factuges, config):
# Insertamos cabecera de la factura
# Generar un ID único para la tabla customer_invoices
id_customer_invoice = str(uuid4())
id_customer_invoice = str(uuid7())
contador_serie = contador_serie + 1
logging.info(
f"Inserting customer_invoice {id_customer_invoice} {invoice_number} {invoice_date}")
cursorMySQL.execute(insert_customer_invoices_query, (id_customer_invoice, cte_company_id, invoice_status, invoice_series, invoice_number, invoice_date, operation_date,
f"Inserting customer_invoice {id_customer_invoice} {reference} {invoice_date}")
cursorMySQL.execute(SQL.INSERT_INVOICE, (id_customer_invoice, cte_company_id, contador_serie, invoice_status, invoice_series, reference, invoice_date, operation_date, description,
subtotal_amount_value, discount_amount_value, discount_percentage_value, taxable_amount_value, tax_amount_value, total_amount_value,
customer_id, customer_tin, customer_name, customer_street, customer_city, customer_province, customer_postal_code, customer_country,
payment_method_id, payment_method_description))
# Insertamos el IVA y RE si viene
if (factura_detalle['IVA'] > 0):
if (factura_detalle['IVA'] >= 0):
taxable_amount_value = (
factura_detalle['BASE_IMPONIBLE'])*100
tax_amount_value = (factura_detalle['IMPORTE_IVA'])*100
cursorMySQL.execute(insert_customer_invoices_taxes_query, (str(uuid4()),
cursorMySQL.execute(SQL.INSERT_INVOICE_TAX, (str(uuid7()),
id_customer_invoice, tax_code, taxable_amount_value, tax_amount_value))
if (factura_detalle['RECARGO_EQUIVALENCIA'] > 0):
@ -362,22 +327,22 @@ def insertar_datos(conn_mysql, filas, conn_factuges, config):
taxable_amount_value = (
factura_detalle['BASE_IMPONIBLE'])*100
tax_amount_value = (factura_detalle['IMPORTE_RE'])*100
cursorMySQL.execute(insert_customer_invoices_taxes_query, (str(uuid4()),
cursorMySQL.execute(SQL.INSERT_INVOICE_TAX, (str(uuid7()),
id_customer_invoice, tax_code, taxable_amount_value, tax_amount_value))
# Guardamos en Factuges el id de la customer_invoice
logging.info(
f"Updating FACTURAS_CLIENTE {id_customer_invoice} {factuges_id}")
cursor_FactuGES.execute(
update_FACTURAS_CLIENTE_query, (id_customer_invoice, factuges_id))
SQL.UPDATE_FACTUGES_LINK, (id_customer_invoice, factuges_id))
num_fac_procesed += 1
# Insertamos detalles y taxes correspondientes siempre
# Generar un ID único para la tabla customer_invoice_items
item_id = str(uuid4())
item_id = str(uuid7())
logging.info(
f"Inserting customer_invoice_items {id_customer_invoice} {item_position} {item_quantity_value}")
cursorMySQL.execute(insert_customer_invoice_items_query, (item_id, id_customer_invoice, item_position, item_description,
cursorMySQL.execute(SQL.INSERT_INVOICE_ITEM, (item_id, id_customer_invoice, item_position, item_description,
item_quantity_value, item_unit_amount_value, item_discount_percentage_value, item_discount_amount, item_total_amount))
if tax_code == 'iva_21':
@ -392,13 +357,16 @@ def insertar_datos(conn_mysql, filas, conn_factuges, config):
elif tax_code == 'iva_10':
tax_amount_value = (
(Decimal(str(factura_detalle['IMPORTE_TOTAL_DET'] or 0))*Decimal('0.10')).quantize(Decimal('0.01'), rounding=ROUND_HALF_UP))*100
elif tax_code == 'iva_exenta':
tax_amount_value = (
(Decimal(str(factura_detalle['IMPORTE_TOTAL_DET'] or 0))*Decimal('0')).quantize(Decimal('0.01'), rounding=ROUND_HALF_UP))*100
else:
tax_amount_value = (
factura_detalle['IMPORTE_TOTAL_DET'] or 0)*100
logging.info(
f"Inserting customer_invoice_item_taxes {item_id} {item_position} {tax_code} {item_total_amount} {tax_amount_value}")
cursorMySQL.execute(insert_customer_invoice_item_taxes_query, (str(uuid4()), item_id, tax_code,
cursorMySQL.execute(SQL.INSERT_INVOICE_ITEM_TAX, (str(uuid7()), item_id, tax_code,
item_total_amount, tax_amount_value))
# Asignamos el id factura anterior para no volver a inserta cabecera

464
app/db/sync_invoices_verifactu.py
View File

@ -1,7 +1,9 @@
import logging
from uuid import uuid4
from typing import Dict, Any, Tuple, Optional, List, Iterable
from config import load_config
from decimal import Decimal
from utils import validar_nif, estado_factura, crear_factura, TaxCatalog, unscale_to_str
def sync_invoices_verifactu(conn_mysql, last_execution_date):
@ -22,18 +24,21 @@ def sync_invoices_verifactu(conn_mysql, last_execution_date):
# FROM customer_invoices as ci
# LEFT JOIN customer_invoice_items cii on (ci.id = cii.invoice_id)
# LEFT JOIN customer_invoice_item_taxes ciit on (cii.item_id = ciit.item_id)
# WHERE (ci.is_proforma = 0) AND (ci.status= 'draft')
# WHERE (ci.is_proforma = 0) AND (ci.status= 'issued')
# group by 1,2,3,4,5,6,7,8,9
# Recorrer todas las facturas emitidas para madarlas o refrescar los campos
consulta_sql_customer_invoices_issue = (
f"SELECT ci.id, ci.series, ci.invoice_number, ci.invoice_date, ci.description, ci.customer_tin, ci.customer_name, ci.total_amount_value, "
f"cit.tax_code, sum(cit.taxable_amount_value), sum(cit.taxes_amount_value) "
f"SELECT ci.id, ci.series, ci.invoice_number, ci.invoice_date, ci.description, ci.customer_tin, ci.customer_name, ci.total_amount_value, ci.total_amount_scale, ci.reference, "
f"cit.taxable_amount_scale, cit.taxes_amount_scale, cit.tax_code, sum(cit.taxable_amount_value) as taxable_amount_value, sum(cit.taxes_amount_value) as taxes_amount_value, "
f"vr.id as vrId, vr.uuid, vr.estado "
f"FROM customer_invoices as ci "
f"LEFT JOIN customer_invoice_taxes cit on (ci.id = cit.invoice_id) "
f"WHERE "
f"(ci.is_proforma = 0) AND (ci.status= 'draft')"
f"group by 1,2,3,4,5,6,7,8,9"
f"LEFT JOIN verifactu_records vr on (ci.id = vr.invoice_id) "
f"WHERE (ci.is_proforma = 0) AND (ci.status= 'issued') "
f"AND ((vr.estado is null) OR (vr.estado <> 'Correcto')) "
f"group by 1,2,3,4,5,6,7,8,9,10,11 "
f"order by reference"
)
# Crear un cursor para ejecutar consultas SQL
@ -46,60 +51,74 @@ def sync_invoices_verifactu(conn_mysql, last_execution_date):
# Obtener los nombres de las columnas
columnas = [desc[0] for desc in cursor_mysql.description]
cursor_mysql.close()
# Convertir las filas en diccionarios con nombres de columnas como claves
tuplas_seleccionadas = []
for fila in filas:
tupla = dict(zip(columnas, fila))
tuplas_seleccionadas.append(tupla)
# Crear un conjunto con los IDs [0] de los customer_inovices que debo liberar en FactuGES
# invoices_to_verifactu = {str(fila[0]) for fila in filas}
logging.info(f"Customer invoices rows to be send Verifactu: {
len(tuplas_seleccionadas)}")
logging.info(
f"Customer invoices rows to be send Verifactu: {len(tuplas_seleccionadas)}")
# Verificar si hay filas en el resultado
if tuplas_seleccionadas:
enviar_datos(tuplas_seleccionadas, config)
enviar_datos(tuplas_seleccionadas, cursor_mysql, config)
logging.info(f"Ha ido bien enviar_datos")
else:
logging.info(f"There are no rows to send")
except Exception as e:
except Exception as error:
if cursor_mysql is not None:
cursor_mysql.close()
logging.error(f"(ERROR) Failed to fetch from database:{
config['UECKO_MYSQL_DATABASE']} - using user:{config['UECKO_MYSQL_USER']}")
logging.error(e)
raise e
logging.error(
f"(ERROR) Failed to fetch from database:{config['UECKO_MYSQL_DATABASE']} - using user:{config['UECKO_MYSQL_USER']}")
logging.error(error)
raise error
def enviar_datos(invoices_to_verifactu, config):
def enviar_datos(invoices_to_verifactu, cursor_mysql, config):
# Recorrer todas las facturas para crear json de envio
try:
logging.info(f"Send to Verifactu")
invoice_id = None
factura = None
for fila in invoices_to_verifactu:
factura = {
# REQUERIDOS
"id": str(fila['id']),
"serie": fila['series'],
"numero": fila['invoice_number'],
"fecha_expedicion": fila['invoice_date'].strftime("%d-%m-%Y"),
# F1: Factura (Art. 6, 7.2 Y 7.3 del RD 1619/2012)
"tipo_factura": "F1",
"descripcion": fila['description'],
"nif": fila['customer_tin']
# "tin": fila[5],
# "name": fila[6]
# },
# "totals": {
# "total_amount_value": float(fila[7]) if isinstance(fila[7], Decimal) else fila[7],
# "taxable_amount_value": float(fila[9]) if fila[9] is not None else 0,
# "taxes_amount_value": float(fila[10]) if fila[10] is not None else 0,
# },
# "tax_code": fila[8]
}
logging.info(f"Send to Verifactu: {factura}")
# Si los ids de factura anterior y actual no coinciden o empezamos factura nueva, miramos si ya existe una factura si es así la mandamos AEAT
# y creamos la cabecera de la factura siguiente, si no existe factura solo la creamos
if invoice_id != str(fila['id']):
procesar_factura_verifactu(factura, cursor_mysql, config)
# preparamos nueva factura
ok, respuesta = preparar_factura(fila)
if not ok:
logging.info(
f">>> Factura {fila['reference']} no cumple requisitos para ser mandada a Verifactu:")
logging.info(
f">>>>>> Faltan campos requeridos: {respuesta}")
factura = None
continue
factura = respuesta
# Validamos que el cif de la factura exista en la AEAT si no es así no se hace el envío
if not validar_nif(factura.get('nif'), factura.get('nombre'), config):
logging.info(
f">>> Factura {factura.get('reference')} no cumple requisitos para ser mandada a Verifactu:")
logging.info(
f">>>>>> El cif de la factura no existe en AEAT: {factura.get('nif')}")
continue
ok, linea = preparar_linea(fila)
if not ok:
logging.info(
f">>> Factura {factura.get('reference')} no cumple requisitos para ser mandada a Verifactu:")
logging.info(f">>>>>> Faltan campos requeridos: {linea}")
factura = None
else:
factura["lineas"].append(linea)
procesar_factura_verifactu(factura, cursor_mysql, config)
except Exception as e:
# Escribir el error en el archivo de errores
@ -107,261 +126,130 @@ def enviar_datos(invoices_to_verifactu, config):
raise e # Re-lanzar la excepción para detener el procesamiento
def insertar_datos(conn_mysql, filas, conn_factuges, config):
cte_company_id = '5e4dc5b3-96b9-4968-9490-14bd032fec5f'
insert_customer_query = (
"INSERT INTO customers (id, name, tin, street, city, province, postal_code, country, phone_primary, phone_secondary, mobile_primary, mobile_secondary, "
"email_primary, email_secondary, website, factuges_id, company_id, is_company, language_code, currency_code, status, created_at, updated_at ) "
"VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 1, 'es', 'EUR', 'active', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) "
)
def procesar_factura_verifactu(
factura: Optional[Dict[str, Any]],
cursor_mysql,
config: Dict[str, Any]
) -> bool:
insert_payment_methods_query = (
"INSERT INTO payment_methods (id, description, factuges_id, created_at, updated_at ) "
"VALUES (%s, %s, %s, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) "
)
insert_verifactu_records = ("INSERT INTO verifactu_records (id, invoice_id, estado, uuid, url, qr, created_at, updated_at) "
"VALUES (%s, %s, %s, %s, %s, %s, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) "
)
insert_customer_invoices_query = (
"INSERT INTO customer_invoices (id, company_id, status, series, reference, invoice_date, operation_date, "
"subtotal_amount_value, discount_amount_value, discount_percentage_value, taxable_amount_value, taxes_amount_value, total_amount_value, "
"customer_id, customer_tin, customer_name, customer_street, customer_city, customer_province, customer_postal_code, customer_country, "
"payment_method_id, payment_method_description, "
"subtotal_amount_scale, discount_amount_scale, discount_percentage_scale, taxable_amount_scale, taxes_amount_scale, total_amount_scale, "
"language_code, currency_code, created_at, updated_at) "
"VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 2, 2, 2, 2, 2, 2, 'es', 'EUR', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)"
)
update_verifactu_records = ("UPDATE verifactu_records "
"set estado = %s,"
"operacion = %s, "
"updated_at = CURRENT_TIMESTAMP "
"WHERE uuid = %s "
)
insert_customer_invoices_taxes_query = (
"INSERT INTO customer_invoice_taxes (tax_id, invoice_id, tax_code, taxable_amount_value, taxes_amount_value, taxable_amount_scale, taxes_amount_scale, "
"created_at, updated_at) "
"VALUES (%s, %s, %s, %s, %s, 2, 2, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)"
)
insert_customer_invoice_items_query = (
"INSERT INTO customer_invoice_items (item_id, invoice_id, position, description, quantity_value, unit_amount_value, discount_percentage_value, discount_amount_value, total_amount_value, "
"quantity_scale, unit_amount_scale, discount_amount_scale, total_amount_scale, discount_percentage_scale, created_at, updated_at) "
"VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, 2, 4, 2, 4, 2, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)"
)
insert_customer_invoice_item_taxes_query = (
"INSERT INTO customer_invoice_item_taxes (tax_id, item_id, tax_code, taxable_amount_value, taxes_amount_value, taxable_amount_scale, taxes_amount_scale, "
"created_at, updated_at) "
"VALUES (%s, %s, %s, %s, %s, 4, 2, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)"
)
update_FACTURAS_CLIENTE_query = (
"UPDATE FACTURAS_CLIENTE set ID_VERIFACTU = ? WHERE ID = ?")
select_customer_query = (
"SELECT customers.id "
"FROM customers "
"WHERE customers.factuges_id = %s"
)
select_payment_method_query = (
"SELECT payment_methods.id "
"FROM payment_methods "
"WHERE payment_methods.factuges_id = %s"
)
cursorMySQL = None
cursor_FactuGES = None
factuges_id_anterior = None
id_customer_invoice = None
num_fac_procesed = 0
try:
cursorMySQL = conn_mysql.cursor()
cursor_FactuGES = conn_factuges.cursor()
# Insertar datos en la tabla 'customer_invoices'
for factura_detalle in filas:
factuges_id = int(factura_detalle['ID_FACTURA'])
invoice_status = str('draft')
invoice_series = str('A')
invoice_number = str(factura_detalle['REFERENCIA'])
invoice_date = str(factura_detalle['FECHA_FACTURA'])
operation_date = str(factura_detalle['FECHA_FACTURA'])
# siempre tendrán 2 decimales
subtotal_amount_value = (factura_detalle['IMPORTE_NETO'] or 0)*100
discount_amount_value = (
factura_detalle['IMPORTE_DESCUENTO'] or 0)*100
discount_percentage_value = (
factura_detalle['DESCUENTO'] or 0)*100 if (factura_detalle['DESCUENTO']) is not None else 0
taxable_amount_value = (factura_detalle['BASE_IMPONIBLE'] or 0)*100
# Preparamos el tipo de IVA, en FactuGES es único
tax_code = str(factura_detalle['DES_TIPO_IVA'])
if tax_code == 'IVA21':
tax_code = 'iva_21'
elif tax_code == 'IVA18':
tax_code = 'iva_18'
elif tax_code == 'IVA16':
tax_code = 'iva_16'
elif tax_code == 'IVA10':
tax_code = 'iva_10'
else:
tax_code = ''
# La cuota de impuestos es el IVA + RE
tax_amount_value = (
(factura_detalle['IMPORTE_IVA'] or 0) + (factura_detalle['IMPORTE_RE'] or 0))*100
total_amount_value = (factura_detalle['IMPORTE_TOTAL'] or 0)*100
payment_method_id = str(uuid4())
factuges_payment_method_id = str(factura_detalle['ID_FORMA_PAGO'])
payment_method_description = str(factura_detalle['DES_FORMA_PAGO'])
customer_id = str(uuid4())
factuges_customer_id = str(factura_detalle['ID_CLIENTE'])
customer_tin = str(factura_detalle['NIF_CIF'])
customer_name = str(factura_detalle['NOMBRE'])
customer_street = str(factura_detalle['CALLE'])
customer_city = str(factura_detalle['POBLACION'])
customer_province = str(factura_detalle['PROVINCIA'])
customer_postal_code = str(factura_detalle['CODIGO_POSTAL'])
customer_phone_primary = factura_detalle['TELEFONO_1']
customer_phone_secondary = factura_detalle['TELEFONO_2']
customer_mobile_primary = factura_detalle['MOVIL_1']
customer_mobile_secondary = factura_detalle['MOVIL_2']
customer_email_primary = factura_detalle['EMAIL_1']
customer_email_secondary = factura_detalle['EMAIL_2']
customer_webside = str(factura_detalle['PAGINA_WEB'])
customer_country = 'es'
item_position = int(factura_detalle['POSICION'])
item_description = str(factura_detalle['CONCEPTO'])
item_quantity_value = None if factura_detalle['CANTIDAD'] is None else (
factura_detalle['CANTIDAD'] or 0)*100
item_unit_amount_value = None if factura_detalle['IMPORTE_UNIDAD'] is None else (
factura_detalle['IMPORTE_UNIDAD'] or 0)*10000
Descuento = factura_detalle['DESCUENTO']
item_discount_percentage_value = None if Descuento is None else None if Descuento == 0 else (
factura_detalle['DESCUENTO'])*100
item_discount_amount = None
# 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
item_total_amount = (factura_detalle['IMPORTE_TOTAL_DET'] or 0)*100
# None if factura_detalle['IMPORTE_TOTAL_DET'] is None else (
# factura_detalle['IMPORTE_TOTAL_DET'] or 0)*100
# campos pendiente de revisar en un futuro
# xxxxxxx = str(factura_detalle['ID_EMPRESA'])
# xxxxxxx = str(factura_detalle['ID_FORMA_PAGO']) según este id se debe de guardar en la factura los vencimiento asociados a la forma de pago
# xxxxxxx = str(factura_detalle['OBSERVACIONES'])
if factuges_id_anterior is None or factuges_id_anterior != factuges_id:
# Comprobamos si existe el cliente del primer item de la factura
cursorMySQL.execute(select_customer_query,
(factuges_customer_id, ))
row = cursorMySQL.fetchone()
is_new = (row is None) or (row[0] is None)
if is_new:
logging.info(
f"Inserting customer {factuges_customer_id} {customer_tin} {customer_name}")
cursorMySQL.execute(insert_customer_query, (customer_id, customer_name, customer_tin, customer_street, customer_city, customer_province,
customer_postal_code, customer_country, customer_phone_primary, customer_phone_secondary, customer_mobile_primary,
customer_mobile_secondary, customer_email_primary, customer_email_secondary, customer_webside, factuges_customer_id, cte_company_id))
else:
# Si ya exite ponemos el id del customer correspondiente
customer_id = str(row[0])
logging.info(
f"Updating customer {factuges_customer_id} {customer_id}")
# cursorMySQL.execute(update_customer_query, .....)
# Comprobamos si existe la forma de pago del primer item de la factura
cursorMySQL.execute(select_payment_method_query,
(factuges_payment_method_id, ))
row = cursorMySQL.fetchone()
is_new = (row is None) or (row[0] is None)
if is_new:
logging.info(
f"Inserting cuspayment method {factuges_payment_method_id} {payment_method_id} {payment_method_description}")
cursorMySQL.execute(insert_payment_methods_query, (
payment_method_id, payment_method_description, factuges_payment_method_id))
else:
# Si ya exite ponemos el id del customer correspondiente
payment_method_id = str(row[0])
logging.info(
f"Updating customer {factuges_payment_method_id} {payment_method_id}")
# cursorMySQL.execute(update_customer_query, .....)
# Insertamos cabecera de la factura
# Generar un ID único para la tabla customer_invoices
id_customer_invoice = str(uuid4())
if factura != None:
# Creamos registro de factura en verifactu
if factura.get('uuid') is None:
# logging.info(f"Send to create Verifactu: {factura}")
respuesta = crear_factura(factura, config)
if respuesta.get("status") == 200 and respuesta.get("ok"):
data = respuesta.get("data")
cursor_mysql.execute(insert_verifactu_records, (str(uuid4()), factura.get(
"id"), data.get("estado"), data.get("uuid"), data.get("url"), data.get("qr")))
logging.info(
f"Inserting customer_invoice {id_customer_invoice} {invoice_number} {invoice_date}")
cursorMySQL.execute(insert_customer_invoices_query, (id_customer_invoice, cte_company_id, invoice_status, invoice_series, invoice_number, invoice_date, operation_date,
subtotal_amount_value, discount_amount_value, discount_percentage_value, taxable_amount_value, tax_amount_value, total_amount_value,
customer_id, customer_tin, customer_name, customer_street, customer_city, customer_province, customer_postal_code, customer_country,
payment_method_id, payment_method_description))
# Insertamos el IVA y RE si viene
if (factura_detalle['IVA'] > 0):
taxable_amount_value = (
factura_detalle['BASE_IMPONIBLE'])*100
tax_amount_value = (factura_detalle['IMPORTE_IVA'])*100
cursorMySQL.execute(insert_customer_invoices_taxes_query, (str(uuid4()),
id_customer_invoice, tax_code, taxable_amount_value, tax_amount_value))
if (factura_detalle['RECARGO_EQUIVALENCIA'] > 0):
tax_code = 're_5_2'
taxable_amount_value = (
factura_detalle['BASE_IMPONIBLE'])*100
tax_amount_value = (factura_detalle['IMPORTE_RE'])*100
cursorMySQL.execute(insert_customer_invoices_taxes_query, (str(uuid4()),
id_customer_invoice, tax_code, taxable_amount_value, tax_amount_value))
# Guardamos en Factuges el id de la customer_invoice
logging.info(
f"Updating FACTURAS_CLIENTE {id_customer_invoice} {factuges_id}")
cursor_FactuGES.execute(
update_FACTURAS_CLIENTE_query, (id_customer_invoice, factuges_id))
num_fac_procesed += 1
# Insertamos detalles y taxes correspondientes siempre
# Generar un ID único para la tabla customer_invoice_items
item_id = str(uuid4())
logging.info(
f"Inserting customer_invoice_items {id_customer_invoice} {item_position} {item_quantity_value}")
cursorMySQL.execute(insert_customer_invoice_items_query, (item_id, id_customer_invoice, item_position, item_description,
item_quantity_value, item_unit_amount_value, item_discount_percentage_value, item_discount_amount, item_total_amount))
if tax_code == 'IVA21':
tax_amount_value = (
(factura_detalle['IMPORTE_TOTAL'] or 0)*Decimal(0.21))*100
elif tax_code == 'IVA18':
tax_amount_value = (
(factura_detalle['IMPORTE_TOTAL'] or 0)*Decimal(0.18))*100
elif tax_code == 'IVA16':
tax_amount_value = (
(factura_detalle['IMPORTE_TOTAL'] or 0)*Decimal(0.16))*100
elif tax_code == 'IVA10':
tax_amount_value = (
(factura_detalle['IMPORTE_TOTAL'] or 0)*Decimal(0.10))*100
f">>> Factura {factura.get("reference")} registrada en Verifactu")
return True
else:
tax_amount_value = (factura_detalle['IMPORTE_TOTAL'] or 0)*100
logging.info(
f">>> Factura {factura.get("reference")} enviada a Verifactu con error {respuesta}")
return False
# Actualizamos registro de factura en verifactu
else:
# logging.info(f"Send to update Verifactu: {factura}")
respuesta = estado_factura(factura.get('uuid'), config)
if respuesta.get("status") == 200 and respuesta.get("ok"):
data = respuesta.get("data")
cursor_mysql.execute(update_verifactu_records, (data.get(
'estado'), data.get('operacion'), factura.get('uuid')))
logging.info(
f">>> Factura {factura.get("reference")} actualizado registro de Verifactu")
return True
else:
return False
logging.info(
f"Inserting customer_invoice_item_taxes {item_id} {item_position} {tax_code} {item_total_amount} {tax_amount_value}")
cursorMySQL.execute(insert_customer_invoice_item_taxes_query, (str(uuid4()), item_id, tax_code,
item_total_amount, tax_amount_value))
# Asignamos el id factura anterior para no volver a inserta cabecera
factuges_id_anterior = factuges_id
def preparar_factura(fila: Dict[str, Any]) -> Tuple[bool, Dict[str, Any] | list]:
"""
Prepara el JSON de factura para Verifactu a partir de 'fila'.
"""
campos_requeridos = ("series", "invoice_number",
"invoice_date", "description", "total_amount_value")
ok, missing = validar_requeridos(fila, campos_requeridos)
if not ok:
return False, missing
logging.info(
f"FACTURAS_CLIENTE rows to be processed: {str(num_fac_procesed)}")
factura = {
"nif": fila['customer_tin'],
"nombre": fila['customer_name'],
"serie": fila['series'],
"numero": fila['invoice_number'],
# convertimos la fecha al formato requerido en el api
"fecha_expedicion": fila['invoice_date'].strftime("%d-%m-%Y"),
# F1: Factura (Art. 6, 7.2 Y 7.3 del RD 1619/2012)
"tipo_factura": "F1",
"descripcion": fila['description'],
# desescalamos el importe para dar su importe real
"importe_total": unscale_to_str(str(fila['total_amount_value']), str(fila['total_amount_scale'])),
"lineas": [],
except Exception as e:
# Escribir el error en el archivo de errores
logging.error(str(e))
raise e # Re-lanzar la excepción para detener el procesamiento
# CAMPOS PARA LOGICA NUESTRA
"id": str(fila['id']),
"reference": str(fila['reference']),
"uuid": fila['uuid'],
}
finally:
# Cerrar la conexión
if cursorMySQL is not None:
cursorMySQL.close()
if cursor_FactuGES is not None:
cursor_FactuGES.close()
return True, factura
def preparar_linea(fila: Dict[str, Any]) -> Tuple[bool, Dict[str, Any] | list]:
"""
Prepara el JSON de línea para Verifactu a partir de 'fila'.
"""
campos_requeridos = ("taxable_amount_value",
"taxable_amount_scale")
ok, missing = validar_requeridos(fila, campos_requeridos)
if not ok:
return False, missing
catalog = TaxCatalog.create()
base_imponible = unscale_to_str(
str(fila['taxable_amount_value']), str(fila['taxable_amount_scale']))
# Si el tipo impositivo es exento
if catalog.is_non_exempt(fila['tax_code']):
calificacion_operacion = "S1"
# FALTA COMPROBAR SI IMPUESTO IGIC (03) o IPSI (02)
impuesto = "01"
tipo_impositivo = str(catalog.get_percent_reduced(fila['tax_code']))
cuota_repercutida = unscale_to_str(
str(fila['taxes_amount_value']), str(fila['taxes_amount_scale']))
linea = {
"base_imponible": base_imponible,
"impuesto": impuesto,
"tipo_impositivo": tipo_impositivo,
"cuota_repercutida": cuota_repercutida,
}
else:
# FALTA REVISAR DIFERENCIAS ENTRE E2,E3,E4...
operacion_exenta = "E1"
linea = {
"base_imponible": base_imponible,
"operacion_exenta": operacion_exenta,
}
return True, linea
def validar_requeridos(fila: Dict[str, Any], campos: Iterable[str]) -> Tuple[bool, List[str]]:
missing = [k for k in campos if fila.get(k) is None]
return (len(missing) == 0), missing

216
app/db/sync_orders.py
View File

@ -1,216 +0,0 @@
import logging
from decimal import Decimal
from config import load_config
from utils import text_converter
def sync_orders(conn_factuges, conn_mysql):
config = load_config()
consulta_quotes_uecko = (
"SELECT quotes.id, quotes.date_sent, quotes.reference, quotes.customer_reference, "
"quotes.customer_information, quotes.dealer_id, dealers.id_contact, dealers.name "
"FROM quotes INNER JOIN dealers ON (dealers.id = quotes.dealer_id) "
"WHERE quotes.date_sent IS NOT NULL AND "
"quotes.id_contract IS NULL"
)
update_quotes_uecko = (
"UPDATE quotes SET "
"id_contract = %s "
"WHERE id = %s"
)
inserted_orders = []
cursor_MySQL = None
try:
cursor_MySQL = conn_mysql.cursor()
cursor_MySQL.execute(consulta_quotes_uecko)
quotes = cursor_MySQL.fetchall()
quote_columns = [desc[0] for desc in cursor_MySQL.description]
selected_quotes = []
for quote in quotes:
tupla = dict(zip(quote_columns, quote))
selected_quotes.append(tupla)
logging.info(f"Quotes rows to be processed: {len(selected_quotes)}")
if selected_quotes:
for quote in selected_quotes:
logging.info(f"Quote reference: {quote['reference']}")
if (quote['id_contact'] is None):
logging.info(
f"Error: Quote unprocesable (id_contact missing)")
continue
items = fetch_quote_items(conn_mysql, quote['id'])
id_contrato = insert_quote_to_factuges(
conn_factuges, quote, items, config)
cursor_MySQL.execute(update_quotes_uecko,
(int(id_contrato), str(quote['id'])))
inserted_orders.append({
"customer_reference": quote['customer_reference'],
"dealer_name": quote['name'],
})
cursor_MySQL.close()
return inserted_orders
except Exception as e:
# Escribir el error en el archivo de errores
logging.error(msg=e, stack_info=True)
raise e # Re-lanzar la excepción para detener el procesamiento
finally:
# Cerrar la conexión
if cursor_MySQL is not None:
cursor_MySQL.close()
def fetch_quote_items(conn_mysql, quote_id):
consulta_quotes_items_uecko = (
"SELECT quote_items.item_id, quote_items.id_article, quote_items.position, "
"quote_items.description, quote_items.quantity, quote_items.unit_price, "
"quote_items.discount, quote_items.total_price "
"FROM quote_items "
"WHERE quote_items.quote_id = %s "
"ORDER BY quote_items.position"
)
cursor_MySQL = None
try:
cursor_MySQL = conn_mysql.cursor()
cursor_MySQL.execute(consulta_quotes_items_uecko, (quote_id, ))
items = cursor_MySQL.fetchall()
items_columns = [desc[0] for desc in cursor_MySQL.description]
cursor_MySQL.close()
selected_items = []
for item in items:
tupla = dict(zip(items_columns, item))
selected_items.append(tupla)
return selected_items
except Exception as e:
# Escribir el error en el archivo de errores
logging.error(msg=e, stack_info=True)
raise e # Re-lanzar la excepción para detener el procesamiento
finally:
if cursor_MySQL is not None:
cursor_MySQL.close()
def insert_quote_to_factuges(conn_factuges, quote, items, config):
id_empresa = int(config['FACTUGES_ID_EMPRESA'])
situacion = str(config['FACTUGES_CONTRATO_SITUACION'])
id_tienda = int(config['FACTUGES_CONTRATO_ID_TIENDA'])
enviada_revisada = int(config['FACTUGES_CONTRATO_ENVIADA_REVISADA'])
id_cliente = int(quote['id_contact'])
# nombre_clliente = str(quote['name'])
fecha_presupuesto = quote['date_sent'].date()
persona_contacto = str(quote['customer_information'])
referencia_cliente = str(quote['customer_reference'])
select_gen_id_contrato_cliente = (
"select GEN_ID(GEN_CONTRATOS_CLI_ID, 1) from RDB$DATABASE"
)
select_gen_id_presupuesto_cliente = (
"select GEN_ID(GEN_PRESUPUESTOS_CLI_ID, 1) from RDB$DATABASE"
)
insert_contrato_cliente_data = (
"insert into CONTRATOS_CLIENTE ("
"ID, ID_EMPRESA, ID_TIENDA, ID_CLIENTE, NOMBRE, SITUACION, "
"NOTAS_ENVIO, REFERENCIA_CLIENTE, "
"ENVIADA_REVISADA, FECHA_CONTRATO "
") values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
)
insert_presupuesto_cliente_data = (
"insert into PRESUPUESTOS_CLIENTE ("
"ID, ID_EMPRESA, ID_TIENDA, ID_CLIENTE, SITUACION, "
"OBSERVACIONES, REFERENCIA_CLIENTE, "
"ENVIADA_REVISADA, FECHA_PRESUPUESTO "
") values (?, ?, ?, ?, ?, ?, ?, ?, ?)"
)
insert_contrato_cliente_detalles_data = (
"insert into CONTRATOS_CLIENTE_DETALLES ("
"ID, ID_CONTRATO, POSICION, ID_ARTICULO, TIPO_DETALLE, "
"CONCEPTO, CANTIDAD, IMPORTE_UNIDAD, "
"VALORADO, VISIBLE, FECHA_ALTA "
") values ("
"GEN_ID(GEN_CONTRATOS_CLI_DETALLE_ID, 1), ?, ?, ?, ?, "
"?, ?, ?, "
"1, 1, CURRENT_TIMESTAMP"
")"
)
insert_presupuesto_cliente_detalles_data = (
"insert into PRESUPUESTOS_CLIENTE_DETALLES ("
"ID, ID_PRESUPUESTO, POSICION, ID_ARTICULO, TIPO_DETALLE, "
"CONCEPTO, CANTIDAD, IMPORTE_UNIDAD, "
"VALORADO, VISIBLE, FECHA_ALTA "
") values ("
"GEN_ID(GEN_PRESUPUESTOS_CLI_DETALLE_ID, 1), ?, ?, ?, ?, "
"?, ?, ?, "
"1, 1, CURRENT_TIMESTAMP"
")"
)
cursor_FactuGES = None
try:
cursor_FactuGES = conn_factuges.cursor()
cursor_FactuGES.execute(select_gen_id_presupuesto_cliente)
id_presupuesto = int(cursor_FactuGES.fetchone()[0])
logging.info(
f"Inserting quote on FactuGES -> id_preupuesto = {str(id_presupuesto)}")
logging.info(insert_presupuesto_cliente_data)
logging.info((id_presupuesto, id_empresa, id_tienda, id_cliente,
situacion, fecha_presupuesto, persona_contacto,
referencia_cliente, enviada_revisada, fecha_presupuesto))
cursor_FactuGES.execute(insert_presupuesto_cliente_data,
(id_presupuesto, id_empresa, id_tienda, id_cliente,
situacion, persona_contacto,
referencia_cliente, enviada_revisada, fecha_presupuesto))
logging.info(
f"Inserting items. Quote items length to be processed: {len(items)}")
for item in items:
descripcion_iso = text_converter(
item['description'], charset_destino='ISO8859_1', longitud_maxima=2000)
quantity = Decimal(
int(item['quantity'])) / Decimal(100) if item['quantity'] is not None else None
unit_price = Decimal(int(
item['unit_price'])) / Decimal(100) if item['unit_price'] is not None else None
# total_price = item['total_price']
logging.info(str(insert_presupuesto_cliente_detalles_data))
logging.info((
id_presupuesto, item['position'], item['id_article'], config['FACTUGES_CONTRATO_TIPO_DETALLE'],
descripcion_iso, quantity, unit_price
))
cursor_FactuGES.execute(insert_presupuesto_cliente_detalles_data, (
id_presupuesto, item['position'], item['id_article'], config['FACTUGES_CONTRATO_TIPO_DETALLE'],
descripcion_iso, quantity, unit_price
))
cursor_FactuGES.close()
return id_presupuesto
except Exception as e:
# Escribir el error en el archivo de errores
logging.error(msg=e, stack_info=True)
raise e # Re-lanzar la excepción para detener el procesamiento
finally:
if cursor_FactuGES is not None:
cursor_FactuGES.close()

27
app/main.py
View File

@ -6,7 +6,7 @@ from datetime import datetime
from dateutil import tz
from config import setup_logging, load_config
from db import get_mysql_connection, get_factuges_connection, sync_invoices, sync_invoices_verifactu
from utils import obtener_fecha_ultima_ejecucion, actualizar_fecha_ultima_ejecucion, log_system_metrics, send_orders_mail
from utils import obtener_fecha_ultima_ejecucion, actualizar_fecha_ultima_ejecucion, log_system_metrics, send_orders_mail, limpiar_cadena
def main():
@ -37,22 +37,33 @@ def main():
logging.info("Last execution (Local time): %s",
last_execution_date_local_tz)
# Abrimos conexiones con una única transacción para que todo esté controlado
conn_factuges = get_factuges_connection(config)
conn_mysql = get_mysql_connection(config)
# Sync invoices
logging.info(f"Sync invoices")
logging.info(f">>>>>>>>>>> Sync invoices FactuGES escritorio to FactuGES web")
sync_invoices(conn_factuges, conn_mysql, last_execution_date_local_tz)
# Sync Verifactu
logging.info(f"Sync Verifactu")
sync_invoices_verifactu(conn_mysql, last_execution_date_local_tz)
# actualizar_fecha_ultima_ejecucion()
# Confirmar los cambios
conn_mysql.commit()
conn_factuges.commit()
conn_factuges.close()
conn_mysql.close()
logging.info(f"FIN Sync FactuGES web >>>>>>>>>>")
# ESTO OTRO DEBERIA SER OTRA TRANSACCION POR LO QUE HACEMOS NUEVA CONEXION
# Vamos que deberia ir en otro lado
conn_mysql = get_mysql_connection(config)
# Sync Verifactu
logging.info(f">>>>>>>>>> Sync facturas emitidas en FactuGES web to Verifactu")
sync_invoices_verifactu(conn_mysql, last_execution_date_local_tz)
conn_mysql.commit()
conn_mysql.close()
logging.info(f"FIN Sync Verifactu >>>>>>>>>>")
# actualizar_fecha_ultima_ejecucion()
# Enviar email
# send_orders_mail(inserted_orders)

9
app/utils/__init__.py
View File

@ -2,5 +2,10 @@ from .last_execution_helper import actualizar_fecha_ultima_ejecucion, obtener_fe
from .log_system_metrics import log_system_metrics
from .password import hashPassword
from .send_orders_mail import send_orders_mail
from .text_converter import text_converter
from .send_rest_api import send_rest_api
from .text_converter import text_converter, limpiar_cadena
from .send_rest_api import validar_nif, estado_factura, crear_factura
from .tax_catalog_helper import TaxCatalog, get_default_tax_catalog
from .importes_helper import unscale_to_str, unscale_to_decimal
from .telefonos_helper import normalizar_telefono_con_plus
from .mails_helper import corregir_y_validar_email
from .websites_helper import normalizar_url_para_insert

37
app/utils/importes_helper.py Normal file
View File

@ -0,0 +1,37 @@
from decimal import Decimal, ROUND_HALF_UP
from typing import Any, Optional
def unscale_to_decimal(value: Any, scale: Any) -> Decimal:
"""
Convierte un valor escalado (p. ej. 24200 con scale=2) a su valor en unidades (242).
No redondea; sólo mueve el punto decimal.
"""
if value is None:
return Decimal("0")
d = Decimal(str(value))
s = int(scale or 0)
return d.scaleb(-s) # divide por 10**s
def unscale_to_str(
value: Any,
scale: Any,
*,
decimals: Optional[int] = None,
strip_trailing_zeros: bool = True
) -> str:
"""
Igual que unscale_to_decimal, pero devuelve str.
- decimals: fija nº de decimales (p. ej. 2). Si None, no fuerza decimales.
- strip_trailing_zeros: si True, quita ceros y el punto sobrantes.
"""
d = unscale_to_decimal(value, scale)
if decimals is not None:
q = Decimal("1").scaleb(-decimals) # p.ej. 2 -> Decimal('0.01')
d = d.quantize(q, rounding=ROUND_HALF_UP)
s = format(d, "f")
if strip_trailing_zeros and "." in s:
s = s.rstrip("0").rstrip(".")
return s

53
app/utils/mails_helper.py Normal file
View File

@ -0,0 +1,53 @@
import re
from typing import Any, Optional, Tuple
# Regex práctico (no RFC completo) para validar emails
_EMAIL_RE = re.compile(
r"""^(?=.{3,254}$) # longitud total razonable
(?=.{1,64}@) # parte local máx. 64
[A-Za-z0-9!#$%&'*+/=?^_`{|}~-]+
(?:\.[A-Za-z0-9!#$%&'*+/=?^_`{|}~-]+)* # puntos en la parte local
@
(?:[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?\.)+ # labels dominio
[A-Za-z]{2,63}$ # TLD
""",
re.X,
)
def corregir_y_validar_email(texto: Any) -> Tuple[bool, Optional[str]]:
"""
Normaliza y valida un email.
Correcciones aplicadas:
- Reemplaza comas por puntos.
- Elimina espacios alrededor y dentro (p. ej. 'a @ b . com' -> 'a@b.com').
- Convierte el dominio a minúsculas.
- Colapsa puntos consecutivos en el dominio ('..' -> '.').
Devuelve (es_valido, email_corregido | None).
"""
if texto is None:
return False, None
s = str(texto).strip()
# 1) comas -> puntos
s = s.replace(",", ".")
# 2) quita espacios
s = re.sub(r"\s+", "", s)
# 3) separar local y dominio (si es posible)
if s.count("@") != 1:
# no intentamos correcciones más agresivas si hay 0 o >1 '@'
return False, s or None
local, domain = s.split("@", 1)
# 4) normalizaciones de dominio
domain = domain.lower()
domain = re.sub(r"\.{2,}", ".", domain) # colapsar puntos dobles
candidato = f"{local}@{domain}"
# 5) validar
es_valido = _EMAIL_RE.match(candidato) is not None
return es_valido, (candidato if candidato else None)

142
app/utils/send_rest_api.py
View File

@ -3,13 +3,120 @@ import logging
from typing import Optional, Dict, Any, Tuple
def estado_factura(uuid_str: str,
config,
) -> Tuple[bool, Optional[Dict[str, Any]], Optional[str]]:
"""
Llama al endpoint de Verifacti para cosultar el estado de registro.
Retorna:
(ok, data, error)
- ok: True si la llamada fue exitosa y resultado == 'IDENTIFICADO'
- data: dict con la respuesta JSON completa
- error: mensaje de error si algo falló
"""
url = config['VERIFACTU_BASE_URL'] + "/verifactu/status"
timeout: int = 10
headers = {"Content-Type": "application/json",
"Accept": "application/json"}
headers["Authorization"] = "Bearer " + config['VERIFACTU_API_KEY']
params = {"uuid": uuid_str}
try:
resp = requests.get(
url, headers=headers, params=params, timeout=timeout)
if resp.status_code == 200:
try:
data = resp.json()
except ValueError:
return {"ok": False, "status": 200, "error": "Respuesta 200 sin JSON válido", "raw": resp.text}
return {"ok": True, "status": 200, "data": data}
if resp.status_code == 400:
try:
body = resp.json()
msg = body.get(
"error") or "Error de validación (400) sin detalle"
except ValueError:
msg = f"Error de validación (400): {resp.text}"
return {"ok": False, "status": 400, "error": msg}
# Otros códigos: devuelve mensaje genérico, intenta extraer JSON si existe
try:
body = resp.json()
msg = body.get("error") or body
return {"ok": False, "status": resp.status_code, "error": str(msg)}
except ValueError:
return {"ok": False, "status": resp.status_code, "error": f"HTTP {resp.status_code}", "raw": resp.text}
except requests.RequestException as e:
logging.error("Error de conexión con la API Verifacti: %s", e)
return False, None, str(e)
except ValueError as e:
logging.error("Respuesta no es JSON válido: %s", e)
return False, None, "Respuesta no es JSON válido"
def crear_factura(payload,
config,
) -> Tuple[bool, Optional[Dict[str, Any]], Optional[str]]:
"""
Llama al endpoint de Verifacti para crear una factura.
Retorna:
(ok, data, error)
- ok: True si la llamada fue exitosa y resultado == 'IDENTIFICADO'
- data: dict con la respuesta JSON completa
- error: mensaje de error si algo falló
"""
url = config['VERIFACTU_BASE_URL'] + "/verifactu/create"
timeout: int = 10
headers = {"Content-Type": "application/json",
"Accept": "application/json"}
headers["Authorization"] = "Bearer " + config['VERIFACTU_API_KEY']
try:
resp = requests.post(
url, json=payload, headers=headers, timeout=timeout)
if resp.status_code == 200:
try:
data = resp.json()
logging.info(data)
except ValueError:
return {"ok": False, "status": 200, "error": "Respuesta 200 sin JSON válido", "raw": resp.text}
return {"ok": True, "status": 200, "data": data}
if resp.status_code == 400:
try:
body = resp.json()
msg = body.get(
"error") or "Error de validación (400) sin detalle"
except ValueError:
msg = f"Error de validación (400): {resp.text}"
return {"ok": False, "status": 400, "error": msg}
# Otros códigos: devuelve mensaje genérico, intenta extraer JSON si existe
try:
body = resp.json()
msg = body.get("error") or body
return {"ok": False, "status": resp.status_code, "error": str(msg)}
except ValueError:
return {"ok": False, "status": resp.status_code, "error": f"HTTP {resp.status_code}", "raw": resp.text}
except requests.RequestException as e:
logging.error("Error de conexión con la API Verifacti: %s", e)
return False, None, str(e)
except ValueError as e:
logging.error("Respuesta no es JSON válido: %s", e)
return False, None, "Respuesta no es JSON válido"
def validar_nif(
nif: str,
nombre: str,
*,
timeout: int = 10,
api_key: Optional[str] = None,
bearer_token: Optional[str] = None,
config,
) -> Tuple[bool, Optional[Dict[str, Any]], Optional[str]]:
"""
Llama al endpoint de Verifacti para validar un NIF.
@ -20,32 +127,27 @@ def validar_nif(
- data: dict con la respuesta JSON completa
- error: mensaje de error si algo falló
"""
url = "https://api.verifacti.com/nifs/validar"
url = config['VERIFACTU_BASE_URL'] + "/nifs/validar"
timeout: int = 10
headers = {"Content-Type": "application/json",
"Accept": "application/json"}
if api_key:
headers["X-API-KEY"] = api_key
if bearer_token:
headers["Authorization"] = f"Bearer {bearer_token}"
payload = {
"nif": nif,
"nombre": nombre,
}
headers["Authorization"] = "Bearer " + config['VERIFACTU_NIFS_API_KEY']
payload = {"nif": nif,
"nombre": nombre,
}
try:
resp = requests.post(
url, json=payload, headers=headers, timeout=timeout)
if resp.status_code != 200:
return False, None, f"HTTP {resp.status_code}: {resp.text}"
logging.info(f"ERRRRRROOOOOOORRRRR LLAMADA REST API")
# return False, None, f"HTTP {resp.status_code}: {resp.text}"
data = resp.json()
resultado = data.get("resultado")
resultado = data.get("resultado", "NO IDENTIFICADO")
logging.info(f"Resultado Verifacti: {resultado}")
# La lógica de validación: 'IDENTIFICADO' = válido
ok = resultado == "IDENTIFICADO"
return ok, data, None
return resultado == "IDENTIFICADO"
except requests.RequestException as e:
logging.error("Error de conexión con la API Verifacti: %s", e)

207
app/utils/tax_catalog_helper.py Normal file
View File

@ -0,0 +1,207 @@
import json
from decimal import Decimal
from typing import Any, Dict, Iterable, Optional, Tuple
from functools import lru_cache
from pathlib import Path
def reduce_scale_pair(value: int, scale: int, *, min_scale: int = 0) -> Tuple[int, int]:
"""
Reduce la escala todo lo posible eliminando ceros finales de 'value',
sin bajar de 'min_scale'. Mantiene exactitud (no redondea).
Ejemplos:
(2100, 2) -> (21, 0)
(750, 2) -> (75, 1)
(400, 2) -> (4, 0)
(275, 2) -> (275, 2) # no se puede reducir (no termina en 0)
"""
v, s = int(value), int(scale)
while s > min_scale and v % 10 == 0:
v //= 10
s -= 1
return v, s
class TaxCatalog:
"""
Carga un catálogo de tipos (IVA, RE, IGIC, etc.) y permite consultar por 'code'.
- get_value_scale(code) -> (value:int, scale:int) | None
- require_value_scale(code) -> (value:int, scale:int) o KeyError si no existe
- get_percent(code) -> Decimal('21.00') # ejemplo para iva_21
- get_fraction(code) -> Decimal('0.21') # 21% como 0.21
"""
DEFAULT_FILENAME = "spain_tax_catalog.json"
def __init__(self, index: Dict[str, Dict[str, Any]]):
self._index = index # code -> info normalizada
# -------- Factorías / constructores --------
@classmethod
def from_file(cls, path: str) -> "TaxCatalog":
with open(path, "r", encoding="utf-8") as f:
items = json.load(f)
return cls(cls._build_index(items))
@classmethod
def from_json(cls, json_str: str) -> "TaxCatalog":
items = json.loads(json_str)
return cls(cls._build_index(items))
@classmethod
def from_iterable(cls, items: Iterable[Dict[str, Any]]) -> "TaxCatalog":
return cls(cls._build_index(items))
@classmethod
def create(cls, filename: Optional[str] = None) -> "TaxCatalog":
"""
Carga automáticamente el JSON del mismo directorio que este archivo.
Uso: catalog = TaxCatalog.create()
"""
fname = filename or cls.DEFAULT_FILENAME
base_dir = Path(__file__).resolve(
).parent if "__file__" in globals() else Path.cwd()
path = base_dir / fname
if not path.exists():
raise FileNotFoundError(f"No se encontró el catálogo en: {path}")
return cls.from_file(str(path))
def reduce_value_scale(
self, code: str, default: Optional[Tuple[int, int]] = None, *, min_scale: int = 0
) -> Optional[Tuple[int, int]]:
"""
Usa get_value_scale(code) y reduce la escala con exactitud.
"""
vs = self.get_value_scale(code)
if vs is None:
return default
value, scale = vs
return reduce_scale_pair(value, scale, min_scale=min_scale)
# -------- API pública --------
def get_percent_reduced(
self, code: str, default: Optional[Decimal] = None
) -> Optional[Decimal]:
"""
Devuelve el porcentaje como Decimal usando la pareja (value, scale)
ya reducida: p. ej., iva_21 -> Decimal('21'), iva_7_5 -> Decimal('7.5').
"""
rs = self.reduce_value_scale(code)
if rs is None:
return default
value_r, scale_r = rs
return Decimal(value_r).scaleb(-scale_r)
def get_value_scale(
self, code: str, default: Optional[Tuple[int, int]] = None
) -> Optional[Tuple[int, int]]:
info = self._index.get(code.lower())
if info is None:
return default
return info["value"], info["scale"]
def require_value_scale(self, code: str) -> Tuple[int, int]:
res = self.get_value_scale(code)
if res is None:
raise KeyError(f"Código no encontrado en catálogo: {code}")
return res
def get_percent(self, code: str, default: Optional[Decimal] = None) -> Optional[Decimal]:
vs = self.get_value_scale(code)
if vs is None:
return default
value, scale = vs
return Decimal(value).scaleb(-scale) # 2100, scale 2 -> 21.00
def get_fraction(self, code: str, default: Optional[Decimal] = None) -> Optional[Decimal]:
pct = self.get_percent(code)
if pct is None:
return default
# 21.00 -> 0.21
return (pct / Decimal(100)).quantize(Decimal("0.0000001")).normalize()
def is_non_exempt(
self,
code: str,
*,
include_zero_rate_as_exempt: bool = False,
default: Optional[bool] = None,
) -> Optional[bool]:
"""
Devuelve True si 'code' NO es exento/asimilado (es decir, se repercute impuesto).
Devuelve False si es exento/asimilado.
Si include_zero_rate_as_exempt=True, los tipos 0% también se tratan como exentos.
Si el code no existe en el catálogo, devuelve 'default'.
"""
key = (code or "").lower()
info = self._index.get(key)
if info is None:
return default
value = info.get("value", 0) # p.ej. 2100
aeat = (info.get("aeat_code") or "").upper() # p.ej. "01"
name = (info.get("name") or "").lower()
code_l = key
# Códigos AEAT típicos de exento/no sujeto/intracomunitario/exportación
# (según tu catálogo de ejemplo)
EXEMPT_AEAT = {"04", "06", "E5", "E6", "E2", "12"}
# Heurística de "exento o asimilado"
exempt_like = (
"exent" in code_l # exenta/exento
or "exent" in name
or "no_sujeto" in code_l
or aeat in EXEMPT_AEAT
)
# Opcional: tratar 0% como exento (iva_0, igic_0...)
if include_zero_rate_as_exempt and int(value) == 0:
exempt_like = True
return not exempt_like
# -------- Internos --------
@staticmethod
def _build_index(items: Iterable[Dict[str, Any]]) -> Dict[str, Dict[str, Any]]:
index: Dict[str, Dict[str, Any]] = {}
for it in items:
code = (it.get("code") or "").lower().strip()
if not code:
continue
raw_value = str(it.get("value", "0")).strip()
raw_scale = str(it.get("scale", "0")).strip()
try:
value_int = int(Decimal(raw_value))
except Exception as e:
raise ValueError(
f"value inválido para code={code}: {raw_value}") from e
try:
scale_int = int(raw_scale)
except Exception as e:
raise ValueError(
f"scale inválido para code={code}: {raw_scale}") from e
index[code] = {
"name": it.get("name"),
"group": it.get("group"),
"description": it.get("description"),
"aeat_code": it.get("aeat_code"),
"value": value_int, # p.ej. 2100
"scale": scale_int, # p.ej. 2
}
return index
# -------- Singleton con caché (opcional, recomendable) --------
@lru_cache(maxsize=1)
def get_default_tax_catalog() -> TaxCatalog:
"""
Devuelve una instancia singleton cargada de 'spain_tax_catalog.json'
en el mismo directorio. Se cachea para evitar relecturas.
"""
return TaxCatalog.create()

19
app/utils/telefonos_helper.py Normal file
View File

@ -0,0 +1,19 @@
import re
from typing import Optional, Any
def normalizar_telefono_con_plus(texto: Any) -> Optional[str]:
"""
Mantiene '+' si está al principio y el resto sólo dígitos.
- ' (+34) 600-123-456 ' -> '+34600123456'
- '620 61 24 91 Tamara' -> '620612491'
- None o sin dígitos -> None
"""
if texto is None:
return None
s = str(texto).strip()
keep_plus = s.startswith("+")
# Quitar todo lo que NO sea dígito
digits = re.sub(r"\D", "", s)
if not digits:
return None
return ("+" if keep_plus else "") + digits

14
app/utils/text_converter.py
View File

@ -1,4 +1,5 @@
import logging
import re
def text_converter(texto, charset_destino='ISO8859_1', longitud_maxima=None):
@ -36,3 +37,16 @@ def text_converter(texto, charset_destino='ISO8859_1', longitud_maxima=None):
except Exception as e:
logging.error(f"Error inesperado al convertir texto: {str(e)}")
return ""
def limpiar_cadena(texto: str) -> str:
"""
Elimina espacios, guiones y cualquier carácter no alfanumérico.
Ejemplos:
'B 83999441' -> 'B83999441'
'B-83999441' -> 'B83999441'
'B_83 99-94.41' -> 'B83999441'
"""
if not isinstance(texto, str):
texto = str(texto)
return re.sub(r'[^A-Za-z0-9]', '', texto)

83
app/utils/websites_helper.py Normal file
View File

@ -0,0 +1,83 @@
import re
from typing import Any, Optional
from urllib.parse import urlsplit, urlunsplit
_SCHEMES = {"http", "https"}
_LABEL_RE = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?$")
_TLD_RE = re.compile(r"^[A-Za-z]{2,63}$")
def _is_valid_host(host: str) -> bool:
if not host or len(host) > 253:
return False
if host.endswith("."):
host = host[:-1]
parts = host.split(".")
if len(parts) < 2:
return False
if not all(_LABEL_RE.match(p) for p in parts):
return False
if not _TLD_RE.match(parts[-1]):
return False
return True
def _normalize_host(host: str) -> str:
host = host.strip().strip(".").lower()
host = re.sub(r"\.{2,}", ".", host) # colapsa '..' -> '.'
return host
def normalizar_url_para_insert(texto: Any) -> Optional[str]:
"""
Devuelve una URL normalizada lista para insertar en BD.
Si no es válida, devuelve None (-> SQL NULL).
"""
if texto is None:
return None
s = str(texto).strip()
if not s:
return None
# Correcciones ligeras
s = s.replace(",", ".")
s = re.sub(r"\s+", "", s)
# Añadir esquema por defecto si falta
candidate = s if "://" in s else f"http://{s}"
sp = urlsplit(candidate)
scheme = (sp.scheme or "http").lower()
netloc, path = sp.netloc, sp.path
# Caso: dominio sin esquema puede quedar en path
if not netloc and path and "." in path and "/" not in path:
netloc, path = path, ""
# Separar userinfo/host:port (no soportamos IPv6 con corchetes aquí)
hostport = netloc.rsplit("@", 1)[-1]
host, port = hostport, ""
if ":" in hostport:
h, p = hostport.rsplit(":", 1)
if h and p.isdigit():
host, port = h, p
host = _normalize_host(host)
# Validaciones
if scheme not in _SCHEMES:
return None
if not _is_valid_host(host):
return None
if port:
try:
pi = int(port)
if not (1 <= pi <= 65535):
return None
except ValueError:
return None
# Reconstrucción netloc (preservando userinfo si existía)
userinfo = netloc[:-len(hostport)] if netloc.endswith(hostport) else ""
new_netloc = f"{userinfo}{host}{(':' + port) if port else ''}"
fixed = urlunsplit((scheme, new_netloc, path or "", sp.query, sp.fragment))
return fixed