Yandex Cloud function to switch TV VPN on Keenetic router via Yandex Alice skill

switch-tv-vpn.py

import hashlib
import json
import logging
import os

import requests

logger = logging.getLogger()
logger.setLevel(logging.INFO)

ROUTER = os.environ.get("ROUTER_URL")
USER = os.environ.get("ROUTER_USER", "admin")
PASSWORD = os.environ.get("ROUTER_PASSWORD")
TV_MAC = os.environ.get("TV_MAC")
VPN_POLICY = os.environ.get("VPN_POLICY_NAME", "Policy0")
NO_VPN_POLICY = os.environ.get("NO_VPN_POLICY_NAME", "Policy1")

HTTP_UNAUTHORIZED = 401

ENABLE_WORDS = frozenset({"включи", "включить", "включай", "врубить", "врубай", "вкл"})
DISABLE_WORDS = frozenset({
    "выключи", "выключить", "выключай", "вырубить", "вырубай", "выкл",
    "отключи", "отключить", "отключай",
})


def parse_action(text):
    for w in text.lower().split():
        if w in ENABLE_WORDS:
            return True
        if w in DISABLE_WORDS:
            return False
    return None


def apply_policy(enable_vpn):
    policy = VPN_POLICY if enable_vpn else NO_VPN_POLICY
    with requests.Session() as session:
        response = session.get(f"{ROUTER}/auth", timeout=5)
        if response.status_code == HTTP_UNAUTHORIZED:
            realm = response.headers["X-NDM-Realm"]
            chall = response.headers["X-NDM-Challenge"]
            md5 = hashlib.md5(f"{USER}:{realm}:{PASSWORD}".encode()).hexdigest()
            sha = hashlib.sha256((chall + md5).encode()).hexdigest()
            session.post(f"{ROUTER}/auth", json={"login": USER, "password": sha}, timeout=5)
        session.post(
            f"{ROUTER}/rci/ip/hotspot/host",
            json={"mac": TV_MAC, "permit": True, "policy": policy},
            timeout=5,
        )
        session.post(f"{ROUTER}/rci/system/configuration/save", json={}, timeout=5)


def respond(text, end_session=False):
    return {
        "response": {"text": text, "tts": text, "end_session": end_session},
        "version": "1.0",
    }


def handler(event, context):
    body = event if "request" in event else json.loads(event.get("body", "{}"))
    session = body.get("session", {})
    req = body.get("request", {})

    logger.info("Incoming: %s", json.dumps(body, ensure_ascii=False))

    if session.get("new"):
        logger.info("New session")
        return respond("Привет! Скажи: включи или выключи впн.")

    utterance = req.get("original_utterance", "").strip()
    if not utterance:
        logger.warning("Empty utterance")
        return respond("Не расслышала. Повтори.")

    enable = parse_action(utterance)
    if enable is None:
        logger.warning("Could not parse action from: %s", utterance)
        return respond("Не поняла. Скажи: включи впн или выключи впн.")

    action_text = "включаю" if enable else "выключаю"
    logger.info("%s VPN on TV", action_text)

    try:
        apply_policy(enable)
        done = "включён" if enable else "выключён"
        text = f"Готово! VPN {done} на телевизоре."
        logger.info("Success: %s", text)
    except Exception as exc:
        logger.exception("Failed to apply policy")
        text = f"Не удалось: {exc}"

    return respond(text, end_session=True)