Versión Actualizada

a2billing/a2billing.py

+import mariadb
+
+DB_CREDENTIALS = {
+    'host': '192.168.2.220',
+    'user': 'api-mbilling',
+    'password': 'mbillingmbilling',
+    'port': 3306,
+    'database': 'mya2billing'
+}
+
+
+def account_data(callerid: str) -> dict:
+
+    with mariadb.connect(**DB_CREDENTIALS) as cnx:
+        with cnx.cursor() as cursor:
+            headers = ['account', 'secret', 'credit']
+            cursor.execute(
+                'SELECT car.username, car.uipass, round(car.credit, 2) credit FROM cc_callerid cid\
+                INNER JOIN cc_card car ON car.id=cid.id_cc_card WHERE cid.cid=?',
+                (callerid,)
+            )
+            data = cursor.fetchone()
+
+    return dict(zip(headers, data)) if data else None

mbilling/functions.py

+from decimal import Decimal
 import subprocess
 import pathlib
 import re
@@ -23,6 +24,20 @@ DB_CREDENTIALS = {
 }
 
 
+class SipAccount:
+
+    def __init__(self, callerid: str, account: str, secret: str,
+                 credit: float, plan: int, offer: int, active: bool, postpaid: bool) -> None:
+        self.callerid = callerid
+        self.account = account
+        self.secret = secret
+        self.credit = credit
+        self.plan = plan
+        self.offer = offer
+        self.active = active
+        self.postpaid = postpaid
+
+
 class MagnusCommandError(Exception):
 
     def __init__(self, *args: object, ) -> None:
@@ -110,7 +125,7 @@ def _create_did(phone_number: str):
             raise MagnusCommandError(result.decode('utf-8'))
 
 
-def _update_payment(phone_number: str, postpay: bool = False, credit: float = 0.0):
+def _update_payment(phone_number: str, postpay: bool = False, credit: Decimal = Decimal('0.0')):
 
     with mariadb.connect(**DB_CREDENTIALS) as cnx:
         with cnx.cursor() as cursor:
@@ -156,7 +171,7 @@ def create_user(phone_number: str, name: str, surname: str, password: str = None
     return password
 
 
-def add_balance(phone_number: str, quantity: float, tax: float = 0.21,  description: str = 'Recarga Automática'):
+def add_balance(phone_number: str, quantity: Decimal, tax: Decimal = Decimal('0.21'),  description: str = 'Recarga Automática'):
     """Añade saldo a una cuenta SIP
 
     Args:
@@ -193,14 +208,17 @@ def get_user(phone_number: str) -> dict:
 
     with mariadb.connect(**DB_CREDENTIALS) as cnx:
         with cnx.cursor() as cursor:
-            headers = ['username', 'password', 'host', 'credit',
-                       'firstname', 'lastname', 'plan', 'offer']
+            headers = ['sip','username', 'password', 'host', 'credit',
+                       'firstname', 'lastname', 'plan', 'offer',
+                       'active', 'postpaid']
             cursor.execute(
-                """SELECT u.username, s.secret password, s.host, u.credit, u.firstname, u.lastname, p.id plan, o.id offer FROM mbilling.pkg_user u
-                    INNER JOIN mbilling.pkg_plan p ON u.id_plan=p.id
-                    LEFT JOIN mbilling.pkg_offer o ON u.id_offer=o.id
-                    INNER JOIN mbilling.pkg_sip s ON u.username=s.name
-                    WHERE username=?""",
+                'SELECT s.name, u.username, s.secret password, s.host, u.credit, u.firstname,\
+                    u.lastname, p.id plan, IFNULL(o.id, 0) offer, u.active active, u.typepaid\
+                    FROM mbilling.pkg_user u\
+                    INNER JOIN mbilling.pkg_plan p ON u.id_plan=p.id\
+                    LEFT JOIN mbilling.pkg_offer o ON u.id_offer=o.id\
+                    INNER JOIN mbilling.pkg_sip s ON u.id=s.id_user\
+                    WHERE s.name=?',
                 (phone_number,)
             )
 
@@ -209,7 +227,7 @@ def get_user(phone_number: str) -> dict:
     return dict(zip(headers, query)) if query else None
 
 
-def fully_ported_clis() -> 'list[str]':
+def fully_ported_users() -> 'list[tuple]':
     """Devuelve los CLI completamente portados a MBilling.
 
     Algunos usuarios pueden recibir llamadas en MB pero realizar la salida\
@@ -224,17 +242,16 @@ def fully_ported_clis() -> 'list[str]':
     with mariadb.connect(**DB_CREDENTIALS) as cnx:
         with cnx.cursor() as cursor:
             cursor.execute(
-                """
-                SELECT username cli FROM mbilling.pkg_user u
-                    INNER JOIN mbilling.pkg_cdr c ON c.callerid = u.username
-                    INNER JOIN mbilling.pkg_trunk t ON c.id_trunk = t.id
-                    WHERE c.id_trunk is not NULL AND t.trunkcode != 'a2billing'
-                    GROUP BY username;
-                """)
+                'SELECT s.name cli, u.username account FROM mbilling.pkg_user u\
+                    INNER JOIN mbilling.pkg_cdr c ON c.callerid = u.username\
+                    INNER JOIN mbilling.pkg_trunk t ON c.id_trunk = t.id\
+                    INNER JOIN mbilling.pkg_sip s ON s.id_user = u.id\
+                    WHERE c.id_trunk is not NULL AND t.trunkcode != \'a2billing\' AND u.active=1\
+                    GROUP BY s.name ORDER BY u.username')
 
