Skip to content

S
scrapper-portabilidad
Commit f5340174 by Javier
Options

Modificación y Generación de SIP

parent 8879e81f
Showing with 274 additions and 101 deletions
api.py
from enum import Enum
from fastapi import Body, Depends, FastAPI, HTTPException, Query, Response
from fastapi import Body, Depends, FastAPI, HTTPException, Query, Response, Request
from fastapi.responses import JSONResponse
from typing import Optional, Tuple
from pydantic import BaseModel
import mbilling.models as models
import mbilling.connector as mbc
import mbilling.exceptions as mbe
from mbilling.functions import MagnusCommandError
description = """
Servicio API para la interacción con el Tarficador de Telefonía ☎️
......@@ -18,8 +18,12 @@ Servicio API para la interacción con el Tarficador de Telefonía ☎️
tags_metadata = [
{
"name": "Account",
"description": "Consulta, creación y modificación de cuentas SIP y sus atributos.",
"description": "Consulta, creación y modificación de cuentas de facturación y sus atributos.",
},
{
"name": "SIP",
"description": "Consulta, creación y modificación de cuentas SIP y sus atributos."
}
]
app = FastAPI(
......@@ -34,6 +38,16 @@ app = FastAPI(
)
@app.exception_handler(MagnusCommandError)
def magnus_error_handler(request: Request, ex: MagnusCommandError):
return JSONResponse(
status_code=500,
content={
"detail": f'Se ha producido un error en Magnus Billing. ({str(ex)})',
}
)
@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')):
......@@ -49,7 +63,7 @@ def get_account(account_name: str = Query(..., regex=r'[89][1-9]\d{7}', name='N
@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):
def add_account(user_data: models.CreateUser):
try:
return mbc.create_user(user_data.username, user_data.name,
......@@ -66,14 +80,25 @@ def post_account(user_data: models.CreateUser):
@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):
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.')
raise HTTPException(
status_code=404, detail='No existe ningún usuario con esa numeración.')
@app.post('/account/{account_name}/sip', response_model=models.SipAccount, name="Generar cuenta SIP", description="Agrega una nueva cuenta SIP al sistema",
tags=["SIP"], status_code=201)
def add_sip(sip: models.CreateSipAccount, account_name: str = Query(..., regex=models.PHONE_REGEX, name='Número', description='Cuenta SIP a recargar'),):
if not mbc.find_user_account(sip.cid):
try:
mbc.create_sip(account_name, sip)
except mbe.UserNotFound:
raise HTTPException(
status_code=404, detail='No existe ningún usuario con el número de cuenta indicado.')
else:
raise HTTPException(
status_code=409, detail='La cuenta SIP ya está asignada a otro usuario.')
mbilling/assign_did.php
......@@ -3,13 +3,13 @@
include 'api.php';
// ARGUMENTOS : SIP DID
if (isset($argv[1]) && isset($argv[2])) {
// ARGUMENTOS : USER SIP DID
if (isset($argv[1]) && isset($argv[2]) && isset($argv[3])) {
$result = $magnusBilling->create('diddestination', [
'id_user' => $magnusBilling->getId('user', 'username', $argv[1]),
'id_sip' => $magnusBilling->getId('sip', 'name', $argv[1]),
'id_did' => $magnusBilling->getId('did', 'DID', $argv[2]),
'id_sip' => $magnusBilling->getId('sip', 'name', $argv[2]),
'id_did' => $magnusBilling->getId('did', 'DID', $argv[3]),
'voip_call' => '1'
]);
......@@ -19,7 +19,6 @@ if (isset($argv[1]) && isset($argv[2])) {
print_r($result);
print("[409] ERROR Ya existe una cuenta con mismo usuario.");
}
} else {
print("[404] ERROR No se han introducido todos los parámetros necesarios.");
}
mbilling/connector.py
from ast import Tuple
from sqlalchemy import create_engine, select, update
from sqlalchemy.orm import sessionmaker
import pathlib
......@@ -8,6 +7,7 @@ from sqlalchemy.ext.automap import automap_base
import mbilling.models as models
import mbilling.functions as cli
import mbilling.exceptions as mbe
from time import sleep
# CONFIGURACIÓN
# Path de MBILLING
......@@ -31,6 +31,7 @@ session = Session()
# Carga de Tablas
User = Base.classes.pkg_user
Sip = Base.classes.pkg_sip
Did = Base.classes.pkg_did
def decode_rate(rate: models.Rate = models.Rate.ESTANDAR):
......@@ -93,7 +94,7 @@ def get_user_account(number: str) -> User:
return user[0] if user else None
def get_sip_account(number: str):
def get_sip_account(number: str) -> Sip:
"""Obtiene los detalles de una Cuenta SIP
Args:
......@@ -102,55 +103,22 @@ def get_sip_account(number: str):
Returns:
models.SipAccount: Cuenta SIP Asociada
"""
return session.query(Sip).filter_by(name=number).first()
sip = session.execute(select(Sip).where(Sip.name == number)).fetchone()
return sip[0] if sip else None
def find_user_account(number: str) -> User:
"""Encuentra a un usuario en base a un número de teléfono
def get_sip_credentials(sip: Sip) -> models.SipAccount:
"""Genera los Credenciales de una Cuenta SIP
Devolverá al usuario cuya cuenta sea igual al número de teléfono o \
tenga asociado este mediante una cuenta SIP.
Args:
sip (Sip): Cuenta SIP
number (str): Número de Teléfono.
Returns:
models.SipAccount: Credenciales de Cuenta
User: Cuenta de Facturación.
"""
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:
user = get_user_account(number)
if not user:
......@@ -162,6 +130,14 @@ def find_user_account(number: str) -> User:
def get_user(number: str) -> models.User:
"""Devuelve al usuario cuya cuenta sea igual al número indicado.
Args:
number (str): Número de Teléfono.
Returns:
models.User: Cuenta de Facturación.
"""
user = None
user_account = find_user_account(number)
......@@ -170,7 +146,8 @@ def get_user(number: str) -> models.User:
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]
sip_accounts = [models.SipAccount.generate_from_automap(
sip[0]) for sip in user_sip]
user = models.User(
username=user_account.username,
......@@ -185,6 +162,22 @@ def get_user(number: str) -> models.User:
def create_user(number: str, name: str, surname: str, rate: models.Rate = models.Rate.ESTANDAR, pbx: str = None) -> models.User:
"""Crea una nueva Cuenta de Facturación
Args:
number (str): Número Principal
name (str): Nombre o Denomimación Comercial
surname (str): Apellidos o Denominación Fiscal
rate (models.Rate, optional): Tarifa. Defaults to models.Rate.ESTANDAR.
pbx (str, optional): Dirección IP de la Centralita. Defaults to None.
Raises:
mbe.AccountExistsError: El usuario ya está registrado.
mbe.SipRegisteredError: Existe una cuenta SIP con el número asignada a otro usuario.
Returns:
models.User: Cuenta de Facturación.
"""
user_account = find_user_account(number)
......@@ -205,6 +198,19 @@ def create_user(number: str, name: str, surname: str, rate: models.Rate = models
def update_user(username: str, user_data: models.User):
"""Actualiza una Cuenta de Facturación
Args:
username (str): Número Principal
user_data (models.User): Cuenta Actualizada
Raises:
mbe.UserNotFound: No existe ningún usuario con el número indicado.
Returns:
models.User: Cuenta de Facturación Actualizada
"""
user = find_user_account(user_data.username)
if user:
......@@ -239,3 +245,89 @@ def update_user(username: str, user_data: models.User):
else:
raise mbe.UserNotFound()
def get_dids_by_number(number: str):
"""Obtiene las DIDs asignados a un Número
Args:
number (str): Número de Destino
Returns:
list: Listado de DIDs
"""
return session.query(Did).filter(Did.did.like(f'%{number}')).all()
def update_did_owner(did: Did, new_owner: User):
"""Modifica el propietario de un DID
Args:
did (Did): DID a Modificar
new_owner (User): Nuevo Propietario
"""
session.execute(
update(Did).where(id=did.id).values(
id_user=new_owner.id, activated=True, reserved=True)
)
session.execute(
update(Base.classes.pkg_did_use).where(id_did=did.id).values(
id_user=new_owner.id
)
)
session.execute(
update(Base.classes.pkg_did_destination).where(id_did=did.id).values(
id_user=new_owner.id
)
)
session.commit()
def create_or_update_did(number: str, user: User, sip: Sip = None):
if not sip:
sip = get_sip_account(number)
if not sip:
raise mbe.SipRegisteredError()
for did_number in (number, f'34{number}'):
did = session.execute(select(Did).where(Did.did == did_number)).first()
if did:
update_did_owner(did, user)
else:
cli._create_did(did_number, user.username,
auto_international=False)
def create_sip(username: str, create_sip: models.CreateSipAccount) -> models.SipAccount:
user_account = find_user_account(username)
if not user_account:
raise mbe.UserNotFound()
if find_user_account(create_sip.cid):
raise mbe.SipRegisteredError()
# Se generan las cuentas SIP
cli._create_sip(create_sip.cid, create_sip.secret,
create_sip.host, username)
session.commit()
# Se generan o asignan los DID
sip = get_sip_account(create_sip.cid)
if sip:
create_or_update_did(create_sip.cid, user_account, sip)
return models.SipAccount.generate_from_automap(sip)
else:
raise mbe.SipRegisteredError(
'La cuenta SIP no se ha generado correctamente.')
def test_function(test_number):
print(models.SipAccount.generate_from_automap(
get_sip_account(test_number)))
mbilling/create_sip.php
......@@ -3,20 +3,19 @@
include 'api.php';
// ARGUMENTOS: USER PASSWORD PBX
if (isset($argv[1]) && isset($argv[2])) {
if (isset($argv[1]) && isset($argv[2]) && isset($argv[3]) && isset($argv[4])) {
$data = [
"id_user" => $magnusBilling->getId('user', 'username', $argv[1]),
"id_user" => $magnusBilling->getId('user', 'username', $argv[4]),
"name" => $argv[1],
"defaultuser" => $argv[1],
"callerid" => $argv[1],
"secret" => $argv[2]
"secret" => $argv[2],
"host" => $argv[3],
"insecure" => "invite,port",
"allow" => "g729,gsm,opus,alaw,ulaw"
];
if(isset($argv[3])) {
$data["host"] = $argv[3];
$data["insecure"] = "invite,port";
}
$sip_id = $magnusBilling->getId('sip', 'name', $argv[1]);
if ($sip_id == null) {
......@@ -31,7 +30,6 @@ if (isset($argv[1]) && isset($argv[2])) {
} else {
print_r($result);
}
} else {
print("[404] ERROR No se han introducido todos los parámetros necesarios.");
}
mbilling/exceptions.py
......@@ -6,3 +6,6 @@ class SipRegisteredError(Exception):
class UserNotFound(Exception):
pass
class DidNotFound(Exception):
pass
\ No newline at end of file
mbilling/functions.py
......@@ -12,6 +12,8 @@ mbapi_path = pathlib.Path(pathlib.Path().resolve(), 'mbilling')
# Petición Correcta
succesful = re.compile(r'^\[20\d\]')
error_extraction = re.compile(
r'.*<h1>Internal Server Error<\/h1>.*<h2>(.*)<\/h2>.*', re.MULTILINE | re.DOTALL | re.IGNORECASE)
# Carga de Variables ENV
values = dotenv_values(pathlib.Path(mbapi_path, '.env'))
......@@ -29,8 +31,8 @@ class SipAccount:
def __init__(self, callerid: str, account: str, secret: str,
firstname: str, lastname: str, credit: float,
plan: int, offer: int, active: bool, postpaid: bool,
dtmf: str, insecure: str, nat: str, qualify : str,
type: str, disallow: str, allow : str, host: str = None,
dtmf: str, insecure: str, nat: str, qualify: str,
type: str, disallow: str, allow: str, host: str = None,
dids: list = None) -> None:
self.callerid = callerid
......@@ -51,11 +53,16 @@ class SipAccount:
self.disallow = disallow
self.allow = allow
class MagnusCommandError(Exception):
def __init__(self, *args: object, ) -> None:
super().__init__(*args)
pass
try:
error = re.match(error_extraction, args[0]).group(1)
except AttributeError:
error = args[0]
super().__init__(error)
def _create_user(phone_number: str, password: str, name: str, surname: str, plan_id: int = 1, offer_id: int = 0):
......@@ -86,7 +93,7 @@ def _create_user(phone_number: str, password: str, name: str, surname: str, plan
f'Command: {exception_cmd} Exception: {result.decode(encoding="utf-8")}')
def _create_sip(phone_number: str, password: str, pbx: str = None):
def _create_sip(phone_number: str, password: str, pbx: str = None, username: str = None):
"""Agrega una nueva cuenta SIP y la agrega al usuario con el mismo nombre.
Args:
......@@ -99,7 +106,8 @@ def _create_sip(phone_number: str, password: str, pbx: str = None):
"""
command = ['php', pathlib.Path(mbapi_path, 'create_sip.php'),
phone_number, password]
phone_number, password, pbx if pbx else 'dynamic',
username if username else phone_number]
if pbx:
command.append(pbx)
......@@ -109,7 +117,7 @@ def _create_sip(phone_number: str, password: str, pbx: str = None):
raise MagnusCommandError(result.decode(encoding='utf-8'))
def _create_did(phone_number: str):
def _create_did(phone_number: str, username: str = None, sip: str = None, auto_international: bool = True):
"""Crea un DID tanto para el número como para su versión internacional y la asigna\
al usuario cuya identificación sea el nombre.
......@@ -121,7 +129,16 @@ def _create_did(phone_number: str):
MagnusCommandError: Error en la Insercción del DID internacional.
"""
for did in (phone_number, f'34{phone_number}'):
dids = (phone_number, f'34{phone_number}') if auto_international else (
phone_number,)
if not username:
username = phone_number
if not sip:
sip = username
for did in dids:
result = subprocess.check_output(
['php', pathlib.Path(mbapi_path, 'create_did.php'), did]
)
......@@ -131,7 +148,7 @@ def _create_did(phone_number: str):
result = subprocess.check_output(
['php', pathlib.Path(
mbapi_path, 'assign_did.php'), phone_number, did]
mbapi_path, 'assign_did.php'), username, sip, did]
)
if not re.match(succesful, result.decode('utf-8')):
......@@ -280,7 +297,7 @@ def fully_ported_users() -> 'list[tuple]':
return data
def get_dids(account : str) -> list:
def get_dids(account: str) -> list:
with mariadb.connect(**DB_CREDENTIALS) as cnx:
with cnx.cursor() as cursor:
cursor.execute(
......
mbilling/models.py
from enum import Enum
from typing import Optional
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]$"
IP_REGEX = r"^((?:[0-9]{1,3}\.){3}[0-9]{1,3})|(dynamic)$"
PHONE_REGEX = r"[89][1-9]\d{7}"
class Rate(str, Enum):
......@@ -30,7 +32,7 @@ class Refill(str, Enum):
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'
..., regex=PHONE_REGEX, example='856135158', title='Número Principal', description='Identificador de entrada y salidas de Llamadas'
)
name: str = Body(
......@@ -153,11 +155,66 @@ class SipAccount(BaseModel):
..., title='User Details', description='Credenciales llamadas Entrantes'
)
@staticmethod
def generate_from_automap(sip):
peer_details = 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 = 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 SipAccount(
trunk_name=sip.name,
peer_details=peer_details,
user_context='from-trunk',
user_details=user_details
)
class CreateSipAccount(BaseModel):
cid: str = Body(
..., title='Caller ID', description='Número de Teléfono'
)
secret: Optional[str] = Body(
default=None, title='Contraseña', description='Credencial de Inicio de Sesión'
)
host: Optional[str] = Body(
default=None, title='Dirección del Host', description='Inicio de Sesión mediante dirección IP.'
)
class UpdateSipAccount(CreateSipAccount):
username: Optional[str] = Body(
default=None, title='Nuevo Propietario', description='Cuenta benefactora del cambio de titular.'
)
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'
..., regex=PHONE_REGEX, example='856135158', title='Número', description='Identificador de entrada y salidas de Llamadas'
)
name: str = Body(
......
test.py
import mariadb as mdb
from a2billing.a2billing import account_data
import os
import mbilling.models as models
import mbilling.connector as mbc
import mbilling.exceptions as mbe
DB_CREDENTIALS = {
'host': '192.168.2.84',
'user': 'scripts',
'password': 'scripts',
'port': 3306,
'database': 'mbilling'
}
with mdb.connect(**DB_CREDENTIALS) as cnx:
with cnx.cursor() as cursor:
cursor.execute(
'SELECT id_prefix, rateinitial FROM mbilling.pkg_rate WHERE id_plan=4;'
)
for prefix in cursor.fetchall():
cursor.execute(
'INSERT INTO pkg_rate (id_plan, id_trunk_group, id_prefix, rateinitial) VALUES (?,?,?,?)',
(5,1,prefix[0], prefix[1])
)
cnx.commit()
mbc.test_function('856135158')
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment