descarga_factura_transporte/hana_manager.py

#!/usr/bin/env python3
import os
import sys
import json
from datetime import datetime, date
from hdbcli import dbapi
from dotenv import load_dotenv

# ============================================================================
# CONFIGURACIÓN
# ============================================================================

# Cargar variables de entorno desde .env (ruta absoluta: no depende del
# directorio de trabajo desde el que se lance el proceso)
load_dotenv(dotenv_path=os.path.join(os.path.dirname(__file__), ".env"))

# Configuración de la API
HANA_HOST = os.getenv("HANA_SERVER", "")
HANA_PORT = int(os.getenv("HANA_PORT", 30015))
HANA_USER = os.getenv("HANA_USER", "")
HANA_PASSWORD = os.getenv("HANA_PASSWORD", "")
HANA_SCHEMA = os.getenv("HANA_SCHEMA", "")

# ============================================================================
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 get_delivery(tipo: str, serie: str, numero: str) -> list:
    """
    Consulta la tabla ODLN en SAP HANA para obtener DeliveryNotes.
    Si no encuentra el tipo, serie y número en ODLN, busca en OWTR
    (Traslados de Inventario) como fallback.

    Args:
        tipo: Código de tipo de documento SUNAT, ej. "09"/"31" (U_BPP_MDTD).
        serie: Serie del documento, ej. "T001" (U_BPP_MDSD).
        numero: Número del documento (se busca con LIKE).

    Returns:
        list[dict] — Lista con DocEntry, DocNum, CardCode, CardName, Tabla.
        "Tabla" indica en qué tabla se encontró el registro ("ODLN" u
        "OWTR"), para que el llamador sepa qué endpoint de Service Layer
        usar (DeliveryNotes o StockTransfers).
    """
    connection = get_connection()
    if not connection:
        return []

    try:
        results = _query_delivery_table(connection, "ODLN", tipo, serie, numero)
        if results:
            return results
        return _query_delivery_table(connection, "OWTR", tipo, serie, numero)
    finally:
        connection.close()

# ============================================================================
def _query_delivery_table(connection, tabla: str, tipo: str, serie: str, numero: str) -> list:
    """Ejecuta la consulta de tipo/serie/número contra la tabla indicada (ODLN u OWTR)."""
    schema = HANA_SCHEMA
    query = f'''SELECT "DocEntry", "DocNum", "CardCode", "CardName"
    FROM "{schema}"."{tabla}"
    WHERE "CANCELED" = 'N'
      AND "U_BPP_MDTD" = ?
      AND "U_BPP_MDSD" = ?
      AND "U_BPP_MDCD" LIKE ?'''

    pattern = f"%{numero}%"

    cursor = None
    try:
        cursor = connection.cursor()
        cursor.execute(query, (tipo, serie, pattern))
        columns = [column[0] for column in cursor.description]
        results = []
        for row in cursor.fetchall():
            registro = dict(zip(columns, row))
            registro["Tabla"] = tabla
            results.append(registro)
        return results
    except Exception as e:
        print(f"    ❌ Error en consulta HANA get_delivery ({tabla}): {e}")
        return []
    finally:
        if cursor:
            cursor.close()

# ============================================================================
def obtener_valor_referencial(serie: str, numero: str) -> float:
    """
    Ejecuta el stored procedure Sp_ObtenerValorReferencial en SAP HANA para
    obtener el valor referencial de una guía de remisión (tipo 09).

    Args:
        serie: Serie de la guía de remisión (ej. "T001").
        numero: Número de la guía de remisión (ej. "132800").

    Returns:
        float — Columna ``ValorReferencial`` del primer registro devuelto
        por el SP, o 0.0 si no hay resultados o si ocurre un error.
    """
    schema = HANA_SCHEMA
    query = f'CALL "{schema}"."Sp_ObtenerValorReferencial"(?, ?)'

    connection = get_connection()
    if not connection:
        return 0.0

    cursor = None
    try:
        cursor = connection.cursor()
        cursor.execute(query, (serie, numero))
        columns = [column[0] for column in cursor.description]
        row = cursor.fetchone()
        if not row:
            return 0.0
        valor = dict(zip(columns, row)).get("ValorReferencial")
        return float(valor) if valor is not None else 0.0
    except Exception as e:
        print(f"    ❌ Error en consulta HANA obtener_valor_referencial: {e}")
        return 0.0
    finally:
        if cursor:
            cursor.close()
        if connection:
            connection.close()

# ============================================================================