这篇记录一个很实用的小自动化：在自建 Matrix Synapse 上创建一个真正的机器人账号 @bot:666666.xyz。本服用户可以主动邀请它，发送“帮助”查看命令，发送“设置天气时间 07:30”自行设置每日天气推送时间；管理员仍然可以接收磁盘、Docker、TURN / LiveKit 等服务器监控提醒。
这个方案不是 Synapse Server Notices，而是普通 Matrix 机器人账号。好处是体验更像一个联系人：用户可以手动邀请它进私聊，普通用户只使用天气功能，服务器监控提醒则只发给管理员。
一、最终效果
完成后会得到这些文件和配置：
机器人账号:
@bot:666666.xyz
机器人 token:
/root/.config/matrix-bot/access_token
机器人密码:
/root/.config/matrix-bot/bot-password
主脚本:
/root/data/docker_data/synapse/scripts/matrix-bot-weather.py
配置文件:
/root/data/docker_data/synapse/scripts/matrix-bot-weather.json
定时包装脚本:
/root/data/docker_data/synapse/scripts/run-matrix-bot-weather.sh
天气循环脚本:
/root/data/docker_data/synapse/scripts/run-matrix-bot-weather-loop.sh
天气 OpenRC 服务:
/etc/init.d/matrix-bot-weather
聊天同步脚本:
/root/data/docker_data/synapse/scripts/matrix-bot-sync.py
聊天同步 OpenRC 服务:
/etc/init.d/matrix-bot-sync
日志:
/var/log/matrix-bot-weather.log
/var/log/matrix-bot-sync.log
监控脚本:
/root/data/docker_data/synapse/scripts/matrix-bot-monitor.py
监控配置:
/root/data/docker_data/synapse/scripts/matrix-bot-monitor.json
监控日志:
/var/log/matrix-bot-monitor.log天气推送和聊天命令使用 OpenRC 常驻服务控制。天气服务每 20 秒检查一次配置，谁到了自己的推送时间，就给谁发送天气；聊天同步服务每 20 秒处理一次邀请和命令。
rc-service matrix-bot-weather status
rc-service matrix-bot-sync status服务器监控提醒仍然由 root crontab 控制：
*/5 * * * * /root/data/docker_data/synapse/scripts/run-matrix-bot-monitor.sh其中 matrix-bot-weather.py 只负责天气内容和到点判断，matrix-bot-sync.py 负责邀请和聊天命令，matrix-bot-monitor.py 每 5 分钟检查服务器异常。天气和监控共用同一个机器人账号，但普通用户只会收到天气，监控提醒只发给监控配置里的管理员。
天气数据使用 Open-Meteo，无需申请 API key。脚本里写的是上海坐标：
latitude: 31.2304
longitude: 121.4737
timezone: Asia/Shanghai二、前置条件
本文假设 Synapse 在本机 8008 提供客户端 API：
http://127.0.0.1:8008并且有一个可用的 Synapse 管理员 access token：
/root/.config/synapse-admin/access_token先在服务器上确认 Synapse 正常：
curl -sS -o /tmp/synapse_versions.json -w 'http_code=%{http_code}\n' \
http://127.0.0.1:8008/_matrix/client/versions
rm -f /tmp/synapse_versions.json返回 http_code=200 就可以继续。
再准备脚本目录：
mkdir -p /root/data/docker_data/synapse/scripts
mkdir -p /root/.config/matrix-bot
chmod 700 /root/.config/matrix-bot三、创建安装脚本
注意：下面的安装脚本用于创建账号、保存 token 和生成基础配置；后文第七节会把机器人升级为可交互模式，最终运行以 matrix-bot-sync.py、新版 matrix-bot-weather.py 和 weather_users 配置为准。
下面这个安装脚本会做几件事：
- 创建或复用
@bot:666666.xyz。 - 给机器人设置显示名“机器人”。
- 为机器人生成本地保存的随机密码。
- 登录机器人并保存 access token。
- 写入天气推送配置文件。
- 写入主推送脚本。
- 写入天气调度脚本。
- 写入聊天同步脚本，用于自动接受本服用户邀请和处理命令。
- 配置天气和聊天同步后台服务。
创建文件：
vi /root/data/docker_data/synapse/scripts/install-matrix-bot.py写入：
#!/usr/bin/env python3
import json
import os
import secrets
import subprocess
import urllib.error
import urllib.parse
import urllib.request
BASE = "http://127.0.0.1:8008"
SERVER_NAME = "666666.xyz"
BOT_LOCALPART = "bot"
BOT_MXID = f"@{BOT_LOCALPART}:{SERVER_NAME}"
TARGET_MXID = "@honey:666666.xyz"
TOKEN_PATH = "/root/.config/synapse-admin/access_token"
BOT_TOKEN_PATH = "/root/.config/matrix-bot/access_token"
BOT_PASSWORD_PATH = "/root/.config/matrix-bot/bot-password"
SCRIPT_PATH = "/root/data/docker_data/synapse/scripts/matrix-bot-weather.py"
CONFIG_PATH = "/root/data/docker_data/synapse/scripts/matrix-bot-weather.json"
def req(method, url, token=None, data=None):
headers = {}
body = None
if token:
headers["Authorization"] = f"Bearer {token}"
if data is not None:
body = json.dumps(data, ensure_ascii=False).encode("utf-8")
headers["Content-Type"] = "application/json"
request = urllib.request.Request(url, data=body, headers=headers, method=method)
try:
with urllib.request.urlopen(request, timeout=20) as resp:
raw = resp.read().decode("utf-8")
return resp.status, json.loads(raw) if raw else {}
except urllib.error.HTTPError as e:
raw = e.read().decode("utf-8", errors="replace")
try:
payload = json.loads(raw)
except Exception:
payload = {"raw": raw}
return e.code, payload
with open(TOKEN_PATH, "r", encoding="utf-8") as f:
admin_token = f.read().strip()
os.makedirs(os.path.dirname(BOT_TOKEN_PATH), mode=0o700, exist_ok=True)
if not os.path.exists(BOT_PASSWORD_PATH):
with open(BOT_PASSWORD_PATH, "w", encoding="utf-8") as f:
f.write(secrets.token_urlsafe(36) + "\n")
os.chmod(BOT_PASSWORD_PATH, 0o600)
with open(BOT_PASSWORD_PATH, "r", encoding="utf-8") as f:
bot_password = f.read().strip()
encoded_bot = urllib.parse.quote(BOT_MXID, safe="")
status, payload = req(
"GET",
f"{BASE}/_synapse/admin/v2/users/{encoded_bot}",
admin_token,
)
if status == 404:
status, payload = req(
"PUT",
f"{BASE}/_synapse/admin/v2/users/{encoded_bot}",
admin_token,
{
"password": bot_password,
"displayname": "机器人",
"admin": False,
"deactivated": False,
"user_type": None,
},
)
if status not in (200, 201):
raise SystemExit(f"create bot failed: HTTP {status} {payload}")
print(f"created {BOT_MXID}")
elif status == 200:
status, payload = req(
"PUT",
f"{BASE}/_synapse/admin/v2/users/{encoded_bot}",
admin_token,
{
"password": bot_password,
"displayname": "机器人",
"admin": False,
"deactivated": False,
"user_type": None,
},
)
if status not in (200, 201):
raise SystemExit(f"update bot failed: HTTP {status} {payload}")
print(f"reused {BOT_MXID}")
else:
raise SystemExit(f"check bot failed: HTTP {status} {payload}")
status, payload = req(
"POST",
f"{BASE}/_matrix/client/v3/login",
None,
{
"type": "m.login.password",
"identifier": {"type": "m.id.user", "user": BOT_LOCALPART},
"password": bot_password,
"initial_device_display_name": "matrix-bot",
},
)
if status != 200 or "access_token" not in payload:
raise SystemExit(f"bot login failed: HTTP {status} {payload}")
with open(BOT_TOKEN_PATH, "w", encoding="utf-8") as f:
f.write(payload["access_token"] + "\n")
os.chmod(BOT_TOKEN_PATH, 0o600)
config = {
"homeserver": BASE,
"server_name": SERVER_NAME,
"bot_user_id": BOT_MXID,
"token_path": BOT_TOKEN_PATH,
"city": "中国上海",
"latitude": 31.2304,
"longitude": 121.4737,
"timezone": "Asia/Shanghai",
"default_weather_time": "05:00",
"allow_local_users_only": True,
"weather_users": {
TARGET_MXID: {
"enabled": True,
"time": "05:00",
},
},
"direct_rooms": {},
"state_path": "/root/data/docker_data/synapse/scripts/matrix-bot-weather.state.json",
"sync_state_path": "/root/data/docker_data/synapse/scripts/matrix-bot-sync.state.json",
"schedule_grace_minutes": 15,
}
if os.path.exists(CONFIG_PATH):
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
old = json.load(f)
for key in ("weather_users", "direct_rooms", "default_weather_time", "sync_state_path", "schedule_grace_minutes"):
if key in old:
config[key] = old[key]
config.setdefault("weather_users", {}).setdefault(TARGET_MXID, {"enabled": True, "time": "05:00"})
with open(CONFIG_PATH, "w", encoding="utf-8") as f:
json.dump(config, f, ensure_ascii=False, indent=2)
f.write("\n")
os.chmod(CONFIG_PATH, 0o600)
print(f"bot={BOT_MXID}")
print(f"target={TARGET_MXID}")
print(f"config={CONFIG_PATH}")先赋予执行权限：
chmod 700 /root/data/docker_data/synapse/scripts/install-matrix-bot.py这个安装脚本先负责账号和配置，主推送脚本下一节单独写，方便排查和修改。
四、创建主推送脚本
创建：
vi /root/data/docker_data/synapse/scripts/matrix-bot-weather.py写入：
#!/usr/bin/env python3
import datetime as dt
import hashlib
import html
import json
import os
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
CONFIG_PATH = "/root/data/docker_data/synapse/scripts/matrix-bot-weather.json"
WEATHER_CODES = {
0: "晴", 1: "大部晴朗", 2: "局部多云", 3: "阴", 45: "雾", 48: "雾凇",
51: "小毛毛雨", 53: "毛毛雨", 55: "较强毛毛雨", 56: "冻毛毛雨", 57: "较强冻毛毛雨",
61: "小雨", 63: "中雨", 65: "大雨", 66: "冻雨", 67: "强冻雨",
71: "小雪", 73: "中雪", 75: "大雪", 77: "雪粒",
80: "小阵雨", 81: "阵雨", 82: "强阵雨", 85: "小阵雪", 86: "强阵雪",
95: "雷雨", 96: "雷雨伴小冰雹", 99: "雷雨伴强冰雹",
}
HELP_HINT = "发送“设置天气时间 07:30”可修改推送时间，发送“帮助”可查询更多功能。"
def load_json(path, default):
try:
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
except FileNotFoundError:
return default
def save_json(path, data, mode=0o600):
tmp = path + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
f.write("\n")
os.chmod(tmp, mode)
os.replace(tmp, path)
def request(method, url, token=None, data=None, timeout=30):
headers = {}
body = None
if token:
headers["Authorization"] = f"Bearer {token}"
if data is not None:
body = json.dumps(data, ensure_ascii=False).encode("utf-8")
headers["Content-Type"] = "application/json"
req = urllib.request.Request(url, data=body, headers=headers, method=method)
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
raw = resp.read().decode("utf-8", errors="replace")
return resp.status, json.loads(raw) if raw else {}
except urllib.error.HTTPError as e:
raw = e.read().decode("utf-8", errors="replace")
try:
payload = json.loads(raw)
except Exception:
payload = {"raw": raw}
return e.code, payload
except Exception as e:
return 0, {"error": str(e)}
def quote(value):
return urllib.parse.quote(value, safe="")
def token(config):
with open(config["token_path"], "r", encoding="utf-8") as f:
return f.read().strip()
def get_weather(config):
params = urllib.parse.urlencode({
"latitude": config["latitude"],
"longitude": config["longitude"],
"timezone": config.get("timezone", "Asia/Shanghai"),
"forecast_days": 3,
"daily": "weather_code,temperature_2m_max,temperature_2m_min,precipitation_probability_max,precipitation_sum,wind_speed_10m_max",
})
url = "https://api.open-meteo.com/v1/forecast?" + params
with urllib.request.urlopen(url, timeout=30) as resp:
data = json.loads(resp.read().decode("utf-8"))
daily = data["daily"]
lines = [f"{config['city']}天气预报", ""]
today = dt.date.today().isoformat()
alerts = []
advice = []
for i, day in enumerate(daily["time"]):
label = "今天" if day == today else ("明天" if i == 1 else "后天")
code = daily["weather_code"][i]
desc = WEATHER_CODES.get(code, f"天气代码 {code}")
tmin = daily["temperature_2m_min"][i]
tmax = daily["temperature_2m_max"][i]
pop = daily["precipitation_probability_max"][i]
rain = daily["precipitation_sum"][i]
wind = daily["wind_speed_10m_max"][i]
lines.append(
f"{label} {day}: {desc}，{tmin:.0f}-{tmax:.0f}℃，"
f"降水概率 {pop}% ，降水量 {rain:.1f} mm，最大风速 {wind:.0f} km/h"
)
if i == 0:
if tmax >= 35:
alerts.append("高温提醒：今天最高温达到 35℃ 以上，注意防暑补水。")
if tmin <= 0:
alerts.append("低温提醒：今天最低温 0℃ 以下，注意防寒防冻。")
if rain >= 20 or pop >= 70:
alerts.append("降雨提醒：今天降水概率较高或雨量较大，出门建议带伞。")
if wind >= 40:
alerts.append("大风提醒：今天风力偏强，骑车和户外活动注意安全。")
if code in (95, 96, 99):
alerts.append("雷雨提醒：今天可能有雷雨，尽量避开强对流时段外出。")
if tmax >= 30:
advice.append("穿衣建议：短袖、轻薄透气衣物为主，注意防晒。")
elif tmin <= 8:
advice.append("穿衣建议：建议外套或薄棉服，早晚注意保暖。")
elif tmin <= 15:
advice.append("穿衣建议：长袖加薄外套比较稳妥。")
else:
advice.append("穿衣建议：轻薄长袖或短袖均可，按体感调整。")
if (tmax - tmin) >= 10:
advice.append("温差提醒：昼夜温差较大，早晚可多备一件外套。")
if pop >= 50 or rain > 0:
advice.append("出行提醒：有降水可能，建议带伞。")
if alerts:
lines.extend(["", "天气提醒："] + [f"- {x}" for x in alerts])
if advice:
lines.extend(["", "生活建议："] + [f"- {x}" for x in advice])
lines.extend(["", "数据源：Open-Meteo。", HELP_HINT])
return "\n".join(lines)
def send_room(config, token_value, room_id, body):
txn = "weather-" + dt.datetime.now(dt.timezone.utc).strftime("%Y%m%dT%H%M%S")
txn += "-" + hashlib.sha1((room_id + body).encode("utf-8")).hexdigest()[:10]
html_body = "<br>".join(html.escape(line) for line in body.splitlines())
content = {"msgtype": "m.text", "body": body, "format": "org.matrix.custom.html", "formatted_body": html_body}
url = f"{config['homeserver']}/_matrix/client/v3/rooms/{quote(room_id)}/send/m.room.message/{txn}"
status, payload = request("PUT", url, token_value, content)
if status not in (200, 201):
raise RuntimeError(f"send to {room_id} failed: HTTP {status} {payload}")
def create_dm(config, token_value, user_id):
status, payload = request("POST", f"{config['homeserver']}/_matrix/client/v3/createRoom", token_value, {
"preset": "trusted_private_chat",
"is_direct": True,
"invite": [user_id],
"name": "天气预报",
"topic": "天气查询与每日自动推送",
})
if status not in (200, 201) or "room_id" not in payload:
raise RuntimeError(f"create DM for {user_id} failed: HTTP {status} {payload}")
return payload["room_id"]
def get_room(config, token_value, user_id):
rooms = config.setdefault("direct_rooms", {})
room_id = rooms.get(user_id)
if room_id:
return room_id
room_id = create_dm(config, token_value, user_id)
rooms[user_id] = room_id
save_json(CONFIG_PATH, config)
return room_id
def parse_hhmm(value):
hour_text, minute_text = str(value).split(":", 1)
hour = int(hour_text)
minute = int(minute_text)
if not (0 <= hour <= 23 and 0 <= minute <= 59):
raise ValueError(f"invalid HH:MM value: {value}")
return hour, minute
def normalized_hhmm(value):
hour, minute = parse_hhmm(value)
return f"{hour:02d}:{minute:02d}"
def sent_key(today, time_value):
return f"{today} {normalized_hhmm(time_value)}"
def due_users(config, now=None):
now = now or dt.datetime.now()
today = now.date().isoformat()
state = load_json(config["state_path"], {})
sent = state.setdefault("last_sent", {})
grace_minutes = int(config.get("schedule_grace_minutes", 15))
users = []
for user_id, settings in config.get("weather_users", {}).items():
if not settings.get("enabled", True):
continue
time_value = settings.get("time", config.get("default_weather_time", "05:00"))
try:
hour, minute = parse_hhmm(time_value)
except Exception as exc:
print(f"skip {user_id}: invalid weather time {time_value!r}: {exc}", file=sys.stderr)
continue
if sent.get(user_id) == sent_key(today, time_value):
continue
scheduled = now.replace(hour=hour, minute=minute, second=0, microsecond=0)
delta_seconds = (now - scheduled).total_seconds()
if 0 <= delta_seconds < grace_minutes * 60:
users.append(user_id)
return users, state, today
def send_weather_to_users(config, user_ids, mark_sent=False, state=None, today=None):
token_value = token(config)
body = get_weather(config)
sent = 0
for user_id in user_ids:
room_id = get_room(config, token_value, user_id)
send_room(config, token_value, room_id, body)
sent += 1
if mark_sent and state is not None and today is not None:
user_time = config.get("weather_users", {}).get(user_id, {}).get("time", config.get("default_weather_time", "05:00"))
state.setdefault("last_sent", {})[user_id] = sent_key(today, user_time)
time.sleep(0.2)
if mark_sent and state is not None:
save_json(config["state_path"], state)
return sent
def main():
config = load_json(CONFIG_PATH, None)
if not config:
raise SystemExit(f"missing config: {CONFIG_PATH}")
if len(sys.argv) >= 2 and sys.argv[1] == "--send-now":
users = sys.argv[2:] or list(config.get("weather_users", {}).keys())
print(f"sent weather forecast to {send_weather_to_users(config, users)} room(s)")
return
users, state, today = due_users(config)
if not users:
print("weather schedule: no due users")
return
print(f"sent weather forecast to {send_weather_to_users(config, users, True, state, today)} room(s)")
if __name__ == "__main__":
main()赋予执行权限并检查语法：
chmod 700 /root/data/docker_data/synapse/scripts/matrix-bot-weather.py
python3 -m py_compile /root/data/docker_data/synapse/scripts/matrix-bot-weather.py五、运行安装脚本
现在运行安装脚本，创建账号、保存 token、生成配置：
/root/data/docker_data/synapse/scripts/install-matrix-bot.py正常输出类似：
created @bot:666666.xyz
bot=@bot:666666.xyz
target=@honey:666666.xyz
config=/root/data/docker_data/synapse/scripts/matrix-bot-weather.json如果机器人已经存在，则会显示：
reused @bot:666666.xyz六、手动发送测试
先手动发一次天气，确认 token、房间和天气 API 都正常：
/root/data/docker_data/synapse/scripts/matrix-bot-weather.py --send-now成功时会输出：
sent weather forecast to 1 room(s)此时目标用户应该能在 Matrix 客户端里看到 @bot:666666.xyz 发来的天气预报。
消息内容大致是：
中国上海天气预报
今天 2026-05-25: 阴，18-26℃，降水概率 20% ，降水量 0.0 mm，最大风速 15 km/h
明天 2026-05-26: 小雨，19-24℃，降水概率 60% ，降水量 3.2 mm，最大风速 18 km/h
后天 2026-05-27: 局部多云，18-28℃，降水概率 10% ，降水量 0.0 mm，最大风速 16 km/h
数据源：Open-Meteo。
发送“天气”可立即查询，发送“设置天气时间 07:30”可修改推送时间。具体天气以 Open-Meteo 当天返回为准。
七、配置天气调度与聊天同步
这一节把机器人变成真正可交互的联系人。天气推送使用 OpenRC 常驻服务，每 20 秒检查一次所有用户的设置时间；聊天同步也使用 OpenRC 常驻服务，每 20 秒处理一次邀请和命令。这样用户刚发完“设置天气时间 07:30”，到点后就由后台服务自动推送，不依赖 cron 正好踩中某一分钟。
先创建聊天同步脚本：
vi /root/data/docker_data/synapse/scripts/matrix-bot-sync.py写入：
#!/usr/bin/env python3
import datetime as dt
import hashlib
import html
import importlib.util
import json
import os
import re
import urllib.error
import urllib.parse
import urllib.request
CONFIG_PATH = "/root/data/docker_data/synapse/scripts/matrix-bot-weather.json"
WEATHER_SCRIPT_PATH = "/root/data/docker_data/synapse/scripts/matrix-bot-weather.py"
WELCOME = """你好，我是天气机器人。
你可以这样使用我：
天气
立即查看今天和未来两天天气
设置天气时间 07:30
设置每天自动推送时间
天气时间
查看当前推送时间
关闭天气
停止每日天气推送
开启天气
恢复每日天气推送
帮助
再次查看这份说明"""
def load_json(path, default):
try:
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
except FileNotFoundError:
return default
def save_json(path, data, mode=0o600):
tmp = path + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
f.write("\n")
os.chmod(tmp, mode)
os.replace(tmp, path)
def request(method, url, token=None, data=None, timeout=30):
headers = {}
body = None
if token:
headers["Authorization"] = f"Bearer {token}"
if data is not None:
body = json.dumps(data, ensure_ascii=False).encode("utf-8")
headers["Content-Type"] = "application/json"
req = urllib.request.Request(url, data=body, headers=headers, method=method)
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
raw = resp.read().decode("utf-8", errors="replace")
return resp.status, json.loads(raw) if raw else {}
except urllib.error.HTTPError as e:
raw = e.read().decode("utf-8", errors="replace")
try:
payload = json.loads(raw)
except Exception:
payload = {"raw": raw}
return e.code, payload
except Exception as e:
return 0, {"error": str(e)}
def quote(value):
return urllib.parse.quote(value, safe="")
def token(config):
with open(config["token_path"], "r", encoding="utf-8") as f:
return f.read().strip()
def local_user(config, user_id):
return user_id.endswith(":" + config.get("server_name", ""))
def send_room(config, token_value, room_id, body):
txn = "bot-reply-" + dt.datetime.now(dt.timezone.utc).strftime("%Y%m%dT%H%M%S")
txn += "-" + hashlib.sha1((room_id + body).encode("utf-8")).hexdigest()[:10]
html_body = "<br>".join(html.escape(line) for line in body.splitlines())
content = {"msgtype": "m.text", "body": body, "format": "org.matrix.custom.html", "formatted_body": html_body}
status, payload = request("PUT", f"{config['homeserver']}/_matrix/client/v3/rooms/{quote(room_id)}/send/m.room.message/{txn}", token_value, content)
if status not in (200, 201):
print(f"send failed {room_id}: HTTP {status} {payload}")
def enable_user(config, user_id, room_id=None):
settings = config.setdefault("weather_users", {}).setdefault(user_id, {})
settings.setdefault("enabled", True)
settings.setdefault("time", config.get("default_weather_time", "05:00"))
if room_id:
config.setdefault("direct_rooms", {})[user_id] = room_id
def import_weather_module():
spec = importlib.util.spec_from_file_location("matrix_bot_weather", WEATHER_SCRIPT_PATH)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
def handle_command(config, token_value, room_id, sender, body):
text = body.strip()
lower = text.lower()
enable_user(config, sender, room_id)
if text in ("帮助", "help", "菜单"):
send_room(config, token_value, room_id, WELCOME)
elif text == "天气":
weather = import_weather_module()
send_room(config, token_value, room_id, weather.get_weather(config))
elif text == "天气时间":
s = config["weather_users"][sender]
status = "已开启" if s.get("enabled", True) else "已关闭"
send_room(config, token_value, room_id, f"你的天气推送{status}。\n当前时间：每天 {s.get('time', config.get('default_weather_time', '05:00'))}。\n城市：{config.get('city')}。")
elif text == "关闭天气":
config["weather_users"][sender]["enabled"] = False
send_room(config, token_value, room_id, "已关闭每日天气推送。发送“开启天气”可以恢复。")
elif text == "开启天气":
config["weather_users"][sender]["enabled"] = True
send_room(config, token_value, room_id, f"已开启每日天气推送。当前时间：每天 {config['weather_users'][sender].get('time', config.get('default_weather_time', '05:00'))}。")
else:
m = re.fullmatch(r"设置天气时间\s+([0-2]?\d):([0-5]\d)", text)
if m:
hour = int(m.group(1))
minute = int(m.group(2))
if hour > 23:
send_room(config, token_value, room_id, "时间格式不对。示例：设置天气时间 07:30")
else:
config["weather_users"][sender]["time"] = f"{hour:02d}:{minute:02d}"
config["weather_users"][sender]["enabled"] = True
send_room(config, token_value, room_id, f"已设置成功：每天 {hour:02d}:{minute:02d} 给你推送天气预报。")
elif lower in ("hi", "hello", "你好"):
send_room(config, token_value, room_id, WELCOME)
else:
send_room(config, token_value, room_id, "我没看懂。可以发送“帮助”查看全部命令，或发送：设置天气时间 07:30")
save_json(CONFIG_PATH, config)
def invite_sender(config, invite):
events = invite.get("invite_state", {}).get("events", [])
for event in events:
if event.get("type") == "m.room.member" and event.get("state_key") == config.get("bot_user_id"):
return event.get("sender")
return None
def main():
config = load_json(CONFIG_PATH, None)
if not config:
raise SystemExit(f"missing config: {CONFIG_PATH}")
state = load_json(config.get("sync_state_path"), {})
token_value = token(config)
params = {"timeout": 0}
if state.get("since"):
params["since"] = state["since"]
status, payload = request("GET", f"{config['homeserver']}/_matrix/client/v3/sync?{urllib.parse.urlencode(params)}", token_value, timeout=35)
if status != 200:
raise SystemExit(f"sync failed: HTTP {status} {payload}")
for room_id, invite in payload.get("rooms", {}).get("invite", {}).items():
sender = invite_sender(config, invite)
if config.get("allow_local_users_only", True) and (not sender or not local_user(config, sender)):
print(f"skip non-local invite {room_id} from {sender}")
continue
join_status, join_payload = request("POST", f"{config['homeserver']}/_matrix/client/v3/rooms/{quote(room_id)}/join", token_value, {})
if join_status in (200, 201):
enable_user(config, sender, room_id)
send_room(config, token_value, room_id, WELCOME)
else:
print(f"join failed {room_id}: HTTP {join_status} {join_payload}")
first_sync = not state.get("since")
if not first_sync:
for room_id, room in payload.get("rooms", {}).get("join", {}).items():
for event in room.get("timeline", {}).get("events", []):
if event.get("type") != "m.room.message":
continue
sender = event.get("sender")
if sender == config.get("bot_user_id") or not sender or not local_user(config, sender):
continue
content = event.get("content", {})
if content.get("msgtype") != "m.text":
continue
handle_command(config, token_value, room_id, sender, content.get("body", ""))
state["since"] = payload.get("next_batch", state.get("since"))
save_json(config.get("sync_state_path"), state)
save_json(CONFIG_PATH, config)
print("sync ok")
if __name__ == "__main__":
main()检查脚本：
chmod 700 /root/data/docker_data/synapse/scripts/matrix-bot-sync.py
python3 -m py_compile /root/data/docker_data/synapse/scripts/matrix-bot-sync.py创建天气包装脚本：
vi /root/data/docker_data/synapse/scripts/run-matrix-bot-weather.sh写入：
#!/bin/sh
LOCK=/tmp/matrix-bot-weather.lock
if mkdir "$LOCK" 2>/dev/null; then
trap 'rmdir "$LOCK"' EXIT
/root/data/docker_data/synapse/scripts/matrix-bot-weather.py >> /var/log/matrix-bot-weather.log 2>&1
fi创建聊天同步包装脚本：
vi /root/data/docker_data/synapse/scripts/run-matrix-bot-sync.sh写入：
#!/bin/sh
LOCK=/tmp/matrix-bot-sync.lock
if mkdir "$LOCK" 2>/dev/null; then
trap 'rmdir "$LOCK"' EXIT
/root/data/docker_data/synapse/scripts/matrix-bot-sync.py >> /var/log/matrix-bot-sync.log 2>&1
fi设置权限：
chmod 700 /root/data/docker_data/synapse/scripts/run-matrix-bot-weather.sh
chmod 700 /root/data/docker_data/synapse/scripts/run-matrix-bot-sync.sh再创建两个循环脚本。天气循环：
vi /root/data/docker_data/synapse/scripts/run-matrix-bot-weather-loop.sh写入：
#!/bin/sh
while true; do
/root/data/docker_data/synapse/scripts/run-matrix-bot-weather.sh
sleep 20
done聊天同步循环：
vi /root/data/docker_data/synapse/scripts/run-matrix-bot-sync-loop.sh写入：
#!/bin/sh
while true; do
/root/data/docker_data/synapse/scripts/run-matrix-bot-sync.sh
sleep 20
done设置权限：
chmod 700 /root/data/docker_data/synapse/scripts/run-matrix-bot-weather-loop.sh
chmod 700 /root/data/docker_data/synapse/scripts/run-matrix-bot-sync-loop.sh创建天气 OpenRC 服务：
vi /etc/init.d/matrix-bot-weather写入：
#!/sbin/openrc-run
name="Matrix bot weather scheduler"
description="Send scheduled weather forecasts for Matrix bot users"
command="/root/data/docker_data/synapse/scripts/run-matrix-bot-weather-loop.sh"
command_background=true
pidfile="/run/matrix-bot-weather.pid"
output_log="/var/log/matrix-bot-weather-loop.log"
error_log="/var/log/matrix-bot-weather-loop.log"
depend() {
need net
after docker
}创建聊天同步 OpenRC 服务：
vi /etc/init.d/matrix-bot-sync写入：
#!/sbin/openrc-run
name="Matrix bot sync"
description="Poll Matrix events for bot invites and weather commands"
command="/root/data/docker_data/synapse/scripts/run-matrix-bot-sync-loop.sh"
command_background=true
pidfile="/run/matrix-bot-sync.pid"
output_log="/var/log/matrix-bot-sync-loop.log"
error_log="/var/log/matrix-bot-sync-loop.log"
depend() {
need net
after docker
}启动并加入开机自启：
chmod 755 /etc/init.d/matrix-bot-weather
chmod 755 /etc/init.d/matrix-bot-sync
rc-update add matrix-bot-weather default
rc-update add matrix-bot-sync default
rc-service matrix-bot-weather restart
rc-service matrix-bot-sync restart确认服务已经运行：
rc-service matrix-bot-weather status
rc-service matrix-bot-sync status
tail -n 20 /var/log/matrix-bot-weather.log
tail -n 20 /var/log/matrix-bot-sync.log做一次真正的到点测试。先在 Matrix 客户端里给机器人发：
设置天气时间 09:40把 09:40 换成当前时间的下一分钟。机器人应该先回复：
已设置成功：每天 09:40 给你推送天气预报。然后等到这个时间点。天气服务每 20 秒检查一次，正常情况下几十秒内会自动推送天气预报。服务器上可以同时观察：
tail -f /var/log/matrix-bot-weather.log看到类似输出就说明后台服务自己完成了定时发送：
sent weather forecast to 1 room(s)
weather schedule: no due usersstate 文件也会记录这个用户当天这个时间已经发送：
cat /root/data/docker_data/synapse/scripts/matrix-bot-weather.state.json示例：
"last_sent": {
"@honey:666666.xyz": "2026-05-26 09:40"
}这里记录的是“日期 + 推送时间”，不是简单的“今天已发送”。所以同一天临时改一个新的测试时间，到了新时间仍然会推送；同一个时间当天不会重复刷屏。
八、增加服务器监控提醒
天气预报按用户自己的设置时间发送，监控提醒更适合“有事才吱声”。这里增加一个独立脚本，每 5 分钟检查一次：
- 磁盘空间预警：检查
/和/var/lib/docker，默认使用率80%警告、90%严重，或者可用空间低于10GiB警告。 - Docker 容器异常通知：检查关键容器是否存在、是否运行、健康检查是否
unhealthy，以及重启次数是否增加。 - TURN / LiveKit 状态检测：检查 coturn、LiveKit、LiveKit JWT、Element Call Web 相关容器、端口和 HTTP 探针。
创建监控配置：
vi /root/data/docker_data/synapse/scripts/matrix-bot-monitor.json写入示例：
{
"homeserver": "http://127.0.0.1:8008",
"token_path": "/root/.config/matrix-bot/access_token",
"state_path": "/root/data/docker_data/synapse/scripts/matrix-bot-monitor.state.json",
"bot_weather_state_path": "/root/data/docker_data/synapse/scripts/matrix-bot-weather.state.json",
"target_users": ["@honey:666666.xyz"],
"target_rooms": [],
"disk": {
"mounts": ["/", "/var/lib/docker"],
"warn_percent": 80,
"critical_percent": 90,
"min_free_gb": 10
},
"docker": {
"expected_containers": [
"matrix-synapse",
"nginx-proxy-manager",
"web-call",
"matrix-livekit-jwt-service",
"coturn",
"ntfy",
"livekit-server",
"postgres16"
]
},
"realtime": {
"containers": [
"coturn",
"livekit-server",
"matrix-livekit-jwt-service",
"web-call"
],
"tcp_ports": [3478, 7880, 8170, 8181],
"udp_ports": [3478],
"http_probes": [
{
"name": "LiveKit JWT",
"url": "http://127.0.0.1:8170/healthz",
"ok": [200]
},
{
"name": "LiveKit API",
"url": "http://127.0.0.1:7880/",
"ok": [200, 404]
},
{
"name": "Element Call Web",
"url": "http://127.0.0.1:8181/",
"ok": [200, 301, 302]
}
]
}
}如果你的 LiveKit 容器实际叫 livekit-server-test，就把配置里的 livekit-server 改成实际容器名。其他容器名也一样，以 docker ps --format '{{.Names}}' 为准。
创建监控脚本：
vi /root/data/docker_data/synapse/scripts/matrix-bot-monitor.py核心逻辑建议按下面这个结构写：
#!/usr/bin/env python3
import datetime as dt
import hashlib
import html
import json
import os
import socket
import subprocess
import time
import urllib.parse
import urllib.request
CONFIG_PATH = "/root/data/docker_data/synapse/scripts/matrix-bot-monitor.json"
def load_json(path, default):
try:
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
except FileNotFoundError:
return default
def save_json(path, data):
tmp = path + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
f.write("\n")
os.chmod(tmp, 0o600)
os.replace(tmp, path)
def run(cmd):
return subprocess.run(cmd, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False)
def request(method, url, token=None, data=None, timeout=20):
headers = {}
body = None
if token:
headers["Authorization"] = f"Bearer {token}"
if data is not None:
body = json.dumps(data, ensure_ascii=False).encode("utf-8")
headers["Content-Type"] = "application/json"
req = urllib.request.Request(url, data=body, headers=headers, method=method)
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
raw = resp.read().decode("utf-8", errors="replace")
if not raw:
return resp.status, {}
try:
return resp.status, json.loads(raw)
except Exception:
return resp.status, {"raw": raw[:200]}
except Exception as e:
return 0, {"error": str(e)}
def quote(value):
return urllib.parse.quote(value, safe="")
def check_disks(config):
issues = []
disk = config["disk"]
for mount in disk["mounts"]:
proc = run(["df", "-P", "-B1", mount])
if proc.returncode != 0:
issues.append(f"[disk] {mount}: df failed")
continue
parts = proc.stdout.strip().splitlines()[-1].split()
used_percent = int(parts[4].rstrip("%"))
free_gb = int(parts[3]) / 1024 / 1024 / 1024
if used_percent >= disk["critical_percent"]:
issues.append(f"[disk:critical] {parts[5]} used={used_percent}% free={free_gb:.1f}GiB")
elif used_percent >= disk["warn_percent"] or free_gb < disk["min_free_gb"]:
issues.append(f"[disk:warning] {parts[5]} used={used_percent}% free={free_gb:.1f}GiB")
return issues
def docker_inspect(names):
result = {}
for name in names:
proc = run(["docker", "inspect", "--format", "{{json .}}", name])
if proc.returncode != 0:
result[name] = {"missing": True}
continue
data = json.loads(proc.stdout)
state = data.get("State", {})
health = state.get("Health") or {}
result[name] = {
"missing": False,
"running": bool(state.get("Running")),
"status": state.get("Status", "unknown"),
"health": health.get("Status", "none"),
"restart_count": int(data.get("RestartCount", 0)),
}
return result
def check_docker(config, state):
issues = []
previous = state.get("docker_restart_counts", {})
current = {}
for name, info in docker_inspect(config["docker"]["expected_containers"]).items():
if info.get("missing"):
issues.append(f"[docker] {name}: missing")
continue
current[name] = info["restart_count"]
if not info["running"]:
issues.append(f"[docker] {name}: not running, status={info['status']}")
if info["health"] == "unhealthy":
issues.append(f"[docker] {name}: health=unhealthy")
old = previous.get(name)
if old is not None and info["restart_count"] > old:
issues.append(f"[docker] {name}: restart count {old} -> {info['restart_count']}")
state["docker_restart_counts"] = current
return issues
def tcp_open(port):
try:
with socket.create_connection(("127.0.0.1", int(port)), timeout=3):
return True
except Exception:
return False
def udp_listening(port):
proc = run(["ss", "-lun"])
return proc.returncode == 0 and any(f":{port}" in line for line in proc.stdout.splitlines())
def check_realtime(config):
issues = []
rt = config["realtime"]
for name, info in docker_inspect(rt["containers"]).items():
if info.get("missing"):
issues.append(f"[realtime] {name}: missing")
elif not info["running"]:
issues.append(f"[realtime] {name}: not running, status={info['status']}")
for port in rt["tcp_ports"]:
if not tcp_open(port):
issues.append(f"[realtime] tcp/{port}: not listening")
for port in rt["udp_ports"]:
if not udp_listening(port):
issues.append(f"[realtime] udp/{port}: not listening")
for probe in rt["http_probes"]:
status, payload = request("GET", probe["url"], timeout=5)
if status not in probe["ok"]:
issues.append(f"[realtime] {probe['name']}: HTTP {status} {payload}")
return issues
def send_room(config, token, room_id, body):
txn = "server-monitor-" + dt.datetime.now(dt.timezone.utc).strftime("%Y%m%dT%H%M%S")
txn += "-" + hashlib.sha1((room_id + body).encode("utf-8")).hexdigest()[:10]
html_body = "<br>".join(html.escape(line) for line in body.splitlines())
content = {
"msgtype": "m.text",
"body": body,
"format": "org.matrix.custom.html",
"formatted_body": html_body,
}
url = f"{config['homeserver']}/_matrix/client/v3/rooms/{quote(room_id)}/send/m.room.message/{txn}"
status, payload = request("PUT", url, token, content)
if status not in (200, 201):
raise RuntimeError(f"send failed: HTTP {status} {payload}")
def send_alert(config, token, state, body):
rooms = set(config.get("target_rooms", []))
weather_state = load_json(config.get("bot_weather_state_path", ""), {})
for user_id, room_id in weather_state.get("direct_rooms", {}).items():
if user_id in config["target_users"]:
rooms.add(room_id)
for room_id in sorted(rooms):
send_room(config, token, room_id, body)
time.sleep(0.2)
def main():
config = load_json(CONFIG_PATH, None)
if not config:
raise SystemExit(f"missing config: {CONFIG_PATH}")
state = load_json(config["state_path"], {})
with open(config["token_path"], "r", encoding="utf-8") as f:
token = f.read().strip()
issues = sorted(set(check_disks(config) + check_docker(config, state) + check_realtime(config)))
previous = set(state.get("active_issues", []))
current = set(issues)
new_issues = sorted(current - previous)
recovered = sorted(previous - current)
state["active_issues"] = issues
state["last_check"] = dt.datetime.now().isoformat(timespec="seconds")
if new_issues or recovered:
lines = [f"服务器监控提醒 {dt.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", ""]
if new_issues:
lines.append("新增异常：")
lines.extend(f"- {x}" for x in new_issues)
if recovered:
if new_issues:
lines.append("")
lines.append("已恢复：")
lines.extend(f"- {x}" for x in recovered)
send_alert(config, token, state, "\n".join(lines))
save_json(config["state_path"], state)
print(f"monitor ok: active={len(issues)} new={len(new_issues)} recovered={len(recovered)}")
if __name__ == "__main__":
main()设置权限并手动测试：
chmod 700 /root/data/docker_data/synapse/scripts/matrix-bot-monitor.py
python3 -m py_compile /root/data/docker_data/synapse/scripts/matrix-bot-monitor.py
/root/data/docker_data/synapse/scripts/matrix-bot-monitor.py正常时输出类似：
monitor ok: active=0 new=0 recovered=0如果第一次运行时已有异常，会立刻给 Matrix 目标房间发一条“新增异常”。后续同一异常不会反复刷屏；只有出现新异常或旧异常恢复时，才会再次提醒。
九、配置每 5 分钟监控一次
创建包装脚本：
vi /root/data/docker_data/synapse/scripts/run-matrix-bot-monitor.sh写入：
#!/bin/sh
LOCK=/tmp/matrix-bot-monitor.lock
if mkdir "$LOCK" 2>/dev/null; then
trap 'rmdir "$LOCK"' EXIT
/root/data/docker_data/synapse/scripts/matrix-bot-monitor.py >> /var/log/matrix-bot-monitor.log 2>&1
fi设置权限：
chmod 700 /root/data/docker_data/synapse/scripts/run-matrix-bot-monitor.sh编辑 root crontab：
vi /etc/crontabs/root加入：
*/5 * * * * /root/data/docker_data/synapse/scripts/run-matrix-bot-monitor.sh重启 crond：
rc-service crond restart手动跑一次包装脚本，确认日志正常：
/root/data/docker_data/synapse/scripts/run-matrix-bot-monitor.sh
tail -n 20 /var/log/matrix-bot-monitor.log十、聊天命令和用户自助设置
普通用户可以主动邀请机器人：
@bot:666666.xyz机器人只自动接受本服用户邀请，也就是 :666666.xyz 结尾的用户。首次加入后会自动发送帮助说明：
你好，我是天气机器人。
你可以这样使用我：
天气
立即查看今天和未来两天天气
设置天气时间 07:30
设置每天自动推送时间
天气时间
查看当前推送时间
关闭天气
停止每日天气推送
开启天气
恢复每日天气推送
帮助
再次查看这份说明普通用户只会获得天气功能，不会收到服务器监控提醒。监控提醒由 matrix-bot-monitor.json 的 target_users 单独控制，例如只发给管理员：
"target_users": [
"@black:jgaga.tk",
"@honey:jgaga.tk",
"@admin2:jgaga.tk"
]改完白名单后不要只看配置文件，最好发一次无损测试告警，确认每个目标用户都能建立私聊并收到提醒：
/root/data/docker_data/synapse/scripts/matrix-bot-monitor.py --test-alert
python3 -m json.tool /root/data/docker_data/synapse/scripts/matrix-bot-monitor.state.json天气用户则保存在 matrix-bot-weather.json 的 weather_users 里，每个用户可以有自己的推送时间：
"weather_users": {
"@honey:666666.xyz": {
"enabled": true,
"time": "05:00"
},
"@black:666666.xyz": {
"enabled": true,
"time": "07:30"
}
}如果只是给固定管理员开通默认天气提醒，当前 jgaga.tk 这套生产栈默认保留：
"target_users": [
"@honey:jgaga.tk",
"@black:jgaga.tk",
"@admin2:jgaga.tk"
]target_users 只用于初始化默认用户；真实的开关、时间和城市仍以 weather_users 为准。不要把定时发送重新改回只遍历 target_users，否则用户自己发送“设置天气时间”和“设置城市”后，后台推送不会按个人配置生效。
十一、手动维护用户配置
一般不需要手动维护，用户直接邀请 @bot:666666.xyz 后会自动加入 weather_users。如果需要管理员直接改配置，可以编辑：
vi /root/data/docker_data/synapse/scripts/matrix-bot-weather.json示例：
"weather_users": {
"@honey:666666.xyz": {
"enabled": true,
"time": "05:00"
},
"@other:666666.xyz": {
"enabled": true,
"time": "07:30"
}
}保存后可以手动测试立即发送：
/root/data/docker_data/synapse/scripts/matrix-bot-weather.py --send-now十二、关于加密房间
这个脚本使用的是普通 Matrix Client-Server API，没有实现端到端加密设备管理，也没有处理 Olm/Megolm 密钥。
因此建议让机器人使用未加密私聊或未加密房间。上面的脚本没有实现 E2EE，机器人即使能加入加密房间，也不能像正常客户端那样解密和处理消息。
如果旧版本已经把某个用户的天气私聊建成了加密房间，不要尝试原地关闭加密。Matrix 加密状态不能撤回，正确做法是让脚本检测 m.room.encryption，离开旧房间并重新创建未加密私聊，再更新 direct_rooms 映射。
如果一定要在端到端加密房间里使用机器人，需要换成支持 E2EE 的机器人框架，例如 matrix-nio[e2e]，并额外处理设备信任、密钥存储和恢复。
十三、排查命令
查看日志：
tail -n 100 /var/log/matrix-bot-weather.log
tail -n 100 /var/log/matrix-bot-sync.log
tail -n 100 /var/log/matrix-bot-monitor.log检查脚本语法：
python3 -m py_compile /root/data/docker_data/synapse/scripts/matrix-bot-weather.py
python3 -m py_compile /root/data/docker_data/synapse/scripts/matrix-bot-sync.py
python3 -m py_compile /root/data/docker_data/synapse/scripts/matrix-bot-monitor.py确认机器人 token 文件存在：
ls -l /root/.config/matrix-bot/access_token确认 Synapse API 正常：
curl -sS -o /tmp/synapse_versions.json -w 'http_code=%{http_code}\n' \
http://127.0.0.1:8008/_matrix/client/versions
rm -f /tmp/synapse_versions.json确认天气和聊天同步服务正常运行：
rc-service matrix-bot-weather status
rc-service matrix-bot-sync status
ps -ef | grep matrix-bot-weather | grep -v grep
ps -ef | grep matrix-bot-sync | grep -v grep确认监控 cron 正常运行：
rc-service crond status
grep -n 'matrix-bot-monitor' /etc/crontabs/root如果要立即发一次天气测试，可以直接运行：
/root/data/docker_data/synapse/scripts/matrix-bot-weather.py --send-now如果要测试“定时是否真的生效”，不要只跑 --send-now。应该在 Matrix 客户端里把天气时间设置为下一分钟，然后观察 /var/log/matrix-bot-weather.log，确认是 matrix-bot-weather 服务自己到点发送。
天气脚本按每个用户配置的时间判断是否到点，并在 state 文件里按“日期 + 时间”记录已经发送，避免重复推送。聊天命令和天气推送是两条后台循环服务，监控提醒仍然是独立 cron；普通用户改天气时间不会影响监控提醒。