56 lines
2.3 KiB
Python
56 lines
2.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Minimal Matrix status-room bot: mirrors open/closed into a room's name (and avatar).
|
|
|
|
Usage:
|
|
status.py open|closed set once (door switch, button, cron, ...)
|
|
status.py --spaceapi URL [SECS] poll a SpaceAPI endpoint every SECS seconds (default 60)
|
|
|
|
Env: MATRIX_HS (https://matrix.example.org), MATRIX_TOKEN, MATRIX_ROOM (!id:server)
|
|
Optional: NAME_OPEN / NAME_CLOSED, AVATAR_OPEN / AVATAR_CLOSED (mxc:// URIs)
|
|
"""
|
|
import json, os, sys, time, urllib.error, urllib.parse, urllib.request
|
|
|
|
HS, TOKEN, ROOM = (os.environ[k] for k in ("MATRIX_HS", "MATRIX_TOKEN", "MATRIX_ROOM"))
|
|
NAMES = {True: os.getenv("NAME_OPEN", "🟢 Space is open"), False: os.getenv("NAME_CLOSED", "🔴 Space is closed")}
|
|
AVATARS = {True: os.getenv("AVATAR_OPEN"), False: os.getenv("AVATAR_CLOSED")}
|
|
|
|
|
|
def state(method, event_type, body=None):
|
|
url = f"{HS.rstrip('/')}/_matrix/client/v3/rooms/{urllib.parse.quote(ROOM, safe='')}/state/{event_type}"
|
|
req = urllib.request.Request(url, method=method, data=body and json.dumps(body).encode(),
|
|
headers={"Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json"})
|
|
with urllib.request.urlopen(req, timeout=10) as r:
|
|
return json.load(r)
|
|
|
|
|
|
def set_status(is_open):
|
|
try:
|
|
if state("GET", "m.room.name").get("name") == NAMES[is_open]:
|
|
return # unchanged: don't add a "changed the room name" line
|
|
except urllib.error.HTTPError as e:
|
|
if e.code != 404: # 404 = room has no name yet
|
|
raise
|
|
state("PUT", "m.room.name", {"name": NAMES[is_open]})
|
|
if AVATARS[is_open]:
|
|
state("PUT", "m.room.avatar", {"url": AVATARS[is_open]})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
args = sys.argv[1:]
|
|
if args in (["open"], ["closed"]):
|
|
set_status(args[0] == "open")
|
|
elif args[:1] == ["--spaceapi"] and len(args) in (2, 3):
|
|
last = None
|
|
while True:
|
|
try:
|
|
with urllib.request.urlopen(args[1], timeout=10) as r:
|
|
is_open = bool(json.load(r)["state"]["open"])
|
|
if is_open != last:
|
|
set_status(is_open)
|
|
last = is_open
|
|
except Exception as e:
|
|
print("error:", e, file=sys.stderr)
|
|
time.sleep(int(args[2]) if len(args) == 3 else 60)
|
|
else:
|
|
sys.exit(__doc__)
|