Nuevo Conector

mbilling/connector.py

@@ -4,6 +4,7 @@ import pathlib
 import re
 from dotenv import dotenv_values
 from sqlalchemy.ext.automap import automap_base
+import mbilling.models as models
 
 # CONFIGURACIÓN
 # Path de MBILLING
@@ -29,5 +30,106 @@ User = Base.classes.pkg_user
 Sip = Base.classes.pkg_sip
 
 
+def get_user_account(number: str):
+    """Devuelve la Cuenta de Facturación con mismo número
 
+    Args:
+        number (str): Número de Teléfono
 
+    Returns:
+        _type_: _description_
+    """
+    user = session.execute(select(User).where(
+        User.username == number)).fetchone()
+    return user[0] if user else None
+
+
+def get_sip_account(number: str):
+    """Obtiene los detalles de una Cuenta SIP
+
+    Args:
+        number (str): Número de Teléfono
+
+    Returns:
+        models.SipAccount: Cuenta SIP Asociada
+    """
+
+    sip = session.execute(select(Sip).where(Sip.name == number)).fetchone()
+    return sip[0] if sip else None
+
+
+def get_sip_credentials(sip: Sip) -> models.SipAccount:
+    """Genera los Credenciales de una Cuenta SIP
+
+    Args:
+        sip (Sip): Cuenta SIP
+
+    Returns:
+        models.SipAccount: Credenciales de Cuenta
+    """
+
+    peer_details = models.PeerDetails(
+        username=sip.name,
+        secret=sip.secret,
+        fromuser=sip.name,
+        nat=sip.nat,
+        qualify=sip.qualify,
+        type=sip.type,
+        dtmfmode=sip.dtmfmode,
+        insecure=sip.insecure,
+        host=sip.host,
+        allow=sip.allow
+    )
+
+    user_details = models.UserDetails(
+        username=sip.name,
+        secret=sip.secret,
+        fromuser=sip.name,
+        type=sip.type,
+        dtmfmode=sip.dtmfmode,
+        insecure=sip.insecure,
+        host=sip.host,
+        allow=sip.allow
+    )
+
+    return models.SipAccount(
+        trunk_name=sip.name,
+        peer_details=peer_details,
+        user_context='from-trunk',
+        user_details=user_details
+    )
+
+
+def find_user_account(number: str):
+
+    user = get_user_account(number)
+
+    if not user:
+        sip = get_sip_account(number)
+        if sip:
+            user = get_user_account(sip.accountcode)
+
+    return user
+
+
+def get_user(number: str) -> models.User:
+
+    user = None
+    user_account = find_user_account(number)
+
+    if user_account:
+        user_sip = session.execute(
+            select(Sip).where(Sip.id_user == user_account.id)
+        ).fetchall()
+        sip_accounts = [get_sip_credentials(sip[0]) for sip in user_sip]
+
+        user = models.User(
+            username=user_account.username,
+            name=user_account.firstname,
+            surname=user_account.lastname,
+            rate=models.Rate.EMPRESAS,
+            balance=user_account.credit,
+            sip_accounts=sip_accounts
+        )
+
+    return user

mbilling/models.py

+from enum import Enum
+from fastapi import Body
+
+from pydantic import BaseModel
+
+
+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'
+    )
+
+
+    nat: 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'
+    )
+
+    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 User(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'
+    )