import requests import logging from datetime import datetime from collections import namedtuple from urllib.parse import quote_plus from .config import OpenwrtConfig from ..modem.config import ModemsConfig DHCPLease = namedtuple('DHCPLease', 'time, time_s, mac, ip, hostname') _config = OpenwrtConfig() _modems_config = ModemsConfig() _logger = logging.getLogger(__name__) def ipset_list_all() -> list: args = ['ipset-list-all'] args += _modems_config.keys() lines = _to_list(_call(args)) sets = {} cur_set = None for line in lines: if line.startswith('>'): cur_set = line[1:] if cur_set not in sets: sets[cur_set] = [] continue if cur_set is None: _logger.error('ipset_list_all: cur_set is not set') continue sets[cur_set].append(line) return sets def ipset_add(set_name: str, ip: str): return _call(['ipset-add', set_name, ip]) def ipset_del(set_name: str, ip: str): return _call(['ipset-del', set_name, ip]) def set_upstream(ip: str): return _call(['homekit-set-default-upstream', ip]) def get_default_route(): return _call(['get-default-route']) def get_dhcp_leases() -> list[DHCPLease]: return list(map(lambda item: _to_dhcp_lease(item), _to_list(_call(['dhcp-leases'])))) def _call(arguments: list[str]) -> str: url = _get_link(arguments) r = requests.get(url) r.raise_for_status() return r.text.strip() def _get_link(arguments: list[str]) -> str: url = f'http://{_config["ip"]}/cgi-bin/luci/command/{_config["command_id"]}' if arguments: url += '/' url += '%20'.join(list(map(lambda arg: quote_plus(arg.replace('/', '_')), arguments))) return url def _to_list(s: str) -> list: return [] if s == '' else s.split('\n') def _to_dhcp_lease(s: str) -> DHCPLease: words = s.split(' ') time = int(words.pop(0)) mac = words.pop(0) ip = words.pop(0) words.pop() hostname = (' '.join(words)).strip() if not hostname or hostname == '*': hostname = '?' return DHCPLease(time=time, time_s=datetime.fromtimestamp(time).strftime('%d %b, %H:%M:%S'), mac=mac, ip=ip, hostname=hostname)