Nueva API

api.py

@@ -3,10 +3,9 @@ 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
 import mbilling.models as models
 import mbilling.connector as mbc
+import mbilling.exceptions as mbe
 description = """
 Servicio API para la interacción con el Tarficador de Telefonía ☎️
 
@@ -35,11 +34,46 @@ app = FastAPI(
 )
 
 
-@app.get('/account/{account_name}', response_model=models.User, name="Obtener Cuenta", description="Obtiene la información de una Cuenta de Facturación", tags=['Account'])
+@app.get('/account/{account_name}', response_model=models.User, 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 = mbc.get_user(account_name)
     if not user:
-        raise HTTPException(status_code=404, detail="No existe ningún usuario que contenga el número.")
+        raise HTTPException(
+            status_code=404, detail="No existe ningún usuario que contenga el número.")
     else:
         return user
+
+
+@app.post('/account', response_model=models.User, name="Crea una Cuenta de Facturación",
+          description="Agrega una nueva Cuenta de Facturación y las cuentas SIP al sistema.", tags=['Account'],
+          status_code=201)
+def post_account(user_data: models.CreateUser):
+
+    try:
+        return mbc.create_user(user_data.username, user_data.name,
+                        user_data.surname, user_data.rate, user_data.address)
+
+    except mbe.AccountExistsError:
+        raise HTTPException(
+            status_code=409, detail='Ya existe una cuenta con este nombre de usuario.')
+
+    except mbe.SipRegisteredError:
+        raise HTTPException(
+            status_code=409, detail='Existe una cuenta SIP con esta numeración asignada a otro usuario.')
+
+
+@app.put('/account/{account_name}', response_model=models.User, name="Modificar Cuenta", description="Modifica la información de una Cuenta de Facturación",
+         tags=['Account'], status_code=200)
+def modify_account(account_name : str, user_data : models.CreateUser):
+    
+    try:
+        return mbc.update_user(account_name, user_data)
+    except mbe.UserNotFound:
+        raise HTTPException(status_code=404, detail='No existe ningún usuario con esa numeración.')
+
+        
+
+
+

mbilling/connector.py

-from sqlalchemy import create_engine, select
+from ast import Tuple
+from sqlalchemy import create_engine, select, update
 from sqlalchemy.orm import sessionmaker
 import pathlib
 import re
 from dotenv import dotenv_values
 from sqlalchemy.ext.automap import automap_base
 import mbilling.models as models
+import mbilling.functions as cli
+import mbilling.exceptions as mbe
 
 # CONFIGURACIÓN
 # Path de MBILLING
@@ -30,7 +33,52 @@ User = Base.classes.pkg_user
 Sip = Base.classes.pkg_sip
 
 
-def get_user_account(number: str):
+def decode_rate(rate: models.Rate = models.Rate.ESTANDAR):
+    """Genera los ID de plan y oferta.
+
+    Args:
+        rate (Rate, optional): Tarifa Aplicada. Defaults to models.Rate.ESTANDAR.
+
+    Returns:
+        Tuple[int, int]: Códigos de Plan y Oferta que generan la tarifa.
+    """
+    # Tarifa por Defecto
+    plan_id = 1
+    offer_id = 0
+
+    # Selección de Tarifa
+    if rate == models.Rate.PLANA_1000:
+        plan_id = 4
+        offer_id = 4
+    elif rate == models.Rate.PLANA_2000:
+        plan_id = 4
+        offer_id = 5
+    elif rate == models.Rate.EMPRESAS:
+        plan_id = 2
+
+    return (plan_id, offer_id)
+
+
+def encode_rate(plan_id: int, offer_id: int = 0) -> models.Rate:
+    """Obtiene la tarifa en base a un plan y oferta de MagnusBilling
+
+    Args:
+        plan_id (int): ID del Plan
+        offer_id (int, optional): ID de la Oferta. Defaults to 0.
+
+    Returns:
+        models.Rate: Tarifa a Aplicar
+    """
+
+    if plan_id == 1:
+        return models.Rate.ESTANDAR
+    elif plan_id == 2:
+        return models.Rate.EMPRESAS
+    elif plan_id == 4:
+        return models.Rate.PLANA_1000 if offer_id == 4 else models.Rate.PLANA_2000
+
+
+def get_user_account(number: str) -> User:
     """Devuelve la Cuenta de Facturación con mismo número
 
     Args:
