chore: archive mac mini automation baseline
This commit is contained in:
Adolfo Reyna
+156
@@ -0,0 +1,156 @@
|
||||
#!/usr/bin/env python3
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
from tplinkrouterc6u import TPLinkDecoClient
|
||||
|
||||
|
||||
def default_gateway():
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["/sbin/route", "-n", "get", "default"],
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=5,
|
||||
)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
for line in result.stdout.splitlines():
|
||||
key, _, value = line.partition(":")
|
||||
if key.strip() == "gateway":
|
||||
return value.strip()
|
||||
return None
|
||||
|
||||
|
||||
def config():
|
||||
host = os.environ.get("DECO_HOST") or default_gateway()
|
||||
password = os.environ.get("DECO_PASSWORD")
|
||||
if not password:
|
||||
raise RuntimeError("DECO_PASSWORD must be set in .env.")
|
||||
if not host:
|
||||
raise RuntimeError("DECO_HOST must be set in .env; default gateway detection failed.")
|
||||
|
||||
verify_ssl = os.environ.get("DECO_VERIFY_SSL", "true").lower() not in {
|
||||
"0",
|
||||
"false",
|
||||
"no",
|
||||
}
|
||||
timeout = int(os.environ.get("DECO_TIMEOUT", "10"))
|
||||
return {
|
||||
"host": host,
|
||||
"password": password,
|
||||
"username": os.environ.get("DECO_USERNAME", "admin"),
|
||||
"verify_ssl": verify_ssl,
|
||||
"timeout": timeout,
|
||||
}
|
||||
|
||||
|
||||
def client():
|
||||
return TPLinkDecoClient(**config())
|
||||
|
||||
|
||||
def device_to_dict(device):
|
||||
return {
|
||||
"hostname": device.hostname,
|
||||
"mac": device.macaddr,
|
||||
"ip": device.ipaddr,
|
||||
"connection": getattr(device.type, "value", str(device.type)),
|
||||
"upSpeed": device.up_speed,
|
||||
"downSpeed": device.down_speed,
|
||||
"active": device.active,
|
||||
}
|
||||
|
||||
|
||||
def firmware_to_dict(firmware):
|
||||
return {
|
||||
"model": firmware.model,
|
||||
"hardwareVersion": firmware.hardware_version,
|
||||
"firmwareVersion": firmware.firmware_version,
|
||||
}
|
||||
|
||||
|
||||
def ipv4_to_dict(status):
|
||||
return {
|
||||
"wanMac": status.wan_macaddr,
|
||||
"wanIp": status.wan_ipv4_ipaddr,
|
||||
"wanGateway": status.wan_ipv4_gateway,
|
||||
"wanConnectionType": status.wan_ipv4_conntype,
|
||||
"wanNetmask": status.wan_ipv4_netmask,
|
||||
"wanPrimaryDns": status.wan_ipv4_pridns,
|
||||
"wanSecondaryDns": status.wan_ipv4_snddns,
|
||||
"lanMac": status.lan_macaddr,
|
||||
"lanIp": status.lan_ipv4_ipaddr,
|
||||
"lanNetmask": status.lan_ipv4_netmask,
|
||||
}
|
||||
|
||||
|
||||
def status_to_dict(status, include_clients=True):
|
||||
data = {
|
||||
"wanMac": status.wan_macaddr,
|
||||
"lanMac": status.lan_macaddr,
|
||||
"wanIp": status.wan_ipv4_addr,
|
||||
"lanIp": status.lan_ipv4_addr,
|
||||
"wanGateway": status.wan_ipv4_gateway,
|
||||
"connectionType": status.conn_type,
|
||||
"cpuUsage": status.cpu_usage,
|
||||
"memoryUsage": status.mem_usage,
|
||||
"clientsTotal": status.clients_total,
|
||||
"wiredClientsTotal": status.wired_total,
|
||||
"wifiClientsTotal": status.wifi_clients_total,
|
||||
"guestClientsTotal": status.guest_clients_total,
|
||||
"iotClientsTotal": status.iot_clients_total,
|
||||
"wifi": {
|
||||
"host2g": status.wifi_2g_enable,
|
||||
"host5g": status.wifi_5g_enable,
|
||||
"host6g": status.wifi_6g_enable,
|
||||
"guest2g": status.guest_2g_enable,
|
||||
"guest5g": status.guest_5g_enable,
|
||||
"guest6g": status.guest_6g_enable,
|
||||
},
|
||||
}
|
||||
if include_clients:
|
||||
data["clients"] = [device_to_dict(device) for device in status.devices]
|
||||
return data
|
||||
|
||||
|
||||
def run(action):
|
||||
deco = client()
|
||||
try:
|
||||
if action == "overview":
|
||||
status = deco.get_status()
|
||||
firmware = deco.get_firmware()
|
||||
return {
|
||||
"status": status_to_dict(status, include_clients=False),
|
||||
"firmware": firmware_to_dict(firmware),
|
||||
}
|
||||
if action == "clients":
|
||||
status = deco.get_status()
|
||||
return {"clients": [device_to_dict(device) for device in status.devices]}
|
||||
if action == "ipv4":
|
||||
return ipv4_to_dict(deco.get_ipv4_status())
|
||||
if action == "firmware":
|
||||
return firmware_to_dict(deco.get_firmware())
|
||||
raise RuntimeError(f"Unknown action: {action}")
|
||||
finally:
|
||||
try:
|
||||
deco.logout()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def main():
|
||||
if len(sys.argv) != 2:
|
||||
raise RuntimeError("Usage: deco_bridge.py <overview|clients|ipv4|firmware>")
|
||||
print(json.dumps(run(sys.argv[1]), indent=2))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
main()
|
||||
except Exception as err:
|
||||
print(json.dumps({"error": str(err)}), file=sys.stderr)
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,383 @@
|
||||
#!/usr/bin/env python3
|
||||
import base64
|
||||
import hashlib
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import re
|
||||
import secrets
|
||||
import ssl
|
||||
import subprocess
|
||||
import sys
|
||||
from urllib.parse import quote_plus
|
||||
|
||||
import requests
|
||||
from Crypto.Cipher import AES, PKCS1_v1_5
|
||||
from Crypto.PublicKey import RSA
|
||||
from Crypto.Util.Padding import pad, unpad
|
||||
from requests import RequestException
|
||||
|
||||
|
||||
AES_KEY_BYTES = 16
|
||||
MIN_AES_KEY = 10 ** (AES_KEY_BYTES - 1)
|
||||
MAX_AES_KEY = (10**AES_KEY_BYTES) - 1
|
||||
PKCS1_V1_5_HEADER_BYTES = 11
|
||||
|
||||
|
||||
def load_env_file():
|
||||
path = os.path.join(os.path.dirname(os.path.dirname(__file__)), ".env")
|
||||
try:
|
||||
with open(path, "r", encoding="utf-8") as env_file:
|
||||
for line in env_file:
|
||||
line = line.strip()
|
||||
if not line or line.startswith("#") or "=" not in line:
|
||||
continue
|
||||
key, value = line.split("=", 1)
|
||||
os.environ.setdefault(key, value.strip().strip("\"'"))
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
|
||||
|
||||
def default_gateway():
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["/sbin/route", "-n", "get", "default"],
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=5,
|
||||
)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
for line in result.stdout.splitlines():
|
||||
key, _, value = line.partition(":")
|
||||
if key.strip() == "gateway":
|
||||
return value.strip()
|
||||
return None
|
||||
|
||||
|
||||
def config():
|
||||
load_env_file()
|
||||
host = os.environ.get("DECO_HOST") or default_gateway()
|
||||
password = os.environ.get("DECO_PASSWORD")
|
||||
if not password:
|
||||
raise RuntimeError("DECO_PASSWORD must be set in .env.")
|
||||
if not host:
|
||||
raise RuntimeError("DECO_HOST must be set in .env; default gateway detection failed.")
|
||||
if not host.startswith(("http://", "https://")):
|
||||
host = f"http://{host}"
|
||||
verify_ssl = os.environ.get("DECO_VERIFY_SSL", "true").lower() not in {
|
||||
"0",
|
||||
"false",
|
||||
"no",
|
||||
}
|
||||
return {
|
||||
"host": host.rstrip("/"),
|
||||
"username": os.environ.get("DECO_USERNAME", "admin"),
|
||||
"password": password,
|
||||
"verify_ssl": verify_ssl,
|
||||
"timeout": int(os.environ.get("DECO_TIMEOUT", "10")),
|
||||
}
|
||||
|
||||
|
||||
def byte_len(n):
|
||||
return (int(math.log2(n)) + 8) >> 3
|
||||
|
||||
|
||||
def rsa_encrypt(n, e, plaintext):
|
||||
public_key = RSA.construct((n, e)).publickey()
|
||||
encryptor = PKCS1_v1_5.new(public_key)
|
||||
block_size = byte_len(n)
|
||||
bytes_per_block = block_size - PKCS1_V1_5_HEADER_BYTES
|
||||
encrypted_text = ""
|
||||
for index in range(0, len(plaintext), bytes_per_block):
|
||||
encrypted_text += encryptor.encrypt(plaintext[index:index + bytes_per_block]).hex()
|
||||
return encrypted_text
|
||||
|
||||
|
||||
def decode_name(value):
|
||||
if not value:
|
||||
return value
|
||||
try:
|
||||
return base64.b64decode(value).decode()
|
||||
except Exception:
|
||||
return value
|
||||
|
||||
|
||||
def title_from_snake(value):
|
||||
if not value:
|
||||
return value
|
||||
return " ".join(part.title() for part in value.split("_"))
|
||||
|
||||
|
||||
def deco_name(device):
|
||||
return (
|
||||
decode_name(device.get("custom_nickname"))
|
||||
or title_from_snake(device.get("nickname"))
|
||||
or device.get("device_model")
|
||||
or device.get("mac")
|
||||
)
|
||||
|
||||
|
||||
class DecoApi:
|
||||
def __init__(self, host, username, password, verify_ssl, timeout):
|
||||
self.host = host
|
||||
self.username = username
|
||||
self.password = password
|
||||
self.verify_ssl = verify_ssl
|
||||
self.timeout = timeout
|
||||
self.session = requests.Session()
|
||||
self.aes_key = None
|
||||
self.aes_iv = None
|
||||
self.password_rsa_n = None
|
||||
self.password_rsa_e = None
|
||||
self.sign_rsa_n = None
|
||||
self.sign_rsa_e = None
|
||||
self.seq = None
|
||||
self.stok = None
|
||||
self.cookie = None
|
||||
|
||||
def generate_aes_key_and_iv(self):
|
||||
self.aes_key = str(secrets.randbelow(MAX_AES_KEY - MIN_AES_KEY) + MIN_AES_KEY).encode()
|
||||
self.aes_iv = str(secrets.randbelow(MAX_AES_KEY - MIN_AES_KEY) + MIN_AES_KEY).encode()
|
||||
|
||||
def post(self, context, path, params, data):
|
||||
headers = {"Content-Type": "application/json"}
|
||||
cookies = {}
|
||||
if self.cookie:
|
||||
name, _, value = self.cookie.partition("=")
|
||||
if name and value:
|
||||
cookies[name] = value
|
||||
try:
|
||||
response = self.session.post(
|
||||
f"{self.host}{path}",
|
||||
params=params,
|
||||
data=data,
|
||||
headers=headers,
|
||||
cookies=cookies,
|
||||
verify=self.verify_ssl,
|
||||
timeout=self.timeout,
|
||||
)
|
||||
except RequestException:
|
||||
response = self.curl_post(path, params, data, cookies)
|
||||
if response.status_code == 403:
|
||||
self.clear_auth()
|
||||
raise RuntimeError(f"{context}: forbidden")
|
||||
response.raise_for_status()
|
||||
|
||||
for cookie in response.headers.get("Set-Cookie", "").split(","):
|
||||
match = re.search(r"(sysauth=[A-Za-z0-9]+)", cookie)
|
||||
if match:
|
||||
self.cookie = match.group(1)
|
||||
break
|
||||
|
||||
result = response.json()
|
||||
error_code = result.get("error_code")
|
||||
if error_code not in (None, "", 0):
|
||||
raise RuntimeError(f"{context}: response error_code={error_code}")
|
||||
return result
|
||||
|
||||
def curl_post(self, path, params, data, cookies):
|
||||
url = f"{self.host}{path}"
|
||||
if params:
|
||||
query = "&".join(f"{key}={quote_plus(str(value))}" for key, value in params.items())
|
||||
url = f"{url}?{query}"
|
||||
command = [
|
||||
"curl",
|
||||
"-s",
|
||||
"-i",
|
||||
"--connect-timeout",
|
||||
str(self.timeout),
|
||||
"-X",
|
||||
"POST",
|
||||
url,
|
||||
"-H",
|
||||
"Content-Type: application/json",
|
||||
"--data-raw",
|
||||
data,
|
||||
]
|
||||
if not self.verify_ssl:
|
||||
command.insert(2, "-k")
|
||||
if cookies:
|
||||
command.extend(["-H", "Cookie: " + "; ".join(f"{k}={v}" for k, v in cookies.items())])
|
||||
|
||||
result = subprocess.run(command, check=True, capture_output=True, text=True, timeout=self.timeout + 5)
|
||||
head, separator, body = result.stdout.rpartition("\r\n\r\n")
|
||||
if not separator:
|
||||
head, _, body = result.stdout.rpartition("\n\n")
|
||||
status_match = re.search(r"HTTP/\S+\s+(\d+)", head)
|
||||
status = int(status_match.group(1)) if status_match else 200
|
||||
response = requests.Response()
|
||||
response.status_code = status
|
||||
response._content = body.encode()
|
||||
response.url = url
|
||||
for line in head.splitlines():
|
||||
if ":" not in line:
|
||||
continue
|
||||
key, value = line.split(":", 1)
|
||||
response.headers[key.strip()] = value.strip()
|
||||
return response
|
||||
|
||||
def fetch_keys(self):
|
||||
response = self.post(
|
||||
"Fetch keys",
|
||||
"/cgi-bin/luci/;stok=/login",
|
||||
{"form": "keys"},
|
||||
json.dumps({"operation": "read"}),
|
||||
)
|
||||
keys = response["result"]["password"]
|
||||
self.password_rsa_n = int(keys[0], 16)
|
||||
self.password_rsa_e = int(keys[1], 16)
|
||||
|
||||
def fetch_auth(self):
|
||||
response = self.post(
|
||||
"Fetch auth",
|
||||
"/cgi-bin/luci/;stok=/login",
|
||||
{"form": "auth"},
|
||||
json.dumps({"operation": "read"}),
|
||||
)
|
||||
auth = response["result"]
|
||||
self.sign_rsa_n = int(auth["key"][0], 16)
|
||||
self.sign_rsa_e = int(auth["key"][1], 16)
|
||||
self.seq = auth["seq"]
|
||||
|
||||
def encode_payload(self, payload):
|
||||
payload_json = json.dumps(payload, separators=(",", ":")).encode()
|
||||
encrypted = AES.new(self.aes_key, AES.MODE_CBC, self.aes_iv).encrypt(
|
||||
pad(payload_json, AES.block_size)
|
||||
)
|
||||
data = base64.b64encode(encrypted).decode()
|
||||
sign = self.encode_sign(len(data))
|
||||
return f"sign={sign}&data={quote_plus(data)}"
|
||||
|
||||
def encode_sign(self, data_len):
|
||||
auth_hash = hashlib.md5(f"{self.username}{self.password}".encode()).hexdigest()
|
||||
sign_text = (
|
||||
f"k={self.aes_key.decode()}&i={self.aes_iv.decode()}&h={auth_hash}&s={self.seq + data_len}"
|
||||
)
|
||||
return rsa_encrypt(self.sign_rsa_n, self.sign_rsa_e, sign_text.encode())
|
||||
|
||||
def decrypt_data(self, context, data):
|
||||
if not data:
|
||||
self.clear_auth()
|
||||
raise RuntimeError(f"{context}: empty data")
|
||||
decrypted = AES.new(self.aes_key, AES.MODE_CBC, self.aes_iv).decrypt(
|
||||
base64.b64decode(data)
|
||||
)
|
||||
return json.loads(unpad(decrypted, AES.block_size).decode())
|
||||
|
||||
def login(self):
|
||||
if self.aes_key is None:
|
||||
self.generate_aes_key_and_iv()
|
||||
if self.password_rsa_n is None:
|
||||
self.fetch_keys()
|
||||
if self.seq is None:
|
||||
self.fetch_auth()
|
||||
encrypted_password = rsa_encrypt(
|
||||
self.password_rsa_n,
|
||||
self.password_rsa_e,
|
||||
self.password.encode(),
|
||||
)
|
||||
response = self.post(
|
||||
"Login",
|
||||
"/cgi-bin/luci/;stok=/login",
|
||||
{"form": "login"},
|
||||
self.encode_payload({
|
||||
"operation": "login",
|
||||
"params": {"password": encrypted_password},
|
||||
}),
|
||||
)
|
||||
data = self.decrypt_data("Login", response["data"])
|
||||
if data.get("error_code") != 0:
|
||||
result = data.get("result") or {}
|
||||
attempts = result.get("attemptsAllowed", "unknown")
|
||||
raise RuntimeError(f"Login failed: error_code={data.get('error_code')}; attempts={attempts}")
|
||||
self.stok = data["result"]["stok"]
|
||||
if not self.cookie:
|
||||
raise RuntimeError("Login succeeded but no sysauth cookie was returned.")
|
||||
|
||||
def clear_auth(self):
|
||||
self.seq = None
|
||||
self.stok = None
|
||||
self.cookie = None
|
||||
|
||||
def call(self, context, section, form, payload):
|
||||
if not self.stok or not self.cookie:
|
||||
self.login()
|
||||
response = self.post(
|
||||
context,
|
||||
f"/cgi-bin/luci/;stok={self.stok}/admin/{section}",
|
||||
{"form": form},
|
||||
self.encode_payload(payload),
|
||||
)
|
||||
data = self.decrypt_data(context, response["data"])
|
||||
error_code = data.get("error_code") or data.get("errorcode")
|
||||
if error_code:
|
||||
raise RuntimeError(f"{context}: decoded error_code={error_code}")
|
||||
return data["result"]
|
||||
|
||||
def list_decos(self):
|
||||
devices = self.call("List Devices", "device", "device_list", {"operation": "read"}).get("device_list", [])
|
||||
return [
|
||||
{
|
||||
"name": deco_name(device),
|
||||
"mac": device.get("mac"),
|
||||
"ip": device.get("device_ip"),
|
||||
"model": device.get("device_model"),
|
||||
"hardwareVersion": device.get("hardware_ver"),
|
||||
"firmwareVersion": device.get("software_ver"),
|
||||
"role": device.get("role"),
|
||||
"online": device.get("group_status") == "connected",
|
||||
"connectionType": device.get("connection_type"),
|
||||
}
|
||||
for device in devices
|
||||
]
|
||||
|
||||
def list_clients_for_deco(self, deco):
|
||||
clients = self.call(
|
||||
f"List Clients {deco['mac']}",
|
||||
"client",
|
||||
"client_list",
|
||||
{"operation": "read", "params": {"device_mac": deco["mac"]}},
|
||||
).get("client_list", [])
|
||||
return [
|
||||
{
|
||||
"hostname": decode_name(client.get("name")),
|
||||
"mac": client.get("mac"),
|
||||
"ip": client.get("ip"),
|
||||
"connection": client.get("connection_type"),
|
||||
"interface": client.get("interface"),
|
||||
"upSpeed": client.get("up_speed"),
|
||||
"downSpeed": client.get("down_speed"),
|
||||
"active": client.get("online"),
|
||||
"linkedDecoMac": deco.get("mac"),
|
||||
"linkedDecoName": deco.get("name"),
|
||||
"linkedDecoRole": deco.get("role"),
|
||||
}
|
||||
for client in clients
|
||||
if client.get("online")
|
||||
]
|
||||
|
||||
|
||||
def run():
|
||||
if not config()["verify_ssl"]:
|
||||
requests.packages.urllib3.disable_warnings()
|
||||
ssl._create_default_https_context = ssl._create_unverified_context
|
||||
deco = DecoApi(**config())
|
||||
decos = deco.list_decos()
|
||||
clients = {}
|
||||
for node in decos:
|
||||
if not node.get("mac"):
|
||||
continue
|
||||
for client in deco.list_clients_for_deco(node):
|
||||
clients[client["mac"]] = client
|
||||
return {"decos": decos, "clients": list(clients.values())}
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
print(json.dumps(run(), indent=2))
|
||||
except Exception as err:
|
||||
print(json.dumps({"error": str(err)}), file=sys.stderr)
|
||||
sys.exit(1)
|
||||
Executable
+64
@@ -0,0 +1,64 @@
|
||||
#!/bin/zsh
|
||||
set -euo pipefail
|
||||
|
||||
LABEL="com.local.ksay-kokoro"
|
||||
PROJECT_DIR="$(cd "$(dirname "$0")/.." && pwd)"
|
||||
PYTHON_BIN="$PROJECT_DIR/.venv/bin/python"
|
||||
AGENT_DIR="$HOME/Library/LaunchAgents"
|
||||
PLIST="$AGENT_DIR/$LABEL.plist"
|
||||
LOG_DIR="$PROJECT_DIR/.logs"
|
||||
DOMAIN="gui/$(id -u)"
|
||||
CLI_LINK="/opt/homebrew/bin/ksay"
|
||||
|
||||
if [[ ! -x "$PYTHON_BIN" ]]; then
|
||||
echo "Missing $PYTHON_BIN. Run npm run python:install first." >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
mkdir -p "$AGENT_DIR" "$LOG_DIR"
|
||||
|
||||
cat > "$PLIST" <<PLIST
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
|
||||
<plist version="1.0">
|
||||
<dict>
|
||||
<key>Label</key>
|
||||
<string>$LABEL</string>
|
||||
<key>ProgramArguments</key>
|
||||
<array>
|
||||
<string>$PYTHON_BIN</string>
|
||||
<string>$PROJECT_DIR/scripts/ksay_server.py</string>
|
||||
</array>
|
||||
<key>WorkingDirectory</key>
|
||||
<string>$PROJECT_DIR</string>
|
||||
<key>RunAtLoad</key>
|
||||
<true/>
|
||||
<key>KeepAlive</key>
|
||||
<true/>
|
||||
<key>ThrottleInterval</key>
|
||||
<integer>2</integer>
|
||||
<key>StandardOutPath</key>
|
||||
<string>$LOG_DIR/ksay.out.log</string>
|
||||
<key>StandardErrorPath</key>
|
||||
<string>$LOG_DIR/ksay.err.log</string>
|
||||
</dict>
|
||||
</plist>
|
||||
PLIST
|
||||
|
||||
plutil -lint "$PLIST"
|
||||
launchctl bootout "$DOMAIN" "$PLIST" 2>/dev/null || true
|
||||
launchctl bootstrap "$DOMAIN" "$PLIST"
|
||||
launchctl enable "$DOMAIN/$LABEL"
|
||||
launchctl kickstart -k "$DOMAIN/$LABEL"
|
||||
|
||||
if [[ -d "$(dirname "$CLI_LINK")" && -w "$(dirname "$CLI_LINK")" ]]; then
|
||||
ln -sf "$PROJECT_DIR/bin/ksay" "$CLI_LINK"
|
||||
echo "Linked CLI: $CLI_LINK"
|
||||
else
|
||||
echo "Could not link $CLI_LINK. Add $PROJECT_DIR/bin to PATH or link bin/ksay manually." >&2
|
||||
fi
|
||||
|
||||
echo "Installed $LABEL"
|
||||
echo "Endpoint: http://127.0.0.1:7332"
|
||||
echo "Try: ksay --start"
|
||||
echo "Logs: $LOG_DIR/ksay.*.log"
|
||||
+51
@@ -0,0 +1,51 @@
|
||||
#!/bin/zsh
|
||||
set -euo pipefail
|
||||
|
||||
LABEL="com.local.macmini-mcp"
|
||||
PROJECT_DIR="$(cd "$(dirname "$0")/.." && pwd)"
|
||||
NODE_BIN="$(command -v node)"
|
||||
AGENT_DIR="$HOME/Library/LaunchAgents"
|
||||
PLIST="$AGENT_DIR/$LABEL.plist"
|
||||
LOG_DIR="$PROJECT_DIR/.logs"
|
||||
DOMAIN="gui/$(id -u)"
|
||||
|
||||
mkdir -p "$AGENT_DIR" "$LOG_DIR"
|
||||
|
||||
cat > "$PLIST" <<PLIST
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
|
||||
<plist version="1.0">
|
||||
<dict>
|
||||
<key>Label</key>
|
||||
<string>$LABEL</string>
|
||||
<key>ProgramArguments</key>
|
||||
<array>
|
||||
<string>$NODE_BIN</string>
|
||||
<string>--watch</string>
|
||||
<string>src/http.js</string>
|
||||
</array>
|
||||
<key>WorkingDirectory</key>
|
||||
<string>$PROJECT_DIR</string>
|
||||
<key>RunAtLoad</key>
|
||||
<true/>
|
||||
<key>KeepAlive</key>
|
||||
<true/>
|
||||
<key>ThrottleInterval</key>
|
||||
<integer>2</integer>
|
||||
<key>StandardOutPath</key>
|
||||
<string>$LOG_DIR/service.out.log</string>
|
||||
<key>StandardErrorPath</key>
|
||||
<string>$LOG_DIR/service.err.log</string>
|
||||
</dict>
|
||||
</plist>
|
||||
PLIST
|
||||
|
||||
plutil -lint "$PLIST"
|
||||
launchctl bootout "$DOMAIN" "$PLIST" 2>/dev/null || true
|
||||
launchctl bootstrap "$DOMAIN" "$PLIST"
|
||||
launchctl enable "$DOMAIN/$LABEL"
|
||||
launchctl kickstart -k "$DOMAIN/$LABEL"
|
||||
|
||||
echo "Installed $LABEL"
|
||||
echo "Endpoint: configured by .env (see the service log for the listening URL)"
|
||||
echo "Logs: $LOG_DIR"
|
||||
+210
@@ -0,0 +1,210 @@
|
||||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import signal
|
||||
import sys
|
||||
import time
|
||||
import traceback
|
||||
import uuid
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from pathlib import Path
|
||||
from threading import Lock
|
||||
from typing import Any
|
||||
|
||||
|
||||
PROJECT_DIR = Path(__file__).resolve().parents[1]
|
||||
OUTPUT_DIR = Path(os.environ.get("KSAY_OUTPUT_DIR", PROJECT_DIR / "generated-audio"))
|
||||
DEFAULT_HOST = os.environ.get("KSAY_HOST", "127.0.0.1")
|
||||
DEFAULT_PORT = int(os.environ.get("KSAY_PORT", "7332"))
|
||||
DEFAULT_MODEL = os.environ.get("KSAY_MODEL", "mlx-community/Kokoro-82M-8bit")
|
||||
DEFAULT_VOICE = os.environ.get("KSAY_VOICE", "af_heart")
|
||||
DEFAULT_LANG_CODE = os.environ.get("KSAY_LANG_CODE", "a")
|
||||
MAX_TEXT_CHARS = int(os.environ.get("KSAY_MAX_TEXT_CHARS", "8000"))
|
||||
|
||||
|
||||
class KokoroEngine:
|
||||
def __init__(self, model_name: str):
|
||||
self.model_name = model_name
|
||||
self.model = None
|
||||
self.loaded_at = None
|
||||
self.load_seconds = None
|
||||
self.lock = Lock()
|
||||
|
||||
def load(self) -> None:
|
||||
started = time.perf_counter()
|
||||
from mlx_audio.tts.utils import load_model
|
||||
|
||||
self.model = load_model(self.model_name)
|
||||
self.loaded_at = time.time()
|
||||
self.load_seconds = time.perf_counter() - started
|
||||
|
||||
def synthesize(
|
||||
self,
|
||||
*,
|
||||
text: str,
|
||||
voice: str,
|
||||
speed: float,
|
||||
lang_code: str,
|
||||
output: str | None,
|
||||
) -> dict[str, Any]:
|
||||
if self.model is None:
|
||||
raise RuntimeError("Kokoro model is not loaded.")
|
||||
|
||||
clean_text = text.strip()
|
||||
if not clean_text:
|
||||
raise ValueError("text is required.")
|
||||
clean_text = clean_text[:MAX_TEXT_CHARS]
|
||||
|
||||
output_path = resolve_output_path(output)
|
||||
started = time.perf_counter()
|
||||
|
||||
with self.lock:
|
||||
audio_chunks = []
|
||||
sample_rate = None
|
||||
for result in self.model.generate(
|
||||
text=clean_text,
|
||||
voice=voice,
|
||||
speed=speed,
|
||||
lang_code=lang_code,
|
||||
):
|
||||
audio_chunks.append(result.audio)
|
||||
sample_rate = result.sample_rate
|
||||
|
||||
if not audio_chunks:
|
||||
raise RuntimeError("Kokoro did not return audio.")
|
||||
|
||||
import mlx.core as mx
|
||||
import numpy as np
|
||||
from mlx_audio.audio_io import write as audio_write
|
||||
|
||||
audio = (
|
||||
mx.concatenate(audio_chunks, axis=0)
|
||||
if len(audio_chunks) > 1
|
||||
else audio_chunks[0]
|
||||
)
|
||||
audio_write(str(output_path), np.array(audio), sample_rate, format="wav")
|
||||
|
||||
elapsed = time.perf_counter() - started
|
||||
return {
|
||||
"ok": True,
|
||||
"filePath": str(output_path),
|
||||
"model": self.model_name,
|
||||
"voice": voice,
|
||||
"speed": speed,
|
||||
"langCode": lang_code,
|
||||
"sampleRate": sample_rate,
|
||||
"segments": len(audio_chunks),
|
||||
"seconds": round(elapsed, 3),
|
||||
"characters": len(clean_text),
|
||||
}
|
||||
|
||||
|
||||
def resolve_output_path(output: str | None) -> Path:
|
||||
if output:
|
||||
path = Path(output).expanduser()
|
||||
if path.suffix.lower() != ".wav":
|
||||
path = path.with_suffix(".wav")
|
||||
if not path.is_absolute():
|
||||
path = (Path.cwd() / path).resolve()
|
||||
else:
|
||||
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
path = OUTPUT_DIR / f"ksay-{int(time.time() * 1000)}-{uuid.uuid4().hex[:6]}.wav"
|
||||
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
return path
|
||||
|
||||
|
||||
def parse_json_body(handler: BaseHTTPRequestHandler) -> dict[str, Any]:
|
||||
length = int(handler.headers.get("content-length", "0"))
|
||||
if length <= 0:
|
||||
return {}
|
||||
body = handler.rfile.read(length)
|
||||
return json.loads(body.decode("utf-8"))
|
||||
|
||||
|
||||
def make_handler(engine: KokoroEngine):
|
||||
class KsayHandler(BaseHTTPRequestHandler):
|
||||
server_version = "ksay-kokoro/0.1"
|
||||
|
||||
def log_message(self, fmt: str, *args: Any) -> None:
|
||||
sys.stderr.write("%s - %s\n" % (self.log_date_time_string(), fmt % args))
|
||||
|
||||
def write_json(self, status: int, value: dict[str, Any]) -> None:
|
||||
body = json.dumps(value).encode("utf-8")
|
||||
self.send_response(status)
|
||||
self.send_header("content-type", "application/json")
|
||||
self.send_header("content-length", str(len(body)))
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
def do_GET(self) -> None:
|
||||
if self.path == "/health":
|
||||
self.write_json(
|
||||
200,
|
||||
{
|
||||
"ok": True,
|
||||
"model": engine.model_name,
|
||||
"loaded": engine.model is not None,
|
||||
"loadedAt": engine.loaded_at,
|
||||
"loadSeconds": engine.load_seconds,
|
||||
"defaultVoice": DEFAULT_VOICE,
|
||||
"defaultLangCode": DEFAULT_LANG_CODE,
|
||||
},
|
||||
)
|
||||
return
|
||||
self.write_json(404, {"ok": False, "error": "not found"})
|
||||
|
||||
def do_POST(self) -> None:
|
||||
if self.path != "/say":
|
||||
self.write_json(404, {"ok": False, "error": "not found"})
|
||||
return
|
||||
|
||||
try:
|
||||
payload = parse_json_body(self)
|
||||
result = engine.synthesize(
|
||||
text=str(payload.get("text", "")),
|
||||
voice=str(payload.get("voice") or DEFAULT_VOICE),
|
||||
speed=float(payload.get("speed") or 1.0),
|
||||
lang_code=str(payload.get("langCode") or DEFAULT_LANG_CODE),
|
||||
output=payload.get("output"),
|
||||
)
|
||||
self.write_json(200, result)
|
||||
except Exception as exc:
|
||||
traceback.print_exc()
|
||||
self.write_json(500, {"ok": False, "error": str(exc)})
|
||||
|
||||
return KsayHandler
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description="Warm Kokoro TTS server for ksay.")
|
||||
parser.add_argument("--host", default=DEFAULT_HOST)
|
||||
parser.add_argument("--port", type=int, default=DEFAULT_PORT)
|
||||
parser.add_argument("--model", default=DEFAULT_MODEL)
|
||||
args = parser.parse_args()
|
||||
|
||||
engine = KokoroEngine(args.model)
|
||||
print(f"Loading {args.model}...", flush=True)
|
||||
engine.load()
|
||||
print(
|
||||
f"ksay Kokoro ready on http://{args.host}:{args.port} "
|
||||
f"after {engine.load_seconds:.2f}s",
|
||||
flush=True,
|
||||
)
|
||||
|
||||
httpd = ThreadingHTTPServer((args.host, args.port), make_handler(engine))
|
||||
|
||||
def shutdown(_signum: int, _frame: Any) -> None:
|
||||
httpd.shutdown()
|
||||
|
||||
signal.signal(signal.SIGTERM, shutdown)
|
||||
signal.signal(signal.SIGINT, shutdown)
|
||||
httpd.serve_forever()
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Executable
+8
@@ -0,0 +1,8 @@
|
||||
#!/bin/zsh
|
||||
set -euo pipefail
|
||||
|
||||
LABEL="com.local.ksay-kokoro"
|
||||
DOMAIN="gui/$(id -u)"
|
||||
|
||||
launchctl kickstart -k "$DOMAIN/$LABEL"
|
||||
echo "Restarted $LABEL"
|
||||
+8
@@ -0,0 +1,8 @@
|
||||
#!/bin/zsh
|
||||
set -euo pipefail
|
||||
|
||||
LABEL="com.local.macmini-mcp"
|
||||
DOMAIN="gui/$(id -u)"
|
||||
|
||||
launchctl kickstart -k "$DOMAIN/$LABEL"
|
||||
echo "Restarted $LABEL"
|
||||
+11
@@ -0,0 +1,11 @@
|
||||
#!/bin/zsh
|
||||
set -euo pipefail
|
||||
|
||||
PROJECT_DIR="$(cd "$(dirname "$0")/.." && pwd)"
|
||||
cd "$PROJECT_DIR"
|
||||
|
||||
if [[ ! -x .venv/bin/python ]]; then
|
||||
python3 -m venv .venv
|
||||
fi
|
||||
|
||||
.venv/bin/python -m pip install -r requirements.txt
|
||||
+7
@@ -0,0 +1,7 @@
|
||||
#!/bin/zsh
|
||||
set -euo pipefail
|
||||
|
||||
LABEL="com.local.macmini-mcp"
|
||||
DOMAIN="gui/$(id -u)"
|
||||
|
||||
launchctl print "$DOMAIN/$LABEL"
|
||||
+10
@@ -0,0 +1,10 @@
|
||||
#!/bin/zsh
|
||||
set -euo pipefail
|
||||
|
||||
LABEL="com.local.macmini-mcp"
|
||||
PLIST="$HOME/Library/LaunchAgents/$LABEL.plist"
|
||||
DOMAIN="gui/$(id -u)"
|
||||
|
||||
launchctl bootout "$DOMAIN" "$PLIST" 2>/dev/null || true
|
||||
rm -f "$PLIST"
|
||||
echo "Uninstalled $LABEL"
|
||||
Reference in New Issue
Block a user