Skip to content

S
scrapper-portabilidad
Commit 8878df1f by Javier
Options

Mejoras

parent 3b15017c
Showing with 212 additions and 33 deletions
.vscode/launch.json
......@@ -18,7 +18,7 @@
"type": "python",
"request": "launch",
"module": "uvicorn",
"args": ["api:app", "--host", "0.0.0.0", "--port", "8080"],
"args": ["api:app", "--host", "0.0.0.0", "--port", "25001"],
"console": "integratedTerminal",
"justMyCode": true
},
......
a2billing/a2billing.py
from typing import Tuple
from winreg import QueryInfoKey
import mariadb
from requests import get
......
api.py
from site import USER_BASE
from fastapi import Body, Depends, FastAPI, HTTPException, Query, Response, Request
from fastapi.responses import JSONResponse
import mbilling.models as models
......@@ -46,22 +47,40 @@ def magnus_error_handler(request: Request, ex: MagnusCommandError):
)
@app.exception_handler(mbe.UserNotFound)
def magnus_error_handler(request: Request, ex: mbe.UserNotFound):
return JSONResponse(
status_code=404,
content={
"detail": f'No existe ningún usuario que contenga el número.',
}
)
def get_user_response_model(number: str):
"""Devuelve tupla Usuario / Modelo Respuesta
"""
user = mbc.find_user_account(number)
if not user:
raise mbe.UserNotFound
sip = [app.url_path_for('Obtiene una Cuenta SIP', caller_id=sip.name)
for sip in mbc.get_sip_accounts_by_user(user)]
return user, models.User.generate_from_automap(user, sip)
@app.get('/v2/account/{account_name}', response_model=models.User, name="get_account",
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.find_user_account(account_name)
if not user:
raise HTTPException(
status_code=404, detail="No existe ningún usuario que contenga el número.")
elif user.username != account_name:
raise HTTPException(
status_code=302, detail=app.url_path_for('get_account', account_name=user.username)
)
user, response = get_user_response_model(account_name)
if user.username != account_name:
raise HTTPException(status_code=302, detail=app.url_path_for(
'get_account', account_name=user.username))
else:
sip = [app.url_path_for('Obtiene una Cuenta SIP', caller_id=sip.name)
for sip in mbc.get_sip_accounts_by_user(user)]
return models.User.generate_from_automap(user, sip)
return response
@app.post('/v2/account', response_model=models.User, name="Crea una Cuenta de Facturación",
......@@ -70,8 +89,16 @@ def get_account(account_name: str = Query(..., regex=r'[89][1-9]\d{7}', name='N
def add_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)
if not user_data.address:
user_data.address = 'dynamic'
mbc.create_user(user_data.username, user_data.name,
user_data.surname, user_data.rate, user_data.address)
user, response = get_user_response_model(user_data.username)
return response
except mbe.AccountExistsError:
raise HTTPException(
......@@ -87,7 +114,10 @@ def add_account(user_data: models.CreateUser):
def modify_account(account_name: str, user_data: models.CreateUser):
try:
return mbc.update_user(account_name, user_data)
mbc.update_user(account_name, user_data)
user, response = get_user_response_model(account_name)
return response
except mbe.UserNotFound:
raise HTTPException(
status_code=404, detail='No existe ningún usuario con esa numeración.')
......@@ -98,6 +128,9 @@ def modify_account(account_name: str, user_data: models.CreateUser):
def add_account_sip(sip: models.CreateSipAccount, account_name: str = Query(..., regex=models.PHONE_REGEX, name='Número', description='Cuenta SIP a recargar'),):
try:
if not sip.host:
sip.host = 'dynamic'
return mbc.create_sip(account_name, sip)
except mbe.UserNotFound as e:
......@@ -127,7 +160,7 @@ def get_account_sip(account_name: str = Query(..., regex=models.PHONE_REGEX, nam
@app.get('/v2/sip/{caller_id}', response_model=models.SipAccount, name="Obtiene una Cuenta SIP", description="Devuelve los datos de conexión de la cuenta.",
tags=["SIP"], status_code=200)
def add_account_sip(caller_id: str = Query(..., regex=models.PHONE_REGEX, name='Número', description='Cuenta SIP a consultar'),):
try:
sip, user = mbc.get_sip_account(caller_id)
return models.SipAccount.generate_from_automap(sip, user.username)
......@@ -140,12 +173,11 @@ def add_account_sip(caller_id: str = Query(..., regex=models.PHONE_REGEX, name='
@app.post('/v2/sip/{caller_id}/transfer', response_model=models.SipAccount, name="Transfiere la Cuenta", description="Modifica al Propietario de la Cuenta",
tags=["SIP"], status_code=200)
def add_account_sip(account : models.TransferSip, caller_id: str = Query(..., regex=models.PHONE_REGEX, name='Número', description='Cuenta SIP a consultar')):
tags=["SIP"], status_code=200)
def add_account_sip(account: models.TransferSip, caller_id: str = Query(..., regex=models.PHONE_REGEX, name='Número', description='Cuenta SIP a consultar')):
try:
sip, user = mbc.get_sip_account(caller_id)
except mbe.UserNotFound as e:
raise HTTPException(status_code=404, detail=str(e))
......
cli.py
......@@ -10,6 +10,9 @@ import pyodbc
import os
from datetime import timedelta, datetime
from a2billing import a2billing
import pathlib
from dotenv import dotenv_values
import mariadb
# Credenciales SQL Server
sql_server_hostname = "192.168.2.200"
......@@ -535,10 +538,22 @@ def synchronize():
@app.command()
def test():
print(a2billing.account_data('956384412'))
# update_a2billing_balance()
def reset_daily_limit():
mbapi_path = pathlib.Path(pathlib.Path().resolve(), 'mbilling')
values = dotenv_values(pathlib.Path(mbapi_path, '.env'))
try:
with mariadb.connect(user=values['DB_USER'], password=values['DB_PASSWORD'], host=values['DB_HOST'],
port=3306, database=values['DB_NAME']) as cnx:
with cnx.cursor() as cursor:
cursor.execute(
'UPDATE mbilling.pkg_user SET credit = 0 WHERE typepaid=1 AND credit < 0.0;')
cnx.commit()
print("Se han reseteado los límites diarios de consumo.")
except mariadb.Error as e:
print(f"Se ha producido un error de conexión con MariaDB: {e}")
if __name__ == "__main__":
app()
mbilling/connector.py
from random import randint
from typing import Tuple
from sqlalchemy import create_engine, select, update
from sqlalchemy.orm import sessionmaker
......@@ -192,6 +193,8 @@ def create_user(number: str, name: str, surname: str, rate: models.Rate = models
else:
raise mbe.SipRegisteredError()
return get_user(number)
def update_user(username: str, user_data: models.User):
"""Actualiza una Cuenta de Facturación
......@@ -292,14 +295,17 @@ def create_or_update_did(number: str, user: User, sip: Sip = None):
if not did:
cli._create_did(did_number, user.username,
auto_international=False)
update_did_owner(did, sip)
else:
update_did_owner(did, sip)
def create_sip(username: str, create_sip: models.CreateSipAccount) -> models.SipAccount:
user_account = find_user_account(username)
create_sip.secret = create_sip.secret if create_sip.secret else ''.join(
(user_account.firstname.split(' ')[0].lower(), str(randint(100, 999))))
if not user_account:
raise mbe.UserNotFound()
......@@ -326,6 +332,6 @@ def create_sip(username: str, create_sip: models.CreateSipAccount) -> models.Sip
raise mbe.SipRegisteredError(
'La cuenta SIP está asignada a otro usuario.')
def transfer_sip(sip : Sip, user : User):
def transfer_sip(sip: Sip, user: User):
pass
mbilling/models.py
......@@ -171,6 +171,8 @@ class SipAccount(BaseModel):
@staticmethod
def generate_from_automap(sip, account_name: str):
magnus_host = '10.58.6.221' if sip.host == 'dynamic' else '192.168.2.84'
peer_details = PeerDetails(
username=sip.name,
secret=sip.secret,
......@@ -180,7 +182,7 @@ class SipAccount(BaseModel):
type=sip.type,
dtmfmode=sip.dtmfmode,
insecure=sip.insecure,
host=sip.host,
host=magnus_host,
allow=sip.allow
)
......@@ -191,7 +193,7 @@ class SipAccount(BaseModel):
type=sip.type,
dtmfmode=sip.dtmfmode,
insecure=sip.insecure,
host=sip.host,
host=magnus_host,
allow=sip.allow
)
......
mbilling/vendor/symfony/polyfill-php80/PhpToken.php 0 → 100644
<?php
/*
* This file is part of the Symfony package.
*
* (c) Fabien Potencier <fabien@symfony.com>
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Symfony\Polyfill\Php80;
/**
* @author Fedonyuk Anton <info@ensostudio.ru>
*
* @internal
*/
class PhpToken implements \Stringable
{
/**
* @var int
*/
public $id;
/**
* @var string
*/
public $text;
/**
* @var int
*/
public $line;
/**
* @var int
*/
public $pos;
public function __construct(int $id, string $text, int $line = -1, int $position = -1)
{
$this->id = $id;
$this->text = $text;
$this->line = $line;
$this->pos = $position;
}
public function getTokenName(): ?string
{
if ('UNKNOWN' === $name = token_name($this->id)) {
$name = \strlen($this->text) > 1 || \ord($this->text) < 32 ? null : $this->text;
}
return $name;
}
/**
* @param int|string|array $kind
*/
public function is($kind): bool
{
foreach ((array) $kind as $value) {
if (\in_array($value, [$this->id, $this->text], true)) {
return true;
}
}
return false;
}
public function isIgnorable(): bool
{
return \in_array($this->id, [\T_WHITESPACE, \T_COMMENT, \T_DOC_COMMENT, \T_OPEN_TAG], true);
}
public function __toString(): string
{
return (string) $this->text;
}
/**
* @return static[]
*/
public static function tokenize(string $code, int $flags = 0): array
{
$line = 1;
$position = 0;
$tokens = token_get_all($code, $flags);
foreach ($tokens as $index => $token) {
if (\is_string($token)) {
$id = \ord($token);
$text = $token;
} else {
[$id, $text, $line] = $token;
}
$tokens[$index] = new static($id, $text, $line, $position);
$position += \strlen($text);
}
return $tokens;
}
}
mbilling/vendor/symfony/polyfill-php80/Resources/stubs/PhpToken.php 0 → 100644
<?php
if (\PHP_VERSION_ID < 80000 && \extension_loaded('tokenizer')) {
class PhpToken extends Symfony\Polyfill\Php80\PhpToken
{
}
}
requirements.txt
No preview for this file type
test.py
import mbilling.models as models
import mbilling.connector as mbc
import mbilling.exceptions as mbe
import re
import pathlib
from dotenv import dotenv_values
import mariadb
mbc.test_function('856135158')
mbapi_path = pathlib.Path(pathlib.Path().resolve(), 'mbilling')
values = dotenv_values(pathlib.Path(mbapi_path, '.env'))
try:
with mariadb.connect(user=values['DB_USER'], password=values['DB_PASSWORD'], host=values['DB_HOST'],
port=3306, database=values['DB_NAME']) as cnx:
with cnx.cursor() as cursor:
cursor.execute(
'UPDATE mbilling.pkg_user SET credit = 0 WHERE typepaid=1 AND credit < 0.0;')
cnx.commit()
print("Se han reseteado los límites diarios de consumo.")
except mariadb.Error as e:
print(f"Se ha producido un error de conexión con MariaDB: {e}")
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