@@ -41,6 +89,7 @@ def get_user_account(number: str):
     """
     user = session.execute(select(User).where(
         User.username == number)).fetchone()
+
     return user[0] if user else None
 
 
@@ -100,7 +149,7 @@ def get_sip_credentials(sip: Sip) -> models.SipAccount:
     )
 
 
-def find_user_account(number: str):
+def find_user_account(number: str) -> User:
 
     user = get_user_account(number)
 
@@ -133,3 +182,60 @@ def get_user(number: str) -> models.User:
         )
 
     return user
+
+
+def create_user(number: str, name: str, surname: str, rate: models.Rate = models.Rate.ESTANDAR, pbx: str = None) -> models.User:
+
+    user_account = find_user_account(number)
+
+    if not user_account:
+        plan_id, offer_id = decode_rate(rate)
+        cli.create_user(
+            phone_number=number,
+            name=name,
+            surname=surname,
+            pbx=pbx,
+            plan_id=plan_id,
+            offer_id=offer_id
+        )
+    elif user_account.username == number:
+        raise mbe.AccountExistsError()
+    else:
+        raise mbe.SipRegisteredError()
+
+
+def update_user(username: str, user_data: models.User):
+    user = find_user_account(user_data.username)
+
+    if user:
+        if user_data.name:
+            session.execute(
+                update(User).where(username=username).values(
+                    name=user_data.name)
+            )
+
+        if user_data.address:
+            session.execute(
+                update(User).where(username=username).values(
+                    host=user_data.address)
+            )
+
+        if user_data.rate:
+            plan_id, offer_id = decode_rate(user_data.rate)
+            session.execute(
+                update(User).where(username=username).values(
+                    id_plan=plan_id, id_offer=offer_id)
+            )
+
+        if user_data.username:
+            session.execute(
+                update(User).where(username=username).values(
+                    username=user_data.username)
+            )
+
+        session.commit()
+
+        return get_user(username)
+
+    else:
+        raise mbe.UserNotFound()

mbilling/create_user.php

@@ -19,6 +19,7 @@ if (isset($argv[1]) && isset($argv[2]) && isset($argv[3]) && isset($argv[4]) &&
             'id_group' => 3,
             'id_plan' => $argv[5],
             'id_offer' => $argv[6],
+            'language' => 'es',
             'credit' => 0,
         ]);
 

mbilling/exceptions.py

+class AccountExistsError(Exception):
+    pass
+
+class SipRegisteredError(Exception):
+    pass
+
+class UserNotFound(Exception):
+    pass
\ No newline at end of file

mbilling/models.py

@@ -3,6 +3,8 @@ from fastapi import Body
 
 from pydantic import BaseModel
 
+IP_REGEX = r"^[((25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[1-9]?[0-9])\.){3}(25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[1-9]?[0-9])|dynamic]$"
+
 
 class Rate(str, Enum):
     ESTANDAR = "Estandar"
@@ -25,6 +27,29 @@ class Refill(str, Enum):
     INICIAL = 'Inicial'
 
 
+class CreateUser(BaseModel):
+
+    username: str = Body(
+        ..., regex=r'[89][1-9]\d{7}', example='856135158', title='Número Principal', 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'
+    )
+
+    address: Rate = Body(
+        ..., regex=IP_REGEX, example='192.168.2.71', title='Dirección IP', description='Dirección estática del usuario, sólo si es PBX.'
+    )
+
+
 class PeerDetails(BaseModel):
 
     username: str = Body(
@@ -39,7 +64,6 @@ class PeerDetails(BaseModel):
         ..., title='fromuser', description='Usuario SIP'
     )
 
-
     nat: str = Body(
         default='force_rport,comedia', title='NAT', description='Network Address Translation'
     )