Se añade CLI

.vscode/launch.json

@@ -5,15 +5,20 @@
   "version": "0.2.0",
   "configurations": [
     {
+      "name": "CLI: Sincronizar",
+      "type": "python",
+      "request": "launch",
+      "program": "cli.py",
+      "args": ["synchronize"],
+      "console": "integratedTerminal",
+      "justMyCode": true
+    },
+    {
       "name": "API",
       "type": "python",
       "request": "launch",
       "module": "uvicorn",
-            "args": [
-                "main:app",
-                "--port",
-                "36002"
-            ],
+      "args": ["main:app", "--port", "36002"],
       "jinja": true,
       "justMyCode": true
     }

cli.py

+from http.client import CannotSendRequest
+import typer
+from mbilling.functions import active_users, fully_ported_clis, get_cdrs
+import time
+import pyodbc
+import os
+from datetime import timedelta
+# Credenciales SQL Server
+sql_server_hostname = "192.168.2.200"
+sql_server_database = "bdd_wifi"
+sql_server_username = "api-mbilling"
+sql_server_password = "mbillingmbilling"
+sql_server_driver = 'ODBC Driver 17 for SQL Server' if os.name == 'nt' else 'FreeTDS'
+
+# Cadena de Conexión
+SQL_SERVER_CNX_STRING = "DRIVER={driver};SERVER={server};\
+    DATABASE={database};UID={username};PWD={password};TDS_Version=7.0;".format(
+    driver=sql_server_driver, server=sql_server_hostname,
+    database=sql_server_database, username=sql_server_username,
+    password=sql_server_password)
+
+# ID en Magnus de Tarifa Plana
+TARIFA_PLANA_ID = 4
+
+# Inicialización del CLI
+app = typer.Typer()
+
+# Funciones Auxiliares
+
+
+def new_plataform_clis() -> 'list[str]':
+    """Obtiene los CLI activos en la Magnus Billing
+
+    Es posible que algunos números estén en Magnus Billing pero\
+    la gestión de su saldo y llamadas salientes se realize aún\
+    desde A2Billing por alguna razón. Estos números no se deben\
+    tener en cuenta en la actualización de Saldo y valores, por lo\
+    que con está función obtendremos los números que debemos actualizar.
+
+    Returns:
+        list[str]: Listado de números que podemos actualizar.
+    """
+
+    with pyodbc.connect(SQL_SERVER_CNX_STRING) as cnx:
+        with cnx.cursor() as cursor:
+            cursor.execute(
+                "SELECT DDI FROM TB_ServicioFijo WHERE FechaBaja IS NULL AND NuevaPlataforma = 1")
+            active_dids = [d[0] for d in cursor.fetchall()]
+
+    return active_dids
+
+
+def new_plataform_activate(*cli: str):
+    """Agrega al CLI a Nueva Plataforma
+
+    Modifica el Programa de Facturación para indicar que este CLI está\
+    completamente portado a la Nueva Plataforma. Los números con este indicador\
+    serán aquellos que podrán ser modificados por este programa.
+
+    """
+
+    with pyodbc.connect(SQL_SERVER_CNX_STRING) as cnx:
+        with cnx.cursor() as cursor:
+            for c in cli:
+                cursor.execute(
+                    "UPDATE TB_ServicioFijo SET NuevaPlataforma = 1 WHERE DDI = ?", (
+                        c,)
+                )
+
+
+def update_target_clis() -> 'list[str]':
+    """Obtiene el listado de CLIs que deben ser Actualizados.
+
+    Los números que han de ser actualizados deben estar en la nueva plataforma\
+    y estar activos en la Magnus Billing. Esta función busca aquellos CLIs\
+    que cumplen ambas condiciones y deben ser tenidos en cuenta a la hora\
+    de las actualizaciones de las cuentas.
+
+    Returns:
+        list[str]: Listado de CLIs a Actualizar.
+    """
+
+    return list(set(fully_ported_clis()) & set(new_plataform_clis()))
+
+
+def get_last_updated_cdr(account: str = None, only_post_payment: bool = True) -> dict:
+    """Obtiene la fecha de la última actualización del CDR de una Cuenta
+
+    Args:
+        account (str, optional): _description_. Defaults to None.
+        only_post_payment (bool, optional): _description_. Defaults to True.
+
+    Returns:
+        dict: _description_
+    """
+
+    with pyodbc.connect(SQL_SERVER_CNX_STRING) as cnx:
+        with cnx.cursor() as cursor:
+            # Construcción de la Cadena
+            query = "SELECT DDI FROM [bdd_wifi].[dbo].[TB_ServicioFijo] WHERE NuevaPlataforma=1"
+            query = query + " AND Pospago = 1" if only_post_payment else query
+            query = query + f" AND DDI = {account}" if account else query
+
+            # Obtención de Números a Actualizar
+            cursor.execute(query)
+
+            cdr_last_updated = {}
+            for ddi in cursor.fetchall():
+                cursor.execute(
+                    'SELECT TOP 1 fecha FROM bdd_wifi.dbo.TB_OMV_CDR_Cliente WHERE DDi = ? ORDER BY fecha DESC',
+                    (ddi[0])
+                )
+
+                last_updated = cursor.fetchone()
+                cdr_last_updated[ddi[0]] = last_updated[0] if last_updated else None
+
+    return cdr_last_updated
+
+# Definición de Comandos
+
+
+@app.command()
+def update_plataform():
+    """Actualiza las CLIs agregados a la Nueva Plataforma
+    """
+
+    typer.echo('Actualizando Líneas en Nueva Plataforma.')
+    new_plataform_activate(*fully_ported_clis())
+    typer.echo('Actualización finalizada.')
+
+
+@app.command()
+def update_account(account: str = typer.Argument(None)):
+
+    # Obtiene los números completamente portados.
+    target_clis = update_target_clis() if not account else account
+
+    # Líneas Actualizadas
+    updated = 0
+    non_updated = 0
+
+    # Actualización de Balance en Programa de Facturación
+    with pyodbc.connect(SQL_SERVER_CNX_STRING) as cnx:
+        with cnx.cursor() as cursor:
+            with typer.progressbar(active_users(), label="Actualizando Balance") as progress:
+                for sip_account in progress:
+                    if sip_account['username'] in target_clis:
+                        flat_rate = sip_account['offer'] if sip_account['plan'] == TARIFA_PLANA_ID else None
+                        cursor.execute(
+                            """UPDATE [bdd_wifi].[dbo].[TB_ServicioFijo]
+                                SET SaldoActual=?, Activo=?, TarifaPlana=?, Pospago=?, IdTarifa=?
+                                WHERE DDI=? AND FechaBaja is not NULL""",
+                            (sip_account['credit'], sip_account['active'],
+                             flat_rate, sip_account['postpaid'], sip_account['plan'], sip_account['username'])
+                        )
+                        updated = updated + 1
+
+                    else:
+                        non_updated = non_updated + 1
+
+    msg = f"De un total de {updated + non_updated} líneas, "
+    msg = msg + typer.style(updated, fg=typer.colors.GREEN, bold=True)
+    msg = msg + " han sido actualizadas correctamente, "
+    msg = msg + typer.style(non_updated, fg=typer.colors.RED, bold=True)
+    msg = msg + " no han sido actualizadas por estar activas en A2Billing."
+    typer.echo(msg)
+
+
+@app.command()
+def import_cdr(account: str = typer.Argument(None), only_charged: bool = typer.Argument(False), only_post_payment: bool = typer.Argument(True)):
+    """Importa los CDR de las Líneas de Magnus Billing
+
+    Se procesan las llamadas realiadas por las líneas seleccionadas y se incluyen \
+    en el programa de facturación. Se puede seleccionar si sólo se desean importar \
+    líneas de postpago o llamadas con coste.
+
+    Args:
+        account (str, optional): Número de cuenta, por defecto toda las activas. Defaults to typer.Argument(None).
+        paid (bool, optional): Sólo CDRs con Coste. Defaults to typer.Argument(False).
+        postpayment (bool, optional): Sólo CDRs de Líneas Post Pago. Defaults to typer.Argument(True).
+    """
+
+    typer.echo('Obteniendo Líneas a Sincronizar')
+    accounts = get_last_updated_cdr()
+
+    cdrs = []
+    with typer.progressbar(accounts.keys(), label="Descargando CDRs") as account_data:
+        for account in account_data:
+            cdrs.extend(get_cdrs(phone_number=account, since=accounts[account]))
+
+    with typer.progressbar(cdrs, label='Sincronizando CDRs') as cdr_data:
+        with pyodbc.connect(SQL_SERVER_CNX_STRING) as cnx:
+            with cnx.cursor() as cursor:
+                for cdr in cdr_data:
+                    cursor.execute(
+                        'INSERT INTO bdd_wifi.dbo.TB_OMV_CDR_Cliente (Tipo, DDi, Fecha, \
+                        Destino, DesDestino, DuracionSegundos, DuracionFormato, Pvc, Pvp, Fuente) \
+                        VALUES(?,?,?,?,?,?,?,?,?,?)',
+                        ('VOZ', cdr['callerid'], cdr['timestamp'], cdr['destination'], cdr['description'],
+                         cdr['duration'], str(timedelta(seconds=cdr['duration'])), cdr['cost'], cdr['price'],
+                        'Magnus')
+                    )
+
+                cnx.commit()
+
+    msg = f"Se han sincronizado un total de "
+    msg = msg + typer.style(len(cdrs), fg=typer.colors.GREEN, bold=True)
+    msg = msg + " llamadas de un total de "
+    msg = msg + typer.style(len(accounts), fg=typer.colors.GREEN, bold=True)
+    msg = msg + " líneas telefónicas."
+    typer.echo(msg)
+
+
+@app.command()
+def synchronize():
+    update_plataform()
+    update_account()
+    import_cdr()
+
+if __name__ == "__main__":
+    app()

mbilling/functions.py

@@ -4,6 +4,7 @@ import re
 from random import randint
 import mariadb
 from dotenv import dotenv_values
+from datetime import datetime
 
 # Path de MBILLING
 mbapi_path = pathlib.Path(pathlib.Path().resolve(), 'mbilling')
@@ -108,7 +109,8 @@ def _create_did(phone_number: str):
         if not re.match(succesful, result.decode('utf-8')):
             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: float = 0.0):
 
     with mariadb.connect(**DB_CREDENTIALS) as cnx:
         with cnx.cursor() as cursor:
