114 lines
3.4 KiB
Python
114 lines
3.4 KiB
Python
from flask import Flask, request, jsonify
|
||
from wakeonlan import send_magic_packet
|
||
import logging
|
||
import os
|
||
import subprocess
|
||
|
||
app = Flask(__name__)
|
||
logging.basicConfig(level=logging.INFO)
|
||
|
||
# Format: name:mac:ip,name2:mac2:ip2
|
||
# Example: desktop:AA:BB:CC:DD:EE:FF:192.168.1.10
|
||
# Set WOL_DEVICES in .env – never commit real MACs/IPs
|
||
_WOL_DEVICES_RAW = os.getenv("WOL_DEVICES", "")
|
||
BROADCAST_IP = os.getenv("WOL_BROADCAST_IP", "255.255.255.255")
|
||
WOL_PACKET_COUNT = int(os.getenv("WOL_PACKET_COUNT", "3"))
|
||
|
||
|
||
def _parse_devices() -> tuple[dict[str, str], dict[str, str]]:
|
||
"""Parse WOL_DEVICES env into devices (name->mac) and device_ips (name->ip).
|
||
Format: name:mac:ip (mac has colons, so use name:XX:XX:XX:XX:XX:XX:ip)
|
||
Example: desktop:AA:BB:CC:DD:EE:FF:192.168.1.10
|
||
"""
|
||
devices: dict[str, str] = {}
|
||
device_ips: dict[str, str] = {}
|
||
for entry in _WOL_DEVICES_RAW.split(","):
|
||
entry = entry.strip()
|
||
if not entry:
|
||
continue
|
||
parts = entry.split(":")
|
||
if len(parts) >= 8:
|
||
name = parts[0]
|
||
mac = ":".join(parts[1:7])
|
||
ip = parts[7]
|
||
devices[name] = mac
|
||
device_ips[name] = ip
|
||
elif len(parts) == 3:
|
||
name, mac, ip = parts
|
||
devices[name] = mac
|
||
device_ips[name] = ip
|
||
return devices, device_ips
|
||
|
||
|
||
DEVICES, DEVICE_IPS = _parse_devices()
|
||
|
||
|
||
def _ping_host(ip: str) -> bool:
|
||
try:
|
||
subprocess.check_output(
|
||
["ping", "-c", "1", "-W", "1", ip],
|
||
stderr=subprocess.DEVNULL,
|
||
)
|
||
return True
|
||
except (subprocess.CalledProcessError, FileNotFoundError):
|
||
return False
|
||
|
||
|
||
@app.route("/wake/<device>", methods=["POST"])
|
||
def wake_device(device):
|
||
if device not in DEVICES:
|
||
return jsonify({"error": "Device not found"}), 404
|
||
|
||
mac_address = DEVICES[device]
|
||
try:
|
||
for _ in range(max(1, WOL_PACKET_COUNT)):
|
||
send_magic_packet(mac_address, ip_address=BROADCAST_IP)
|
||
app.logger.info(
|
||
f"Magic packet(s) sent to {device} ({mac_address}) via {BROADCAST_IP}"
|
||
)
|
||
return jsonify({"message": f"Wake-on-LAN packet sent to {device}"}), 200
|
||
except Exception as e:
|
||
app.logger.exception(f"Error sending magic packet to {device}")
|
||
return jsonify({"error": f"Failed to send wake packet: {str(e)}"}), 500
|
||
|
||
|
||
@app.route("/devices", methods=["GET"])
|
||
def list_devices():
|
||
return jsonify({"devices": list(DEVICES.keys())})
|
||
|
||
|
||
@app.route("/status", methods=["GET"])
|
||
def device_status():
|
||
"""Return devices with MAC, IP, and online status."""
|
||
status = {
|
||
name: {
|
||
"mac": mac,
|
||
"ip": DEVICE_IPS.get(name, ""),
|
||
"online": _ping_host(DEVICE_IPS.get(name, "")),
|
||
}
|
||
for name, mac in DEVICES.items()
|
||
}
|
||
return jsonify({"devices": status})
|
||
|
||
|
||
@app.route("/debug/<device>", methods=["GET"])
|
||
def debug_device(device):
|
||
if device not in DEVICES:
|
||
return jsonify({"error": "Device not found"}), 404
|
||
return jsonify({
|
||
"device": device,
|
||
"mac_address": DEVICES[device],
|
||
"ip": DEVICE_IPS.get(device, ""),
|
||
"broadcast_ip": BROADCAST_IP,
|
||
"online": _ping_host(DEVICE_IPS.get(device, "")),
|
||
})
|
||
|
||
|
||
@app.route("/health", methods=["GET"])
|
||
def health_check():
|
||
return jsonify({"status": "healthy"}), 200
|
||
|
||
|
||
if __name__ == "__main__":
|
||
app.run(host="0.0.0.0", port=5000, debug=False)
|