desarrollo instalado en el cliente globe italia

sql_manager.py

@@ -0,0 +1,110 @@
+import os
+import pymssql
+from dotenv import load_dotenv
+from logger_manager import get_logger
+# ============================================================================
+load_dotenv(dotenv_path=".env")
+
+SQL_HOST = os.getenv("SQL_HOST", "")
+SQL_USER = os.getenv("SQL_USER", "")
+SQL_PASSWORD = os.getenv("SQL_PASSWORD", "")
+SQL_DATABASE = os.getenv("SQL_DATABASE", "")
+# ============================================================================
+_conn = None
+
+log = get_logger()
+
+# ============================================================================
+def get_connection():
+    global _conn
+    if _conn is None:
+        try:
+            _conn = pymssql.connect(
+                server=SQL_HOST,
+                user=SQL_USER,
+                password=SQL_PASSWORD,
+                database=SQL_DATABASE,
+            )
+        except Exception as e:
+            log.error("❌ Error conectando a SQL Server: {}".format(e))
+            return None
+    return _conn
+
+# ============================================================================
+def execute_query(query: str, params: tuple = ()) -> list[dict]:
+    conn = get_connection()
+    if not conn:
+        return []
+    try:
+        cursor = conn.cursor()
+        if params:
+            cursor.execute(query, params)
+        else:
+            cursor.execute(query)
+        columns = [desc[0] for desc in cursor.description]
+        rows = cursor.fetchall()
+        cursor.close()
+        return [dict(zip(columns, row)) for row in rows]
+    except Exception as e:
+        log.error("❌ Error en query: {}".format(e))
+        return []
+
+# ============================================================================
+def get_client_by_gln(emisor: str, lugar_entrega: str) -> str | None:
+    query = (
+        "SELECT [Address],[CardCode],[Street],[City],[Country],[County] "
+        "FROM CRD1 T1 "
+        f"WHERE [AdresType] = 'S' "
+        f"AND T1.[Address] LIKE '%{emisor}.{lugar_entrega}%'"
+    )
+    rows = execute_query(query)
+
+    if not rows:
+        return None
+    row = rows[0]
+    return {
+        "CardCode": row.get("CardCode"),
+        "City": row.get("City"),
+        "County": row.get("County")
+    }
+
+# ============================================================================
+def get_item_catalog(cardcode: str) -> dict[str, str]:
+    rows = execute_query(
+        "SELECT [Substitute], [ItemCode] FROM OSCN "
+        f"WHERE [CardCode] = '{cardcode}'"
+    )
+    return {r["Substitute"]: r["ItemCode"] for r in rows if r.get("Substitute")}
+
+# ============================================================================
+def get_price_list(cardcode: str) -> int | None:
+    rows = execute_query(
+        f"SELECT T0.[ListNum] FROM OCRD T0 WHERE T0.[CardCode] = '{cardcode}'"
+    )
+    return rows[0].get("ListNum") if rows else None
+
+# ============================================================================
+def get_prices(pricelist: int) -> dict[str, float]:
+    rows = execute_query(
+        "SELECT [ItemCode],[Price] FROM ITM1 "
+        f"WHERE [PriceList] = {pricelist} AND [Price] > 0.000000"
+    )
+    return {r["ItemCode"]: float(r["Price"]) for r in rows if r.get("Price")}
+
+# ============================================================================
+def get_card_name(cardcode: str) -> str | None:
+    rows = execute_query(
+        f"SELECT [CardName] FROM OCRD WHERE [CardCode] = '{cardcode}'"
+    )
+    return rows[0].get("CardName") if rows else None
+
+# ============================================================================
+def cerrar_conexion():
+    global _conn
+    if _conn:
+        try:
+            _conn.close()
+        except:
+            pass
+        _conn = None
+# ============================================================================