287 lines
11 KiB
Python
287 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import os
|
|
import sys
|
|
import re
|
|
|
|
from hdbcli import dbapi
|
|
from dotenv import load_dotenv
|
|
|
|
from constants import TABLAS, ESTADOS_COMPROBANTE, ESTADOS_CONTRIBUYENTE, CONDICIONES_DOMICILIADO
|
|
|
|
# ============================================================================
|
|
# CONFIGURACIÓN
|
|
# ============================================================================
|
|
|
|
# Cargar variables de entorno desde .env
|
|
load_dotenv(dotenv_path=".env")
|
|
|
|
# Configuración de la API
|
|
HANA_HOST = os.getenv("HANA_HOST", "")
|
|
HANA_PORT = int(os.getenv("HANA_PORT", 30015))
|
|
HANA_USER = os.getenv("HANA_USER", "")
|
|
HANA_PASSWORD = os.getenv("HANA_PASSWORD", "")
|
|
|
|
# ============================================================================
|
|
def get_connection():
|
|
"""Crea y retorna una conexión a SAP HANA."""
|
|
try:
|
|
conn = dbapi.connect(
|
|
address=HANA_HOST,
|
|
port=HANA_PORT,
|
|
user=HANA_USER,
|
|
password=HANA_PASSWORD
|
|
)
|
|
return conn
|
|
except Exception as e:
|
|
print(f" ❌ Error conectando a HANA: {e}")
|
|
return None
|
|
|
|
# ============================================================================
|
|
def obtener_documentos_creados_actualizados_hace_60_minutos(schema, minutes=60):
|
|
"""
|
|
Se conecta a la base de datos SAP HANA, realiza una consulta a la tabla OPCH,
|
|
y formatea el resultado para que coincida con el esquema de documentos.json.
|
|
"""
|
|
|
|
query = f"""
|
|
SELECT T0."ObjType", T0."DocEntry", T0."DocNum", T0."CardCode", T0."CardName",
|
|
TO_VARCHAR(T0."DocDate",'DD/MM/YYYY') "DocDate",
|
|
TO_VARCHAR(T0."TaxDate",'DD/MM/YYYY') "TaxDate",
|
|
TO_VARCHAR(T0."CreateDate",'DD/MM/YYYY') "CreateDate",
|
|
T3."LicTradNum" "LicTradNum",
|
|
T1."USER_CODE" "UserCode",
|
|
T0."Indicator", T0."FolioPref", T0."FolioNum",
|
|
CASE
|
|
WHEN IFNULL(T0."DocCur",'SOL') = 'SOL'
|
|
THEN T0."DocTotal" + IFNULL((SELECT SUM(T5."WTAmnt") FROM "{schema}".PCH5 T5 WHERE T5."AbsEntry" = T0."DocEntry" AND T5."Category" = 'I'),0.00)
|
|
ELSE T0."DocTotalFC" + IFNULL((SELECT SUM(T5."WTAmntFC") FROM "{schema}".PCH5 T5 WHERE T5."AbsEntry" = T0."DocEntry" AND T5."Category" = 'I'),0.00)
|
|
END "DocTotal"
|
|
FROM "{schema}".OPCH T0
|
|
INNER JOIN "{schema}".OCRD T3 ON T3."CardCode" = T0."CardCode"
|
|
INNER JOIN "{schema}".OUSR T1 ON T1."USERID" = T0."UserSign2"
|
|
WHERE TO_TIMESTAMP(TO_VARCHAR(T0."UpdateDate", 'YYYY-MM-DD')
|
|
|| ' ' || LPAD(TO_VARCHAR(T0."UpdateTS"), 6, '0'),
|
|
'YYYY-MM-DD HH24MISS')
|
|
>= ADD_SECONDS(CURRENT_TIMESTAMP, {-1 * minutes * 60})
|
|
|
|
AND T0."CANCELED" = 'N'
|
|
AND IFNULL(T0."Indicator",'') IN ('01','02','03','08')
|
|
AND LENGTH(IFNULL(T0."FolioPref",'')) > 0
|
|
AND IFNULL(T0."FolioNum",0) > 0
|
|
AND IFNULL(T3."U_EXX_TIPODOCU",'') IN ('6','4')
|
|
AND (LENGTH(IFNULL(T0."U_PGX_STMENSAJE",'')) = 0 OR LOCATE('NO EXISTE', IFNULL(T0."U_PGX_STMENSAJE",'')) > 0)
|
|
ORDER BY T0."CreateDate" DESC;
|
|
"""
|
|
print(f"🔌 Conectando al servidor SAP HANA ({HANA_HOST}:{HANA_PORT}) para schema {schema}...")
|
|
connection = get_connection()
|
|
if not connection:
|
|
raise ConnectionError("No se pudo establecer la conexión a SAP HANA.")
|
|
|
|
cursor = connection.cursor()
|
|
try:
|
|
cursor.execute(query)
|
|
|
|
# Obtener nombres de columnas
|
|
columns = [column[0] for column in cursor.description]
|
|
|
|
# Convertir resultados a lista de diccionarios
|
|
results = []
|
|
for row in cursor.fetchall():
|
|
results.append(dict(zip(columns, row)))
|
|
|
|
cursor.close()
|
|
return results
|
|
|
|
except Exception as e:
|
|
print(f" ❌ Error en consulta HANA: {e}")
|
|
return []
|
|
finally:
|
|
if cursor:
|
|
cursor.close()
|
|
if connection:
|
|
connection.close()
|
|
print(f" 🔒 [{sys._getframe().f_code.co_name}] Conexión HANA cerrada.")
|
|
|
|
# ============================================================================
|
|
def obtener_documentos_fecha_emision_ultimos_8_dias(schema):
|
|
query = f"""
|
|
SELECT T0."ObjType", T0."DocEntry", T0."DocNum", T0."CardCode", T0."CardName",
|
|
TO_VARCHAR(T0."DocDate",'DD/MM/YYYY') "DocDate",
|
|
TO_VARCHAR(T0."TaxDate",'DD/MM/YYYY') "TaxDate",
|
|
TO_VARCHAR(T0."CreateDate",'DD/MM/YYYY') "CreateDate",
|
|
T3."LicTradNum" "LicTradNum",
|
|
T1."USER_CODE" "UserCode",
|
|
T0."Indicator", T0."FolioPref", T0."FolioNum",
|
|
CASE
|
|
WHEN IFNULL(T0."DocCur",'SOL') = 'SOL'
|
|
THEN T0."DocTotal" + IFNULL((SELECT SUM(T5."WTAmnt") FROM "{schema}".PCH5 T5 WHERE T5."AbsEntry" = T0."DocEntry" AND T5."Category" = 'I'),0.00)
|
|
ELSE T0."DocTotalFC" + IFNULL((SELECT SUM(T5."WTAmntFC") FROM "{schema}".PCH5 T5 WHERE T5."AbsEntry" = T0."DocEntry" AND T5."Category" = 'I'),0.00)
|
|
END "DocTotal"
|
|
FROM "{schema}".OPCH T0
|
|
INNER JOIN "{schema}".OCRD T3 ON T3."CardCode" = T0."CardCode"
|
|
INNER JOIN "{schema}".OUSR T1 ON T1."USERID" = T0."UserSign2"
|
|
WHERE T0."TaxDate" >= ADD_DAYS(CURRENT_DATE, -8)
|
|
AND T0."CANCELED" = 'N'
|
|
AND IFNULL(T0."Indicator",'') IN ('01','02','03','08')
|
|
AND LENGTH(IFNULL(T0."FolioPref",'')) > 0
|
|
AND IFNULL(T0."FolioNum",0) > 0
|
|
AND IFNULL(T3."U_EXX_TIPODOCU",'') IN ('6','4')
|
|
ORDER BY T0."U_PGX_STMENSAJE", T0."CreateDate" DESC;
|
|
"""
|
|
print(f"🔌 Conectando al servidor SAP HANA ({HANA_HOST}:{HANA_PORT}) para schema {schema}...")
|
|
connection = get_connection()
|
|
if not connection:
|
|
raise ConnectionError("No se pudo establecer la conexión a SAP HANA.")
|
|
|
|
cursor = connection.cursor()
|
|
try:
|
|
cursor.execute(query)
|
|
|
|
columns = [column[0] for column in cursor.description]
|
|
|
|
results = []
|
|
for row in cursor.fetchall():
|
|
results.append(dict(zip(columns, row)))
|
|
|
|
cursor.close()
|
|
return results
|
|
|
|
except Exception as e:
|
|
print(f" ❌ Error en consulta HANA: {e}")
|
|
return []
|
|
finally:
|
|
if cursor:
|
|
cursor.close()
|
|
if connection:
|
|
connection.close()
|
|
print(f" 🔒 [{sys._getframe().f_code.co_name}] Conexión HANA cerrada.")
|
|
|
|
# ============================================================================
|
|
def obtener_documentos_creados_hace_x_dias(schema, dias=8):
|
|
query = f"""
|
|
SELECT T0."ObjType", T0."DocEntry", T0."DocNum", T0."CardCode", T0."CardName",
|
|
TO_VARCHAR(T0."DocDate",'DD/MM/YYYY') "DocDate",
|
|
TO_VARCHAR(T0."TaxDate",'DD/MM/YYYY') "TaxDate",
|
|
TO_VARCHAR(T0."CreateDate",'DD/MM/YYYY') "CreateDate",
|
|
T3."LicTradNum" "LicTradNum",
|
|
T1."USER_CODE" "UserCode",
|
|
T0."Indicator", T0."FolioPref", T0."FolioNum",
|
|
CASE
|
|
WHEN IFNULL(T0."DocCur",'SOL') = 'SOL'
|
|
THEN T0."DocTotal" + IFNULL((SELECT SUM(T5."WTAmnt") FROM "{schema}".PCH5 T5 WHERE T5."AbsEntry" = T0."DocEntry" AND T5."Category" = 'I'),0.00)
|
|
ELSE T0."DocTotalFC" + IFNULL((SELECT SUM(T5."WTAmntFC") FROM "{schema}".PCH5 T5 WHERE T5."AbsEntry" = T0."DocEntry" AND T5."Category" = 'I'),0.00)
|
|
END "DocTotal"
|
|
FROM "{schema}".OPCH T0
|
|
INNER JOIN "{schema}".OCRD T3 ON T3."CardCode" = T0."CardCode"
|
|
INNER JOIN "{schema}".OUSR T1 ON T1."USERID" = T0."UserSign2"
|
|
WHERE T0."CreateDate" >= ADD_DAYS(CURRENT_DATE, {-1 * dias})
|
|
AND T0."CANCELED" = 'N'
|
|
AND (LENGTH(IFNULL(T0."U_PGX_STMENSAJE",'')) = 0 OR LOCATE('NO EXISTE', IFNULL(T0."U_PGX_STMENSAJE",'')) > 0)
|
|
AND IFNULL(T0."Indicator",'') IN ('01','02','03','08')
|
|
AND LENGTH(IFNULL(T0."FolioPref",'')) > 0
|
|
AND IFNULL(T0."FolioNum",0) > 0
|
|
AND IFNULL(T3."U_EXX_TIPODOCU",'') IN ('6','4')
|
|
ORDER BY T0."CreateDate" DESC;
|
|
"""
|
|
print(f"🔌 Conectando al servidor SAP HANA ({HANA_HOST}:{HANA_PORT}) para schema {schema}...")
|
|
connection = get_connection()
|
|
if not connection:
|
|
raise ConnectionError("No se pudo establecer la conexión a SAP HANA.")
|
|
|
|
cursor = connection.cursor()
|
|
try:
|
|
cursor.execute(query)
|
|
|
|
columns = [column[0] for column in cursor.description]
|
|
|
|
results = []
|
|
for row in cursor.fetchall():
|
|
results.append(dict(zip(columns, row)))
|
|
|
|
cursor.close()
|
|
return results
|
|
|
|
except Exception as e:
|
|
print(f" ❌ Error en consulta HANA: {e}")
|
|
return []
|
|
finally:
|
|
if cursor:
|
|
cursor.close()
|
|
if connection:
|
|
connection.close()
|
|
print(f" 🔒 [{sys._getframe().f_code.co_name}] Conexión HANA cerrada.")
|
|
|
|
# ============================================================================
|
|
def actualizar_documento_sap(schema, registro,resultado):
|
|
|
|
estado_sunat = "4"
|
|
estado_comprobante = ""
|
|
|
|
if resultado["success"] == True:
|
|
error = "200"
|
|
observaciones = ""
|
|
d = resultado.get("data") or {}
|
|
|
|
if not any(d.values()):
|
|
mensaje = "SIN_DATOS|||"
|
|
estado_comprobante = "0"
|
|
estado_sunat = "2"
|
|
else:
|
|
observaciones = d.get("observaciones", [""])[0] if d.get("observaciones") else ""
|
|
estado_comprobante = d.get("estadoCp", "")
|
|
estado_contribuyente = d.get("estadoRuc", "")
|
|
condicion_domicilio = d.get("condDomiRuc", "")
|
|
|
|
mensaje = "" if estado_comprobante == "" else ESTADOS_COMPROBANTE[estado_comprobante]
|
|
mensaje += "|" if estado_contribuyente == "" else "|" + ESTADOS_CONTRIBUYENTE[estado_contribuyente]
|
|
mensaje += "|" if condicion_domicilio == "" else "|" + CONDICIONES_DOMICILIADO[condicion_domicilio]
|
|
mensaje += "|" + observaciones
|
|
|
|
estado_sunat = "2" if estado_comprobante == "" else ("0" if estado_comprobante == "1" else "2")
|
|
|
|
if resultado["success"] == False:
|
|
error = "400"
|
|
mensaje = resultado["message"]
|
|
|
|
mensaje_limpio = re.sub(r'[^0-9a-zA-Z .|#-]',' ',mensaje)
|
|
|
|
connection = get_connection()
|
|
if not connection:
|
|
raise ConnectionError("No se pudo establecer la conexión a SAP HANA.")
|
|
|
|
cursor = connection.cursor()
|
|
|
|
|
|
# actualizando la tabla SAP
|
|
tabla = TABLAS[registro["objtype"]]
|
|
|
|
query = f"""UPDATE "{registro["companydb"]}".{tabla}
|
|
SET "U_PGX_STMENSAJE" = '{mensaje_limpio}',
|
|
"U_PGX_SSTADO" = '{estado_sunat}',
|
|
"U_PGX_PETOTORFECHA" = CURRENT_DATE,
|
|
"U_PGX_PETOTORHORA" = TO_VARCHAR(CURRENT_TIME,'HH24:MM:SS')
|
|
WHERE "DocEntry" = {registro["fieldvalue"]};"""
|
|
|
|
print(f" ┌─ 🗄️ UPDATE HANA ──────────────────────")
|
|
for line in query.splitlines():
|
|
print(f" │ {line}")
|
|
print(f" └─────────────────────────────────────────")
|
|
|
|
try:
|
|
cursor.execute(query)
|
|
connection.commit()
|
|
cursor.close()
|
|
return True
|
|
except Exception as e:
|
|
print(f" ❌ Error actualizando registro en HANA: {e}")
|
|
return False
|
|
finally:
|
|
if cursor:
|
|
cursor.close()
|
|
if connection:
|
|
connection.close()
|
|
print(f" 🔒 [{sys._getframe().f_code.co_name}] Conexión HANA cerrada.")
|
|
|
|
# ============================================================================
|