322 lines
11 KiB
Python
322 lines
11 KiB
Python
|
|
import requests
|
|
import urllib3
|
|
import os
|
|
import json
|
|
from typing import Optional, Dict, Any
|
|
from dotenv import load_dotenv
|
|
import hana_manager
|
|
|
|
|
|
# ============================================================================
|
|
# CONFIGURACIÓN
|
|
# ============================================================================
|
|
|
|
urllib3.disable_warnings()
|
|
|
|
load_dotenv(dotenv_path=".env")
|
|
|
|
SAPB1_URL = os.getenv("SAP_B1_URL", "")
|
|
SAPB1_USER = os.getenv("SAP_B1_USER", "")
|
|
SAPB1_PASSWORD = os.getenv("SAP_B1_PASSWORD", "")
|
|
SAPB1_COMPANY = os.getenv("SAP_B1_COMPANY", "")
|
|
SAPB1_TIMEOUT = float(os.getenv("SAP_B1_TIMEOUT", 15.0))
|
|
SAPB1_VERIFY_SSL = os.getenv("SAP_B1_VERIFY_SSL", "False").lower() == "true"
|
|
|
|
# Almacenamiento global de sesión
|
|
_sap_session: dict = {"headers": None, "b1session": None, "routeid": None}
|
|
|
|
|
|
# ============================================================================
|
|
# LOGIN
|
|
# ============================================================================
|
|
def login() -> dict:
|
|
payload = {
|
|
"CompanyDB": SAPB1_COMPANY,
|
|
"UserName": SAPB1_USER,
|
|
"Password": SAPB1_PASSWORD,
|
|
}
|
|
print("🔐 Iniciando sesión en SAP B1 Service Layer...")
|
|
resp = requests.post(
|
|
f"{SAPB1_URL}/Login",
|
|
json=payload,
|
|
timeout=SAPB1_TIMEOUT,
|
|
verify=SAPB1_VERIFY_SSL,
|
|
)
|
|
if resp.status_code != 200:
|
|
raise ConnectionError(
|
|
f"Login fallido: HTTP {resp.status_code} - {resp.text}"
|
|
)
|
|
|
|
data = resp.json()
|
|
b1session = data["SessionId"]
|
|
routeid = resp.cookies.get("ROUTEID", "")
|
|
|
|
headers = {"Cookie": f"B1SESSION={b1session}; ROUTEID={routeid}"}
|
|
_sap_session.update({"headers": headers, "b1session": b1session, "routeid": routeid})
|
|
print(f"✅ Sesión SAP B1 establecida (B1SESSION: {b1session[:10]}...)")
|
|
return headers
|
|
|
|
|
|
# ============================================================================
|
|
# LOGOUT
|
|
# ============================================================================
|
|
def logout() -> None:
|
|
if not _sap_session["b1session"]:
|
|
print("⚠️ No hay sesión activa para cerrar.")
|
|
return
|
|
print("🔓 Cerrando sesión SAP B1...")
|
|
try:
|
|
resp = requests.post(
|
|
f"{SAPB1_URL}/Logout",
|
|
headers=_sap_session["headers"],
|
|
timeout=SAPB1_TIMEOUT,
|
|
verify=SAPB1_VERIFY_SSL,
|
|
)
|
|
if resp.status_code == 200:
|
|
print("✅ Sesión SAP B1 cerrada correctamente.")
|
|
else:
|
|
print(f"⚠️ Logout respondió HTTP {resp.status_code}")
|
|
except Exception as exc:
|
|
print(f"❌ Error al cerrar sesión: {exc}")
|
|
finally:
|
|
_sap_session.update({"headers": None, "b1session": None, "routeid": None})
|
|
|
|
|
|
# ============================================================================
|
|
# SESIÓN (interna - lazy login)
|
|
# ============================================================================
|
|
def _ensure_session() -> dict:
|
|
if _sap_session["headers"]:
|
|
return _sap_session["headers"]
|
|
return login()
|
|
|
|
|
|
# ============================================================================
|
|
# LOGGING DE RESPUESTA
|
|
# ============================================================================
|
|
def _log_response(method: str, endpoint: str, response: requests.Response, payload: Any = None) -> None:
|
|
log_line = (
|
|
f" ├── [{method}] {endpoint} → {response.status_code}"
|
|
f" ({len(response.content)} bytes)"
|
|
)
|
|
if response.status_code >= 400:
|
|
print("⚠️ %s - %s", log_line, response.text[:300])
|
|
else:
|
|
print(log_line)
|
|
|
|
|
|
# ============================================================================
|
|
# PROXY CENTRALIZADO
|
|
# ============================================================================
|
|
def forward_to_sapb1(
|
|
method: str,
|
|
endpoint_path: str,
|
|
json_data: Optional[Dict[str, Any]] = None,
|
|
params: Optional[Dict[str, Any]] = None,
|
|
extra_headers: Optional[Dict[str, str]] = None
|
|
) -> requests.Response:
|
|
if not method or not endpoint_path:
|
|
raise ValueError("method y endpoint_path son requeridos")
|
|
|
|
method = method.upper()
|
|
headers = _ensure_session()
|
|
|
|
url = f"{SAPB1_URL}/{endpoint_path}"
|
|
req_headers = {
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json",
|
|
**_sap_session["headers"],
|
|
}
|
|
|
|
if extra_headers:
|
|
req_headers.update(extra_headers)
|
|
|
|
print(f" ├── 🔄 [{method}] {endpoint_path}")
|
|
|
|
try:
|
|
resp = requests.request(
|
|
method=method,
|
|
url=url,
|
|
json=json_data,
|
|
params=params,
|
|
headers=req_headers,
|
|
timeout=SAPB1_TIMEOUT,
|
|
verify=SAPB1_VERIFY_SSL,
|
|
)
|
|
except requests.exceptions.Timeout:
|
|
raise ConnectionError(f"Timeout ({SAPB1_TIMEOUT}s) en {endpoint_path}")
|
|
except requests.exceptions.ConnectionError as exc:
|
|
raise ConnectionError(f"No se pudo conectar con SAP B1: {exc}")
|
|
except requests.RequestException as exc:
|
|
raise ConnectionError(f"Error en solicitud HTTP: {exc}")
|
|
|
|
if resp.status_code == 401:
|
|
print("⚠️ Sesión expirada (401). Reautenticando e intentando una vez...")
|
|
req_headers["Cookie"] = login()["Cookie"]
|
|
resp = requests.request(
|
|
method=method,
|
|
url=url,
|
|
json=json_data,
|
|
params=params,
|
|
headers=req_headers,
|
|
timeout=SAPB1_TIMEOUT,
|
|
verify=SAPB1_VERIFY_SSL,
|
|
)
|
|
|
|
_log_response(method, endpoint_path, resp, json_data)
|
|
|
|
if resp.status_code >= 400:
|
|
try:
|
|
error_body = resp.json()
|
|
sap_message = (
|
|
error_body.get("error", {})
|
|
.get("message", {})
|
|
.get("value", resp.text)
|
|
)
|
|
except (json.JSONDecodeError, AttributeError):
|
|
sap_message = resp.text
|
|
raise ConnectionError(
|
|
f"Error en SAP {method} {endpoint_path}: "
|
|
f"HTTP {resp.status_code} - {sap_message}"
|
|
)
|
|
|
|
return resp
|
|
|
|
|
|
# ============================================================================
|
|
# CREAR DRAFT (POST /Drafts)
|
|
# ============================================================================
|
|
def create_draft(payload: dict) -> dict:
|
|
resp = forward_to_sapb1("POST", "Drafts", json_data=payload)
|
|
result = resp.json()
|
|
print(f" ├── ✅ Draft creado DocEntry={result.get('DocEntry')}")
|
|
return result
|
|
|
|
|
|
# ============================================================================
|
|
# OBTENER FACTURA DE COMPRA (GET /PurchaseInvoices(DocEntry))
|
|
# ============================================================================
|
|
def get_purchase_invoice(doc_entry: int) -> dict:
|
|
resp = forward_to_sapb1("GET", f"PurchaseInvoices({doc_entry})")
|
|
result = resp.json()
|
|
print(f" ├── ✅ PurchaseInvoices({doc_entry}) obtenida")
|
|
return result
|
|
|
|
# ============================================================================
|
|
# OBTENER GUIA (GET /DeliveryNotes)
|
|
# ============================================================================
|
|
def get_delivery_notes(tipo: str, serie: str, numero: str) -> list:
|
|
"""
|
|
Obtiene DeliveryNotes de SAP B1.
|
|
|
|
Fase 1 — Busca DocEntry/DocNum/CardCode/CardName:
|
|
- Primero intenta Service Layer con $select.
|
|
- Si no hay resultados, fallback a HANA (consulta ODLN con LIKE).
|
|
|
|
Fase 2 — Por cada DocEntry obtenido, hace GET /DeliveryNotes({DocEntry})
|
|
para recuperar el documento completo.
|
|
|
|
Returns:
|
|
list[dict] — Documentos completos de DeliveryNotes, o [] si no se
|
|
encontró nada en ninguna fuente.
|
|
"""
|
|
codigo = f"{serie}{tipo}-{numero}"
|
|
|
|
# ── Fase 1: Obtener DocEntries ──────────────────────────────────────
|
|
entries = _get_delivery_entries(tipo, serie, numero)
|
|
|
|
if not entries:
|
|
print(f" │ ⚠️ Guía {codigo}: sin registros en SAP (SL + HANA)")
|
|
return []
|
|
|
|
# ── Fase 2: Obtener documento completo por cada DocEntry ─────────────
|
|
docs = []
|
|
for entry in entries:
|
|
doc_entry = entry["DocEntry"]
|
|
try:
|
|
resp = forward_to_sapb1("GET", f"DeliveryNotes({doc_entry})")
|
|
doc = resp.json()
|
|
docs.append(doc)
|
|
print(f" │ └─ DeliveryNote({doc_entry}) obtenido")
|
|
except Exception as e:
|
|
print(f" │ ⚠️ Error obteniendo DeliveryNote({doc_entry}): {e}")
|
|
|
|
return docs
|
|
|
|
|
|
def _get_delivery_entries(tipo: str, serie: str, numero: str) -> list:
|
|
"""Fase 1: intenta SL, fallback a HANA. Retorna [{DocEntry, DocNum, CardCode, CardName}, ...]."""
|
|
codigo = f"{serie}{tipo}-{numero}"
|
|
|
|
# ── Intentar Service Layer ───────────────────────────────────────────
|
|
try:
|
|
filter_str = (
|
|
f"Cancelled eq 'tNO'"
|
|
f" and U_BPP_MDSD eq '{tipo}'"
|
|
f" and U_BPP_MDTD eq '{serie}'"
|
|
f" and contains(U_BPP_MDCD,'{numero}')"
|
|
)
|
|
resp = forward_to_sapb1(
|
|
"GET", "DeliveryNotes",
|
|
params={
|
|
"$filter": filter_str,
|
|
"$select": "DocEntry,DocNum,CardCode,CardName",
|
|
"$top": "10",
|
|
},
|
|
)
|
|
sl_entries = resp.json().get("value", [])
|
|
if sl_entries:
|
|
print(f" │ 🔍 Guía {codigo}: SL → {len(sl_entries)} DocEntry encontrados")
|
|
return sl_entries
|
|
|
|
print(f" │ 🔍 Guía {codigo}: SL → 0 resultados. Consultando HANA...")
|
|
|
|
except Exception as e:
|
|
print(f" │ ⚠️ Guía {codigo}: SL falló ({e}). Consultando HANA...")
|
|
|
|
# ── Fallback a HANA ──────────────────────────────────────────────────
|
|
try:
|
|
hana_entries = hana_manager.get_delivery(tipo, serie, numero)
|
|
if hana_entries:
|
|
print(f" │ 🔍 Guía {codigo}: HANA → {len(hana_entries)} DocEntry encontrados")
|
|
return hana_entries
|
|
print(f" │ 🔍 Guía {codigo}: HANA → 0 registros")
|
|
except Exception as e:
|
|
print(f" │ ⚠️ Guía {codigo}: HANA falló ({e})")
|
|
|
|
return []
|
|
|
|
# ============================================================================
|
|
# OBTENER ZONAS (GET /U_IDC_ZONAS)
|
|
# ============================================================================
|
|
def get_zonas() -> list:
|
|
|
|
filter_str = (
|
|
f"U_Activo eq 'Y'"
|
|
)
|
|
|
|
sap_params = {
|
|
"$skip": "0",
|
|
"$top": "100",
|
|
"$inlinecount": "allpages"
|
|
}
|
|
|
|
sap_params["$select"] = "Code, Name"
|
|
|
|
base_headers = {"B1S-PageSize": "100"}
|
|
|
|
if filter_str:
|
|
sap_params["$filter"] = filter_str
|
|
|
|
resp = forward_to_sapb1(
|
|
method="GET"
|
|
, endpoint_path="U_IDC_ZONAS"
|
|
, params=sap_params
|
|
, extra_headers=base_headers
|
|
)
|
|
result = resp.json()
|
|
value = result.get("value", [])
|
|
print(f" ├── ✅ U_IDC_ZONAS obtenida ({len(value)} registros)")
|
|
return value
|