-            query = [cli[0] for cli in cursor.fetchall()]
+            data = cursor.fetchall()
 
-    return query
+    return data
 
 
 def active_users() -> 'list[dict]':
@@ -244,25 +261,14 @@ def active_users() -> 'list[dict]':
         list[dict]: Listado de Usuarios y sus Datos
     """
 
-    with mariadb.connect(**DB_CREDENTIALS) as cnx:
-        with cnx.cursor() as cursor:
-            headers = ['username', 'password', 'host', 'credit',
-                       'firstname', 'lastname', 'plan', 'offer', 'active', 'postpaid']
-            cursor.execute(
-                """SELECT u.username, s.secret password, s.host, u.credit, u.firstname, u.lastname,\
-                    p.id plan_id, o.id offer_id, u.active, u.typepaid
-                    FROM mbilling.pkg_user u
-                    INNER JOIN mbilling.pkg_plan p ON u.id_plan=p.id
-                    INNER JOIN mbilling.pkg_sip s ON u.username=s.name
-                    LEFT JOIN mbilling.pkg_offer o ON u.id_offer=o.id"""
-            )
+    accounts = []
+    for account in fully_ported_users():
+        accounts.append(get_user(account[0]))
 
-            query = [dict(zip(headers, account))
-                     for account in cursor.fetchall()]
+    return accounts
 
-    return query
 
-def get_cdrs(phone_number : str, since : datetime = None, upto : datetime = None) -> 'list[dict]':
+def get_cdrs(phone_number: str, since: datetime = None, upto: datetime = None) -> 'list[dict]':
     """Obtiene los CDRs de una Línea
 
     Args:
@@ -278,7 +284,8 @@ def get_cdrs(phone_number : str, since : datetime = None, upto : datetime = None
         with cnx.cursor() as cursor:
 
             # Se eliminan llamadas a 016
-            cursor.execute('DELETE FROM mbilling.pkg_cdr WHERE calledstation IN  (\'016\', \'34016\')')
+            cursor.execute(
+                'DELETE FROM mbilling.pkg_cdr WHERE calledstation IN  (\'016\', \'34016\')')
 
             # Carga de CDRs
             if upto and since:
@@ -313,8 +320,33 @@ def get_cdrs(phone_number : str, since : datetime = None, upto : datetime = None
             data = cursor.fetchall()
 
     cdr = []
-    headers = ['callerid', 'destination', 'description', 'timestamp', 'duration', 'price', 'cost']
+    headers = ['callerid', 'destination', 'description',
+               'timestamp', 'duration', 'price', 'cost']
     for line in data:
         cdr.append(dict(zip(headers, line)))
 
     return cdr
+
+
+def get_sip_accounts(only_postpaid : bool = False) -> 'list[SipAccount]':
+    """Obtiene todas las cuentas SIPs de MagnusBilling
+
+    Returns:
+        list[SipAccount]: Listado de las Cuentas
+    """
+
+    query = "SELECT s.callerid sip, u.username acc, s.secret secret, round(u.credit, 2) credit, u.id_plan plan,\
+                    IFNULL(u.id_offer, 0) offer, u.active active, u.typepaid postpaid\
+                    FROM mbilling.pkg_sip s\
+                    INNER JOIN mbilling.pkg_user u ON u.id=s.id_user\
+                    WHERE s.callerid > ''"
+
+    if only_postpaid:
+        query += ' AND u.typepaid=1'
+
+    with mariadb.connect(**DB_CREDENTIALS) as cnx:
+        with cnx.cursor() as cursor:
+            cursor.execute(query)
+            sip_data = cursor.fetchall()
+
+    return [SipAccount(*sip) for sip in sip_data]