113 lines
3.4 KiB
Python
113 lines
3.4 KiB
Python
"""Lectura segura de datos ASUSWRT mediante la API web del router."""
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import unicodedata
|
|
|
|
import aiohttp
|
|
from asusrouter import AsusData, AsusRouter
|
|
from asusrouter.modules.parental_control import PCRuleType, ParentalControlRule
|
|
|
|
|
|
def _load_env():
|
|
values = {}
|
|
path = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env")
|
|
try:
|
|
with open(path) as stream:
|
|
for line in stream:
|
|
line = line.strip()
|
|
if line and not line.startswith("#") and "=" in line:
|
|
key, value = line.split("=", 1)
|
|
values[key.strip()] = value.strip()
|
|
except OSError:
|
|
pass
|
|
return values
|
|
|
|
|
|
FILE_ENV = _load_env()
|
|
|
|
DEVICE_ALIASES = {
|
|
"ines": {"name": "TABLET INES", "mac": "04:34:F6:0C:8E:48"},
|
|
"tablet ines": {"name": "TABLET INES", "mac": "04:34:F6:0C:8E:48"},
|
|
"carolina": {"name": "Caroline-tablet", "mac": "3A:99:2B:ED:D3:5E"},
|
|
"caroline": {"name": "Caroline-tablet", "mac": "3A:99:2B:ED:D3:5E"},
|
|
"caroline tablet": {"name": "Caroline-tablet", "mac": "3A:99:2B:ED:D3:5E"},
|
|
}
|
|
|
|
|
|
def _env(name, default=""):
|
|
return os.getenv(name, FILE_ENV.get(name, default)).strip()
|
|
|
|
|
|
def normalizar_nombre(nombre):
|
|
value = unicodedata.normalize("NFKD", nombre or "")
|
|
value = "".join(char for char in value if not unicodedata.combining(char))
|
|
return " ".join(value.lower().replace("-", " ").split())
|
|
|
|
|
|
def resolver_dispositivo(nombre):
|
|
return DEVICE_ALIASES.get(normalizar_nombre(nombre))
|
|
|
|
|
|
async def _parental_data():
|
|
username = _env("ASUS_USERNAME")
|
|
password = _env("ASUS_PASSWORD")
|
|
if not username or not password:
|
|
raise RuntimeError(
|
|
"Falta ASUS_USERNAME o ASUS_PASSWORD en el .env de la Raspberry"
|
|
)
|
|
session = aiohttp.ClientSession()
|
|
router = AsusRouter(
|
|
hostname=_env("ASUS_ROUTER_IP", "192.168.50.1"),
|
|
username=username,
|
|
password=password,
|
|
session=session,
|
|
)
|
|
try:
|
|
await router.async_connect()
|
|
return await router.async_get_data(AsusData.PARENTAL_CONTROL, force=True)
|
|
finally:
|
|
await router.async_disconnect()
|
|
await session.close()
|
|
|
|
|
|
async def _set_block(mac, name, blocked):
|
|
username = _env("ASUS_USERNAME")
|
|
password = _env("ASUS_PASSWORD")
|
|
if not username or not password:
|
|
raise RuntimeError(
|
|
"Falta ASUS_USERNAME o ASUS_PASSWORD en el .env de la Raspberry"
|
|
)
|
|
session = aiohttp.ClientSession()
|
|
router = AsusRouter(
|
|
hostname=_env("ASUS_ROUTER_IP", "192.168.50.1"),
|
|
username=username,
|
|
password=password,
|
|
session=session,
|
|
)
|
|
try:
|
|
await router.async_connect()
|
|
rule = ParentalControlRule(
|
|
mac=mac,
|
|
name=name if blocked else None,
|
|
type=PCRuleType.BLOCK if blocked else PCRuleType.REMOVE,
|
|
)
|
|
return await router.async_set_state(rule)
|
|
finally:
|
|
await router.async_disconnect()
|
|
await session.close()
|
|
|
|
|
|
def parental_status():
|
|
data = asyncio.run(_parental_data())
|
|
return json.dumps(data, ensure_ascii=False, indent=2, default=str)
|
|
|
|
|
|
def cambiar_bloqueo(nombre, blocked):
|
|
device = resolver_dispositivo(nombre)
|
|
if not device:
|
|
raise ValueError(f"Dispositivo no reconocido: {nombre}")
|
|
asyncio.run(_set_block(device["mac"], device["name"], blocked))
|
|
return f"{'Bloqueado' if blocked else 'Activado'}: {device['name']}"
|