@@ -145,10 +147,11 @@ def create_user(phone_number: str, name: str, surname: str, password: str = None
     surname = surname.strip()
 
     # Lanzamos las Acciones del API
-    _create_user(phone_number.strip(), password, name, surname, plan_id, offer_id)
+    _create_user(phone_number.strip(), password,
+                 name, surname, plan_id, offer_id)
     _create_sip(phone_number, password, pbx)
     _create_did(phone_number)
-    _update_payment(phone_number,offer_id != 0, 30.00)
+    _update_payment(phone_number, offer_id != 0, 30.00)
 
     return password
 
@@ -204,3 +207,109 @@ def get_user(phone_number: str) -> dict:
             query = cursor.fetchone()
 
     return dict(zip(headers, query)) if query else None
+
+
+def fully_ported_clis() -> 'list[str]':
+    """Devuelve los CLI completamente portados a MBilling.
+
+    Algunos usuarios pueden recibir llamadas en MB pero realizar la salida\
+    por la plataforma antigua, esos usuarios deben ser omitidos para determinados\
+    procesos, por lo que esta lista devuelve aquellos Caller IDs que están\
+    completamente migrados a la nueva plataforma, tanto en salidas como llamadas.
+
+    Returns:
+        list[str]: Listado de los CLIs completamente portados.
+    """
+
+    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;
+                """)
+
+            query = [cli[0] for cli in cursor.fetchall()]
+
+    return query
+
+
+def active_users() -> 'list[dict]':
+    """Devuelve el listado de los Usuarios Activos
+
+    Returns:
+        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"""
+            )
+
+            query = [dict(zip(headers, account))
+                     for account in cursor.fetchall()]
+
+    return query
+
+def get_cdrs(phone_number : str, since : datetime = None, upto : datetime = None) -> 'list[dict]':
+    """Obtiene los CDRs de una Línea
+
+    Args:
+        phone_number (str): Número de Teléfono
+        since (datetime, optional): Fecha desde la que buscar. Defaults to None.
+        upto (datetime, optional): Fecha hasta la que buscar. Defaults to None.
+
+    Returns:
+        list[dict]: _description_
+    """
+
+    with mariadb.connect(**DB_CREDENTIALS) as cnx:
+        with cnx.cursor() as cursor:
+            if upto and since:
+                cursor.execute(
+                    'SELECT callerid, calledstation dest, destination as description, starttime date, sessiontime duration, ROUND(sessionbill,3) \
+                    price, ROUND(buycost,3) cost FROM mbilling.pkg_cdr c INNER JOIN mbilling.pkg_prefix as p ON p.id = id_prefix \
+                    WHERE callerid = ? AND starttime > ? AND starttime <= ?',
+                    (phone_number, since, upto)
+                )
+            elif since:
+                cursor.execute(
+                    'SELECT callerid, calledstation dest, destination as description, starttime date, sessiontime duration, ROUND(sessionbill,3) \
+                    price, ROUND(buycost,3) cost FROM mbilling.pkg_cdr INNER JOIN mbilling.pkg_prefix as p ON p.id = id_prefix\
+                    WHERE callerid = ? AND starttime > ?',
+                    (phone_number, since)
+                )
+            elif upto:
+                cursor.execute(
+                    'SELECT callerid, calledstation dest, destination as description, starttime date, sessiontime duration, ROUND(sessionbill,3) \
+                    price, ROUND(buycost,3) cost FROM mbilling.pkg_cdr INNER JOIN mbilling.pkg_prefix as p ON p.id = id_prefix \
+                    WHERE callerid = ? AND starttime <= ?',
+                    (phone_number, upto)
+                )
+            else:
+                cursor.execute(
+                    'SELECT callerid, calledstation dest, destination as description, starttime date, sessiontime duration, ROUND(sessionbill,3) \
+                    price, ROUND(buycost,3) cost FROM mbilling.pkg_cdr INNER JOIN mbilling.pkg_prefix as p ON p.id = id_prefix \
+                    WHERE callerid = ?',
+                    (phone_number,)
+                )
+
+            data = cursor.fetchall()
+
+    cdr = []
+    headers = ['callerid', 'destination', 'description', 'timestamp', 'duration', 'price', 'cost']
+    for line in data:
+        cdr.append(dict(zip(headers, line))) 
+
+    return cdr