Skip to content

S
scrapper-portabilidad
Commit c9a51610 by Javier
Options

API

parent e818e70e
Showing with 240 additions and 303 deletions
.vscode/launch.json 0 → 100644
{
// Use IntelliSense to learn about possible attributes.
// Hover to view descriptions of existing attributes.
// For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
"version": "0.2.0",
"configurations": [
{
"name": "API",
"type": "python",
"request": "launch",
"module": "uvicorn",
"args": [
"main:app"
],
"jinja": true,
"justMyCode": true
}
]
}
\ No newline at end of file
geckodriver.log deleted 100644 → 0
This source diff could not be displayed because it is too large. You can view the blob instead.
main.py
from fastapi import FastAPI from enum import Enum
import imp
from lib2to3.pytree import Base
from tkinter.tix import Form
from urllib import response
from fastapi import Body, Depends, FastAPI
from fastapi.responses import JSONResponse
from typing import Optional
from pydantic import BaseModel
import mbilling.functions
app = FastAPI() app = FastAPI()
@app.get("/") class Rate(str, Enum):
async def root(): ESTANDAR = "Tarifa Estándar"
return {"message": "Hello World"} EMPRESAS = "Tarifa Empresas"
PLANA_1000 = "Tarifa Plana 1000 Minutos"
PLANA_2000 = "Tarifa Plana 2000 Minutos"
class Account(BaseModel):
"""_summary_
Args:
BaseModel (_type_): _description_
"""
number: 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'
)
pbx_address: Optional[str] = Body(
default=None, example='192.168.2.71', title='Centralita', description='Dirección IP de la Centralita.'
)
class ConnectionData(BaseModel):
"""Datos de Conexión
"""
account : str = Body(
..., title='Cuenta', description='Usuario/Número de cuenta para Autentificación SIP.'
)
password : str = Body(
..., title='Contraseña', description='Credencial para la Autentificación.'
)
peer_details : str = Body(
default=None, title='Peer Details', description='Credenciales para la salida por troncal en Centralitas.'
)
user_details : str = Body(
default=None, title='User Context', description='Credenciales para la entrada por troncal en Centralitas.'
)
class Config:
schema_extra = {
"example" : {
"account" : "856135158",
"password" : "rentel58",
"peer_details": "type = friend username = 951203503 host = 192.168.2.84 nat = force_rport, comedia qualify = yes insecure = invite, port",
"user_details": "type = friend host = 192.168.2.84 nat = force_rport, comedia qualify = yes insecure = invite, port"
}
}
@app.post('/account', response_model=ConnectionData, name='Añadir Cuenta SIP', description='Genera una nueva cuenta en el Tarificador')
def create_account( account: Account = Depends()):
"""Genera una nueva cuenta SIP con los datos proporcionados.
Args:
account (Account, optional): Modelo de Cuenta. Defaults to Depends().
"""
# Tarifa por Defecto
plan_id = 1
offer_id = 0
# Selección de Tarifa
if account.rate == Rate.PLANA_1000:
plan_id = 4
offer_id = 4
elif account.rate == Rate.PLANA_2000:
plan_id = 4
offer_id = 5
elif account.rate == Rate.EMPRESAS:
plan_id = 2
# Generación de la Cuenta
try:
password = mbilling.functions.create_user(
phone_number=account.number,
name=account.name,
surname=account.surname,
pbx=account.pbx_address,
plan_id=plan_id,
offer_id=offer_id
)
if not account.pbx_address:
response = ConnectionData(account=account.number, password=password)
else:
response = ConnectionData(
account=account.number,
password=password,
peer_details=f'type = friend\nusername = {account.number}\nhost = 192.168.2.84\nnat = force_rport, comedia\nqualify = yes\ninsecure = invite, port',
user_details='type = friend\nhost = 192.168.2.84\nnat = force_rport, comedia\nqualify = yes\ninsecure = invite, port'
)
except mbilling.functions.MagnusCommandError as e:
response = JSONResponse(
content={
'error' : 'Se ha producido un error en Magnus durante la ejecución del comando.',
'description' : str(e),
},
status_code=400
)
return response
magnus-api/.env → mbilling/.env
MAGNUS_HOST = 'http://192.168.2.84/mbilling' MAGNUS_HOST = 'http://192.168.2.56/mbilling'
API_KEY = 'FxMqVFWMztbigCeZlFupGULLh6LRaLs8' API_KEY = 'FxMqVFWMztbigCeZlFupGULLh6LRaLs8'
API_SECRET = 'rguvK0sNZMki5yWtv4RKONMJo1Hd9jiX' API_SECRET = 'rguvK0sNZMki5yWtv4RKONMJo1Hd9jiX'
\ No newline at end of file
magnus-api/create_user.php → mbilling/create_user.php
...@@ -3,11 +3,11 @@ ...@@ -3,11 +3,11 @@
include 'api.php'; include 'api.php';
// ARGUMENTOS : USERNAME PASSWORD NOMBRE APELLIDOS // ARGUMENTOS : USERNAME PASSWORD NOMBRE APELLIDOS
if (isset($argv[1]) && isset($argv[2]) && isset($argv[3]) && isset($argv[4])) { if (isset($argv[1]) && isset($argv[2]) && isset($argv[3]) && isset($argv[4]) && isset($argv[5]) && isset($argv[6])) {
$user_id = $magnusBilling->getId('user', 'username', $argv[1]); $user_id = $magnusBilling->getId('user', 'username', $argv[1]);
if (isset($user_id)) { if (isset($user_id)) {
print("[201] OK id=" . $user_id); print("[409] ERROR Ya existe una cuenta con mismo usuario.");
} else { } else {
$result = $magnusBilling->create('user', [ $result = $magnusBilling->create('user', [
'username' => $argv[1], 'username' => $argv[1],
...@@ -17,14 +17,16 @@ if (isset($argv[1]) && isset($argv[2]) && isset($argv[3]) && isset($argv[4])) { ...@@ -17,14 +17,16 @@ if (isset($argv[1]) && isset($argv[2]) && isset($argv[3]) && isset($argv[4])) {
'lastname' => strtoupper($argv[4]), 'lastname' => strtoupper($argv[4]),
'callingcard_pin' => strval(rand(1000, 9999)), 'callingcard_pin' => strval(rand(1000, 9999)),
'id_group' => 3, 'id_group' => 3,
'id_plan' => 1, 'id_plan' => $argv[5],
'id_offer' => $argv[6],
'credit' => 0, 'credit' => 0,
]); ]);
if ($result['success'] == 1) { if ($result['success'] == 1) {
print("[201] OK id=". $result['rows'][0]['id']); print("[201] OK id=". $result['rows'][0]['id']);
} else { } else {
print("[409] ERROR Ya existe una cuenta con mismo usuario."); $error = print_r($result['errors'], true);
print("[400] ".$error);
} }
} }
......
mbilling/functions.py 0 → 100644
import subprocess
import pathlib
import re
from random import randint
mbapi_path = pathlib.Path(pathlib.Path().resolve(), 'mbilling')
succesful = re.compile(r'^\[20\d\]')
class MagnusCommandError(Exception):
def __init__(self, *args: object, ) -> None:
super().__init__(*args)
pass
def _create_user(phone_number: str, password: str, name: str, surname: str, plan_id: int = 1, offer_id: int = 0):
"""Genera una cuenta de usuario en MagnusBilling
"""
result = subprocess.check_output(
['php', pathlib.Path(mbapi_path, 'create_user.php'),
phone_number, password, f'\"{name}\"', f'\"{surname}\"', str(plan_id), str(offer_id)]
)
if not re.match(succesful, result.decode(encoding='utf-8')):
exception_cmd = ' '.join(['php', str(pathlib.Path(mbapi_path, 'create_user.php')),
phone_number, password, f'\"{name}\"', f'\"{surname}\"', str(plan_id), str(offer_id)])
raise MagnusCommandError(f'Command: {exception_cmd} Exception: {result.decode(encoding="utf-8")}')
def _create_sip(phone_number: str, password: str, pbx: str = None):
command = ['php', pathlib.Path(mbapi_path, 'create_sip.php'),
phone_number, password]
if pbx:
command.append(pbx)
result = subprocess.check_output(command)
if not re.match(succesful, result.decode(encoding='utf-8')):
raise MagnusCommandError(result.decode(encoding='utf-8'))
def _create_did(phone_number: str):
"""Genera un DID
"""
for did in (phone_number, f'34{phone_number}'):
result = subprocess.check_output(
['php', pathlib.Path(mbapi_path, 'create_did.php'), did]
)
if not re.match(succesful, result.decode('utf-8')):
raise MagnusCommandError(result.decode('utf-8'))
result = subprocess.check_output(
['php', pathlib.Path(
mbapi_path, 'assign_did.php'), phone_number, did]
)
if not re.match(succesful, result.decode('utf-8')):
raise MagnusCommandError(result.decode('utf-8'))
def create_user(phone_number: str, name: str, surname: str, password: str = None, pbx: str = None, plan_id: int = 1, offer_id: int = 0, balance: float = 0.0) -> str:
"""Crea a un nuevo usuario en MagnusBilling
"""
password = password if password else ''.join(
(name.split(' ')[0].lower(), str(randint(100, 999))))
_create_user(phone_number, password, name, surname, plan_id, offer_id)
_create_sip(phone_number, password, pbx)
_create_did(phone_number)
return password
notebook/data.txt deleted 100644 → 0
rows: {"id":0,"id_user":14,"name":"","accountcode":"","regexten":"","amaflags":"","callgroup":"","callerid":"564123789","directmedia":"no","context":"billing","DEFAULTip":"","dtmfmode":"RFC2833","fromuser":"","fromdomain":"","host":"dynamic","sip_group":"","insecure":"no","language":"","mailbox":"","md5secret":"","nat":"force_rport,comedia","deny":"","permit":"","pickupgroup":"","port":"","qualify":"no","rtptimeout":"","rtpholdtimeout":"","secret":"rentel123","type":"friend","disallow":"all","allow":"g729,gsm,opus,alaw,ulaw","regseconds":null,"ipaddr":"","fullcontact":"","setvar":"","regserver":"","lastms":"","defaultuser":"564123789","auth":"","subscribemwi":"","vmexten":"","cid_number":"","callingpres":"","usereqphone":"","mohsuggest":"","allowtransfer":"no","autoframing":"","maxcallbitrate":"","outboundproxy":"","rtpkeepalive":"","useragent":"","calllimit":0,"lineStatus":"","url_events":"","ringfalse":0,"record_call":0,"voicemail":0,"forward":"","block_call_reg":"","dial_timeout":60,"techprefix":0,"alias":"","addparameter":"","amd":0,"videosupport":"no","type_forward":"","id_ivr":"","id_queue":"","id_sip":"","extension":"","voicemail_email":"","voicemail_password":692889,"sipshowpeer":""}
\ No newline at end of file
notebook/magnusbilling.py deleted 100644 → 0
from requests import Session
from enum import Enum
from requests.models import HTTPError
magnus = "http://192.168.2.56/mbilling/index.php"
login = {
'user' : 'root',
'password' : 'NgtAtONgeRaFtsbu',
}
class Tarifa(Enum):
ESTANDAR = 1
EMPRESA = 2
PTV = 4
def generate_rows(tarifa : Tarifa, telefono : str, id_cliente : int, nombre : str, apellido : str, email : str) -> str:
"""Genera cadena Row para creación de Cliente
"""
row = f'id" : 0,"id_group" : 3,"id_plan" : {tarifa.value},"id_user" : 0,"username" : {telefono},"password" : {id_cliente},"active" : 1,"credit" : 0,'
row = ''.join((row,f'"enableexpire" : 0,"expiredays" : 0,"credit_notification" : -1,"restriction" : 0,"callingcard_ping" : {id_cliente},"callshop" : 0,'))
row = ''.join((row,f'"plan_day" : 0,"active_paypal" : 0,"boleto" : 0,"lastname" : {apellido.capitalize()},"firstname" : {nombre.capitalize()},"redial" : "",'))
row = ''.join((row,f'"tag" : "","company_name" : {nombre.capitalize()},"commercial_name" : {" ".join((nombre.capitalize(), apellido.capitalize()))},"address" : "",'))
row = ''.join((row,f'"city" : "","state" : " ","country" : "1","loginkey" : "","zipcode" : "","phone" : "","mobile" : "","email" : {email},"doc" : "","vat" : "",'))
row = ''.join((row,f'"language" : "es","company_website" : "","prefix_local" : "","boleto_day" : 0,"firstusedate":null,"expirationdate":null,"lastuse":null,'))
row = ''.join((row,f'"description":"","creationdate":null,"id_group_agent":0,"calllimit":-1,"mix_monitor_format":"gsm","disk_space":-1,"sipaccountlimit":-1,'))
row = ''.join((row,f'"cpslimit":-1,"state_number":"","neighborhood":"","calllimit_error":"503","transfer_international":0,"transfer_international_profit":null,'))
row = ''.join((row,f'"transfer_flexiload":0,"transfer_flexiload_profit":null,"transfer_bkash":0,"transfer_bkash_profit":null,"transfer_dbbl_rocket":0,'))
row = ''.join((row,f'"transfer_dbbl_rocket_profit":null,"transfer_bdservice_rate":null,"transfer_show_selling_price":null'))
return '{"id":0,"id_group":3,"id_plan":1,"id_user":0,"id_offer":0,"username":"test","password":"2uQqJH8F","active":1,"credit":0,"enableexpire":0,"expiredays":0,"status":0,"typepaid":0,"creditlimit":0,"credit_notification":-1,"restriction":0,"callingcard_pin":891457,"callshop":0,"plan_day":0,"active_paypal":0,"boleto":0,"lastname":"TEST","firstname":"TEST","redial":"","tag":"","company_name":"","commercial_name":"","address":"","city":"","state":"","country":"1","loginkey":"","zipcode":"","phone":"","mobile":"","email":"","doc":"","vat":"","language":"es","company_website":"","prefix_local":"","boleto_day":0,"firstusedate":null,"expirationdate":null,"lastuse":null,"description":"","creationdate":null,"id_group_agent":0,"calllimit":-1,"mix_monitor_format":"gsm","disk_space":-1,"sipaccountlimit":-1,"cpslimit":-1,"state_number":"","neighborhood":"","calllimit_error":"503","transfer_international":0,"transfer_international_profit":null,"transfer_flexiload":0,"transfer_flexiload_profit":null,"transfer_bkash":0,"transfer_bkash_profit":null,"transfer_dbbl_rocket":0,"transfer_dbbl_rocket_profit":null,"transfer_bdservice_rate":null,"transfer_show_selling_price":null}'
def set_headers(cnx : Session):
cnx.headers['Accept-Encoding'] = 'gzip, deflate'
cnx.headers['Connection'] = 'keep-alive'
cnx.headers['Content-Type'] = 'application/x-www-form-urlencoded; charset=UTF-8'
cnx.headers['Referer'] = magnus.replace('index.php', '')
cnx.headers['Origin'] = magnus.replace('/mbilling/index.php', '')
def create_account(nombre : str, telefono : str, id_cliente : int, email : str = "", apellido : str = '', tarifa : Tarifa = Tarifa.ESTANDAR):
with Session() as cnx:
set_headers(cnx)
cnx.get(magnus)
cnx.post(f'{magnus}/authentication/login', data=login).raise_for_status()
response = cnx.post(f'{magnus}/user/save', data={
'rows' : generate_rows(tarifa, telefono, id_cliente, nombre, apellido, email)
})
try:
response.raise_for_status()
except HTTPError as e:
print(response.request.headers)
finally:
with open('result.html', 'wb') as f:
f.write(response.content)
create_account(tarifa=Tarifa.ESTANDAR,nombre="Cliente",telefono="856856856",id_cliente=256452,email="pruebas@pruebas.es")
e
\ No newline at end of file
notebook/magnusbillingse.py deleted 100644 → 0
import profile
from selenium import webdriver
from selenium.webdriver.firefox.options import Options as FirefoxOptions
from selenium.webdriver.common.desired_capabilities import DesiredCapabilities
from selenium.webdriver.firefox.firefox_profile import FirefoxProfile
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.by import By
from selenium.common.exceptions import TimeoutException, ElementNotInteractableException
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.common.action_chains import ActionChains
import mariadb
address = 'https://192.168.2.71'
number = '956956956'
password = '123456789'
class AuthException(Exception):
pass
def login(driver: webdriver.Firefox, address: str, password: str):
driver.get(address)
WebDriverWait(driver=driver, timeout=5).until(EC.element_to_be_clickable(
(By.XPATH, '//input[@name="input_user"]')
)).send_keys('admin')
WebDriverWait(driver=driver, timeout=5).until(EC.element_to_be_clickable(
(By.XPATH, '//input[@name="input_pass"]')
)).send_keys(password, Keys.ENTER)
try:
WebDriverWait(driver=driver, timeout=5).until(EC.element_to_be_clickable(
(By.XPATH, '//*[text()="System"]')
))
except TimeoutException:
raise AuthException()
if __name__ == '__main__':
options = FirefoxOptions()
options.accept_insecure_certs = True
options.capabilities['acceptSslCerts'] = True
# options.add_argument("--headless")
driver = webdriver.Firefox(options=options)
# Inicio de Sesión
try:
login(driver, address, 'Mur5cKUBatdw')
except AuthException:
login(driver, address, '_RW!2k14')
driver.get(f'{address}/?menu=pbxconfig&type=setup&display=trunks')
WebDriverWait(driver, 5).until(EC.element_to_be_clickable(
(By.XPATH, '//span[contains(text(),"Add SIP Trunk")]')
)).click()
try:
WebDriverWait(driver, 5).until(EC.element_to_be_clickable(
(By.XPATH, '//input[@name="trunk_name"]')
)).send_keys(number)
except TimeoutException:
driver.get(f'{address}/?menu=pbxconfig&type=setup&display=trunks')
WebDriverWait(driver, 5).until(EC.element_to_be_clickable(
(By.XPATH, '//span[contains(text(),"Add SIP Trunk")]')
)).click()
WebDriverWait(driver, 5).until(EC.element_to_be_clickable(
(By.XPATH, '//input[@name="trunk_name"]')
)).send_keys(number)
driver.find_element(By.XPATH, '//input[@name="outcid"]').send_keys(number)
driver.find_element(
By.XPATH, '//input[@name="pattern_pass[0]"]').send_keys('.')
driver.find_element(
By.XPATH, '//input[@name="prepend_digit[0]"]').send_keys('34')
driver.find_element(
By.XPATH, '//input[@name="channelid"]').send_keys(number)
peer_details = driver.find_element(
By.XPATH, '//textarea[@name="peerdetails"]'
)
peer_details.clear()
peer_details.send_keys(
f'type = friend\nusername = {number}\nhost = 192.168.2.84\nnat = force_rport, comedia\nqualify = yes\ninsecure = invite, port'
)
driver.find_element(
By.XPATH, '//input[@name="usercontext"]').send_keys('from-trunk')
user_config = driver.find_element(
By.XPATH, '//textarea[@name="userconfig"]'
)
user_config.clear()
user_config.send_keys(
'type = friend\nhost = 192.168.2.84\nnat = force_rport, comedia\nqualify = yes\ninsecure = invite, port'
)
driver.find_element(By.XPATH, '//input[@name="Submit"]').click()
WebDriverWait(driver, 5).until(EC.element_to_be_clickable((
By.XPATH, '//a[@id="button_reload"]'
))).click()
notebook/test.py deleted 100644 → 0
import numbers
from operator import sub
import subprocess
from pathlib import Path
import re
from unittest import result
import requests
mbapi_path = Path('D:\Proyectos\scrapper-portabilidad\magnus-api')
succesful = re.compile(r'^\[20\d\]')
class MagnusCommandError(Exception):
pass
def _create_user(phone_number: str, password: str, name: str, surname: str):
"""Genera una cuenta de usuario en MagnusBilling
"""
result = subprocess.check_output(
['php', Path(mbapi_path, 'create_user.php'),
phone_number, password, name, surname]
)
if not re.match(succesful, result.decode(encoding='utf-8')):
raise MagnusCommandError(result.decode(encoding='utf-8'))
def _create_sip(phone_number: str, password: str, pbx: str = None):
command = ['php', Path(mbapi_path, 'create_sip.php'),
phone_number, password]
if pbx:
command.append(pbx)
result = subprocess.check_output(command)
if not re.match(succesful, result.decode(encoding='utf-8')):
raise MagnusCommandError(result.decode(encoding='utf-8'))
def _create_did(phone_number: str):
"""Genera un DID
"""
for did in (phone_number, f'34{phone_number}'):
result = subprocess.check_output(
['php', Path(mbapi_path, 'create_did.php'), did]
)
result = subprocess.check_output(
['php', Path(mbapi_path, 'assign_did.php'), phone_number, did]
)
if not re.match(succesful, result.decode('utf-8')):
raise MagnusCommandError(result.decode('utf-8'))
def create_user(phone_number: str, name: str, surname: str, password: str = None, pbx: str = None, balance: float = 0.0):
"""Crea a un nuevo usuario en MagnusBilling
"""
password = password if password else ''.join(
(name.split(' ')[0].lower(), phone_number[4:6]))
_create_user(phone_number, password, name, surname)
_create_sip(phone_number, password, pbx)
_create_did(phone_number)
response = {'number': phone_number}
if pbx:
response['outgoing'] = f'type = friend\nusername = {phone_number}\nhost = 192.168.2.84\nnat = force_rport, comedia\nqualify = yes\ninsecure = invite, port'
response['incoming'] = 'type = friend\nhost = 192.168.2.84\nnat = force_rport, comedia\nqualify = yes\ninsecure = invite, port'
else:
response['password'] = phone_number
return response
def add_trunk_account(pbx: str, username: str, password: str):
"""Añade la troncal a una Centralita
"""
base = pbx if re.match(r'^https:/.*$', pbx) else f'https://{pbx}'
with requests.Session() as s:
r = s.post(
url=base,
verify=False,
data={
'input_user': 'admin',
'input_pass': 'Mur5cKUBatdw',
'submit_login': ''
}
)
r = s.get(
url=f'{base}/config.php?display=trunks&tech=SIP'
)
print(r.content)
account = {
'number': '956385412',
'name': 'AkunaDecor',
'surname': 'Interiorismo S.L.',
'pbx': '192.168.2.54'
}
user = create_user(phone_number = account['number'], name = account['name'], surname = account['surname'], pbx= account['pbx'])
print(f'url: {account["pbx"]}')
print('--------')
print(user['number'])
print('--------')
print(user['outgoing'])
print('--------')
print(user['incoming'])
print('--------')
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