75 lines
2.5 KiB
Python
75 lines
2.5 KiB
Python
#!/usr/bin/env python3
|
|
import os
|
|
import sys
|
|
import json
|
|
import argparse
|
|
from datetime import datetime, date
|
|
from hdbcli import dbapi
|
|
from config_loader import CONFIG
|
|
|
|
########################################################
|
|
def get_connection():
|
|
"""Crea y retorna una conexión a SAP HANA."""
|
|
try:
|
|
conn = dbapi.connect(
|
|
address=CONFIG.hana.server,
|
|
port=CONFIG.hana.port,
|
|
user=CONFIG.hana.user,
|
|
password=CONFIG.hana.password
|
|
)
|
|
return conn
|
|
except Exception as e:
|
|
print(f"Error conectando a HANA: {e}")
|
|
return None
|
|
|
|
########################################################
|
|
def obtener_registros(days=30):
|
|
"""
|
|
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.
|
|
"""
|
|
|
|
schema = CONFIG.hana.schema
|
|
query = f"""
|
|
SELECT T0."DocNum", T0."DocEntry", TO_VARCHAR(T0."DocDate",'YYYY-MM-DD') "DocDate"
|
|
, TO_VARCHAR(T0."TaxDate",'YYYY-MM-DD') "TaxDate", T0."CardName", T0."FolioPref", T0."FolioNum", T0."Indicator", T0."DocCur"
|
|
, CASE T0."DocCur" WHEN 'SOL' THEN T0."DocTotal" ELSE T0."DocTotalFC" END "DocTotal"
|
|
, T0."U_PGX_PDF" "PDF", T0."U_PGX_XML" "XML", T0."LicTradNum"
|
|
FROM "{schema}".OPCH T0
|
|
WHERE T0."CANCELED" = 'N'
|
|
AND T0."LicTradNum" = '20100047218'
|
|
AND (T0."U_PGX_PDF" = 'N' OR T0."U_PGX_XML" = 'N')
|
|
AND T0."CreateDate" >= ADD_DAYS(CURRENT_DATE, -{days});
|
|
"""
|
|
|
|
try:
|
|
print(f"Conectando al servidor SAP HANA ({CONFIG.hana.server}:{CONFIG.hana.port})...")
|
|
connection = get_connection()
|
|
if not connection:
|
|
raise ConnectionError("No se pudo establecer la conexión a SAP HANA.")
|
|
|
|
cursor = connection.cursor()
|
|
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] Ocurrió un error en la conexión o consulta a HANA: {e}")
|
|
return []
|
|
finally:
|
|
if cursor:
|
|
cursor.close()
|
|
if connection:
|
|
connection.close()
|
|
print("Conexión con SAP HANA cerrada.")
|
|
|
|
######################################################## |