Modificación de API

.vscode/launch.json

@@ -9,11 +9,20 @@
       "type": "python",
       "request": "launch",
       "module": "uvicorn",
-      "args": ["main:app", "--host",  "0.0.0.0" ,"--port", "8080"],
+      "args": ["main:app", "--host", "0.0.0.0", "--port", "8080"],
       "console": "integratedTerminal",
       "justMyCode": true
     },
-      {
+    {
+      "name": "API v2",
+      "type": "python",
+      "request": "launch",
+      "module": "uvicorn",
+      "args": ["api:app", "--host", "0.0.0.0", "--port", "8080"],
+      "console": "integratedTerminal",
+      "justMyCode": true
+    },
+    {
       "name": "CLI: Sincronizar",
       "type": "python",
       "request": "launch",

api.py

+from enum import Enum
+from fastapi import Body, Depends, FastAPI, HTTPException, Query, Response
+from fastapi.responses import JSONResponse
+from typing import Optional, Tuple
+from pydantic import BaseModel
+
+import mbilling.functions
+
+description = """
+Servicio API para la interacción con el Tarficador de Telefonía ☎️
+
+### Changelog
+
+
+
+"""
+
+tags_metadata = [
+    {
+        "name": "Account",
+        "description": "Consulta, creación y modificación de cuentas SIP y sus atributos.",
+    },
+]
+
+app = FastAPI(
+    title='API Magnus Billing',
+    description=description,
+    version='0.0.1 Beta',
+    contact={
+        'name': 'Javier Ramírez',
+        'email': 'javier@rentelwifi.com'
+    },
+    openapi_tags=tags_metadata
+)
+
+
+class Rate(str, Enum):
+    ESTANDAR = "Estandar"
+    EMPRESAS = "Empresas"
+    PLANA_1000 = "Plana1000"
+    PLANA_2000 = "Plana2000"
+    PLANA_EUROPA_1000 = "PlanaEuropa1000"
+
+
+class Tax(str, Enum):
+    GENERAL = 'General'
+    EXENTA = 'Exento'
+
+
+class Refill(str, Enum):
+    AUTOMATICA = 'Automatica'
+    LIQUIDACION = 'Liquidacion'
+    MANUAL = 'Manual'
+    TRASPASO = 'Traspaso'
+    INICIAL = 'Inicial'
+
+
+class PeerDetails(BaseModel):
+
+    username: str = Body(
+        ..., title='username', description='Nombre de Usuario'
+    )
+
+    secret: str = Body(
+        ..., title='Secret', description='Contraseña'
+    )
+
+    fromuser: str = Body(
+        ..., title='fromuser', description='Usuario SIP'
+    )
+
+    careinvite: bool = Body(
+        default=False, title='CareInvite', description='Comportamiento Invite, por defecto False.'
+    )
+
+    natforce: str = Body(
+        default='force_rport,comedia', title='NAT', description='Network Address Translation'
+    )
+
+    qualify: str = Body(
+        default='yes', title='Qualify', description='SIP Ping a Central'
+    )
+
+    type: str = Body(
+        default='friend', title='Type', description='Tipo de Conexión'
+    )
+
+    dtmfmode: str = Body(
+        default='RFC2833', title='DTMF', description='Tipo de Señalización'
+    )
+
+    insecure: str = Body(
+        default='port', title='Insecure', description='Permite autentificación relajada si coincide IP'
+    )
+
+    host: str = Body(
+        default='192.168.2.84', title='Host', description='Dirección del Tarificador'
+    )
+
+    disallow: str = Body(
+        default='all', title='Disallow', description='Codecs no Permitidos'
+    )
+
+    allow: str = Body(
+        default='gsm,opus,alaw,g729', title='Allow', description='Códecs Permitidos'
+    )
+
+
+class UserDetails(BaseModel):
+    username: str = Body(
+        ..., title='username', description='Nombre de Usuario'
+    )
+
+    secret: str = Body(
+        ..., title='Secret', description='Contraseña'
+    )
+
+    fromuser: str = Body(
+        ..., title='fromuser', description='Usuario SIP'
+    )
+
+    careinvite: bool = Body(
+        default=False, title='CareInvite', description='Comportamiento Invite, por defecto False.'
+    )
+
+    type: str = Body(
+        default='friend', title='Type', description='Tipo de Conexión'
+    )
+
+    dtmfmode: str = Body(
+        default='RFC2833', title='DTMF', description='Tipo de Señalización'
+    )
+
+    insecure: str = Body(
+        default='port', title='Insecure', description='Permite autentificación relajada si coincide IP'
+    )
+
+    host: str = Body(
+        default='192.168.2.84', title='Host', description='Dirección del Tarificador'
+    )
+
+    disallow: str = Body(
+        default='all', title='Disallow', description='Codecs no Permitidos'
+    )
+
+    allow: str = Body(
+        default='gsm,opus,alaw,g729', title='Allow', description='Códecs Permitidos'
+    )
+
+
+class SipAccount(BaseModel):
+
+    trunk_name: str = Body(
+        ..., title='Trunk', description='Nombre de la Troncal'
+    )
+
+    peer_details: PeerDetails = Body(
+        ..., title='Peer Details', description='Credenciales llamadas Salientes'
+    )
+
+    user_context: str = Body(
+        default='from-trunk', title='Contexto', description='Contexto de Llamada'
+    )
+
+    user_details: UserDetails = Body(
+        ..., title='User Details', description='Credenciales llamadas Entrantes'
+    )
+
+
+class Account(BaseModel):
+
+    username: str = Body(
+        ..., regex=r'[89][1-9]\d{7}', example='856135158', title='Número', description='Identificador de entrada y salidas de Llamadas'
+    )
+
+    name: str = Body(
+        ..., example='Rentel', title='Nombre/Den. Comercial', description='Nombre del titular o Denominación Comercial de la empresa.', max_length=64
+    )
+
+    surname: str = Body(
+        ..., example='Rentel Wifi SL', title='Apellidos/Den. Fiscal', description='Apellidos o Denominación Comercial de la empresa.', max_length=64
+    )
+
+    rate: Rate = Body(
+        default=Rate.ESTANDAR, title='Tarifa', description='Plan de Precios de la Línea'
+    )
+
+    balance: float = Body(
+        default=0.0, title='Saldo', description='Balance del Cliente'
+    )
+
+    sip_accounts : list[SipAccount] = Body(
+        ..., title='Cuentas SIP', description='Números Asociados a la Cuenta de Facturación'
+    )
+
+
+@app.get('/account/{account_name}', response_model=Account, name="Obtener Cuenta", description="Obtiene la información de una Cuenta de Facturación", tags=['Account'])
+def get_account(account_name: str = Query(..., regex=r'[89][1-9]\d{7}', name='Número', description='Cuenta SIP a recargar')):
+
+    user_data = mbilling.functions.get_user(account_name)
+
+    if user_data:
+
+        print(user_data)
+
+    else:
+        response = JSONResponse(
+            content={
+                'result': 'ERROR',
+                'description': 'No existe ningún usuario con la identificación solicitada.'
+            }, status_code=404
+        )
+
+    return response

cli.py

@@ -65,7 +65,7 @@ def update_sip_accounts() -> 'Tuple[int,int]':
                                 UsuarioEXT=?, ClaveEXT=?, NuevaPlataforma=1 WHERE DDI=? AND FechaBaja is NULL",
 
                             (credit, account.active, flat_rate, account.postpaid or flat_rate,
-                            account.plan, account.account, account.secret, account.account)
+                             account.plan, account.account, account.secret, account.account)
                         )
 
                         sip_accounts_updated += 1
