227 lines
7.1 KiB
Python
227 lines
7.1 KiB
Python
"""Lectura segura de datos ASUSWRT mediante la API web del router."""
|
|
|
|
import asyncio
|
|
import html
|
|
import json
|
|
import os
|
|
import unicodedata
|
|
|
|
import aiohttp
|
|
from asusrouter import AsusData, AsusRouter
|
|
from asusrouter.modules.parental_control import PCRuleType, ParentalControlRule
|
|
from asusrouter.modules.system import AsusSystem
|
|
|
|
|
|
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']}"
|
|
|
|
|
|
async def _aimesh_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.AIMESH, force=True)
|
|
finally:
|
|
await router.async_disconnect()
|
|
await session.close()
|
|
|
|
|
|
async def _aimesh_action(cmd, mac=None):
|
|
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()
|
|
if cmd == "reconstruir":
|
|
return await router.async_set_state(AsusSystem.AIMESH_REBUILD)
|
|
if cmd == "reiniciar_todo":
|
|
return await router.async_set_state(AsusSystem.AIMESH_REBOOT)
|
|
return await router.async_set_state(
|
|
AsusSystem.NODE_REBOOT, device_list=mac
|
|
)
|
|
finally:
|
|
await router.async_disconnect()
|
|
await session.close()
|
|
|
|
|
|
def _nodo_texto(node):
|
|
tipo = "ROUTER" if getattr(node, "type", "") == "router" else "NODO"
|
|
alias = node.alias or node.model or "Sin nombre"
|
|
lineas = [f"{tipo} · {alias} ({node.model})"]
|
|
estado = "conectado" if node.status else "apagado"
|
|
lineas.append(f" estado: {estado}")
|
|
if node.ip:
|
|
lineas.append(f" ip: {node.ip}")
|
|
if node.fw:
|
|
lineas.append(f" firmware: {node.fw}")
|
|
if node.mac:
|
|
lineas.append(f" mac: {node.mac}")
|
|
parent = node.parent or {}
|
|
if parent:
|
|
rssi = parent.get("rssi", "?")
|
|
conexion = parent.get("connection", "?")
|
|
lineas.append(f" enlace: {conexion} · rssi={rssi}")
|
|
for banda, ssid in (node.ap or {}).items():
|
|
lineas.append(f" wifi {banda}: {ssid}")
|
|
return "\n".join(lineas)
|
|
|
|
|
|
def aimesh_estado():
|
|
data = asyncio.run(_aimesh_data()) or {}
|
|
if not data:
|
|
return "<pre>Sin datos AiMesh</pre>"
|
|
nodos = list(data.values())
|
|
nodos.sort(key=lambda n: getattr(n, "level", 0))
|
|
texto = "\n\n".join(_nodo_texto(n) for n in nodos)
|
|
return f"<pre>{html.escape(texto)}</pre>"
|
|
|
|
|
|
def _resolver_nodo(nombre):
|
|
data = asyncio.run(_aimesh_data()) or {}
|
|
clave = normalizar_nombre(nombre)
|
|
for mac, node in data.items():
|
|
if mac and mac.lower() == nombre.lower().strip():
|
|
return node
|
|
alias = normalizar_nombre(node.alias or "")
|
|
if clave and (alias == clave or clave in alias or alias in clave):
|
|
return node
|
|
return None
|
|
|
|
|
|
def aimesh_reiniciar_nodo(nombre):
|
|
node = _resolver_nodo(nombre)
|
|
if not node:
|
|
raise ValueError(f"Nodo AiMesh no encontrado: {nombre}")
|
|
asyncio.run(_aimesh_action("nodo", mac=node.mac))
|
|
return f"🔄 Nodo reiniciado: {node.alias or node.model} ({node.mac})"
|
|
|
|
|
|
def aimesh_reconstruir():
|
|
asyncio.run(_aimesh_action("reconstruir"))
|
|
return "🛠️ AiMesh reconstruida. Los nodos se reconectarán en unos minutos."
|
|
|
|
|
|
def aimesh_reiniciar_todo():
|
|
asyncio.run(_aimesh_action("reiniciar_todo"))
|
|
return "⚠️ Router y todos los nodos se están reiniciando..."
|