@@ -165,8 +165,6 @@ def perform_migration():
                      a2billing_user['account'] if a2billing_user else None)
                 )
 
-
-
                 cnx.commit()
 
 
@@ -498,6 +496,14 @@ def import_cdr(account: str = typer.Argument(None), only_charged: bool = typer.A
 
 
 @app.command()
+def refill():
+    # Recargas
+    total_refills, amount = process_refills()
+    if total_refills > 0:
+        update_sip_accounts()
+
+
+@app.command()
 def synchronize():
     """Sincroniza el Servidor de Telefonía con el de Facturación
     """

mbilling/.env

 MAGNUS_HOST = 'http://192.168.2.84/mbilling'
+DB_HOST = '192.168.2.84'
 API_KEY = 'FxMqVFWMztbigCeZlFupGULLh6LRaLs8'
 API_SECRET = 'rguvK0sNZMki5yWtv4RKONMJo1Hd9jiX'
 DB_USER = 'scripts'
-DB_PASSWORD = 'scripts'
\ No newline at end of file
+DB_PASSWORD = 'scripts'
+DB_NAME = 'mbilling'
\ No newline at end of file

mbilling/functions.py

@@ -279,6 +279,7 @@ def fully_ported_users() -> 'list[tuple]':
 
     return data
 
+
 def get_dids(account : str) -> list:
     with mariadb.connect(**DB_CREDENTIALS) as cnx:
         with cnx.cursor() as cursor:
@@ -293,6 +294,7 @@ def get_dids(account : str) -> list:
             
     return dids
 
+
 def get_plan_consumption(*account: str) -> dict:
 
     query = 'SELECT u.username user, ROUND(SUM(c.sessiontime)/60, 0) consumption, o.freetimetocall max FROM mbilling.pkg_user u\