From 19afaffbb789c965d30006c7afac08d71d4f745e Mon Sep 17 00:00:00 2001
From: Lee <1633292@qq.com>
Date: Mon, 7 Sep 2026 17:50:25 +0800
Subject: [PATCH] feat: add shared notification gateway
---
services/notify-gateway/.env.example | 6 +
services/notify-gateway/.gitignore | 10 +
services/notify-gateway/Dockerfile | 10 +
services/notify-gateway/README.md | 46 +++++
services/notify-gateway/compose.yaml | 38 ++++
services/notify-gateway/config.example.yaml | 24 +++
services/notify-gateway/gateway/__init__.py | 0
services/notify-gateway/gateway/__main__.py | 4 +
services/notify-gateway/gateway/app.py | 180 ++++++++++++++++++
.../notify-gateway/gateway/delivery_store.py | 48 +++++
services/notify-gateway/notifiers/__init__.py | 3 +
services/notify-gateway/notifiers/base.py | 18 ++
services/notify-gateway/notifiers/client.py | 36 ++++
services/notify-gateway/notifiers/factory.py | 51 +++++
.../notify-gateway/notifiers/image_utils.py | 52 +++++
services/notify-gateway/notifiers/qq.py | 37 ++++
services/notify-gateway/notifiers/telegram.py | 72 +++++++
.../notify-gateway/notifiers/wecom_webhook.py | 126 ++++++++++++
services/notify-gateway/requirements.txt | 5 +
.../notify-gateway/tests/test_plain_format.py | 32 ++++
20 files changed, 798 insertions(+)
create mode 100644 services/notify-gateway/.env.example
create mode 100644 services/notify-gateway/.gitignore
create mode 100644 services/notify-gateway/Dockerfile
create mode 100644 services/notify-gateway/README.md
create mode 100644 services/notify-gateway/compose.yaml
create mode 100644 services/notify-gateway/config.example.yaml
create mode 100644 services/notify-gateway/gateway/__init__.py
create mode 100644 services/notify-gateway/gateway/__main__.py
create mode 100644 services/notify-gateway/gateway/app.py
create mode 100644 services/notify-gateway/gateway/delivery_store.py
create mode 100644 services/notify-gateway/notifiers/__init__.py
create mode 100644 services/notify-gateway/notifiers/base.py
create mode 100644 services/notify-gateway/notifiers/client.py
create mode 100644 services/notify-gateway/notifiers/factory.py
create mode 100644 services/notify-gateway/notifiers/image_utils.py
create mode 100644 services/notify-gateway/notifiers/qq.py
create mode 100644 services/notify-gateway/notifiers/telegram.py
create mode 100644 services/notify-gateway/notifiers/wecom_webhook.py
create mode 100644 services/notify-gateway/requirements.txt
create mode 100644 services/notify-gateway/tests/test_plain_format.py
diff --git a/services/notify-gateway/.env.example b/services/notify-gateway/.env.example
new file mode 100644
index 0000000..b784db2
--- /dev/null
+++ b/services/notify-gateway/.env.example
@@ -0,0 +1,6 @@
+TZ=Asia/Singapore
+NOTIFY_GATEWAY_TOKEN=replace-with-random-token
+TELEGRAM_BOT_TOKEN=replace-with-bot-token
+TELEGRAM_NOTIFY_CHAT_ID=replace-with-chat-id
+WECOM_WEBHOOK_URLS=https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=replace-with-webhook-key
+SHARED_NETWORK_NAME=shared-services
diff --git a/services/notify-gateway/.gitignore b/services/notify-gateway/.gitignore
new file mode 100644
index 0000000..a2ff316
--- /dev/null
+++ b/services/notify-gateway/.gitignore
@@ -0,0 +1,10 @@
+.env
+config/config.yaml
+data/
+logs/
+secrets/
+backups/
+__pycache__/
+*.py[cod]
+*.db
+*.db-*
diff --git a/services/notify-gateway/Dockerfile b/services/notify-gateway/Dockerfile
new file mode 100644
index 0000000..3eb720d
--- /dev/null
+++ b/services/notify-gateway/Dockerfile
@@ -0,0 +1,10 @@
+FROM python:3.12-slim
+WORKDIR /app
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+COPY gateway ./gateway
+COPY notifiers ./notifiers
+COPY config.example.yaml ./config.example.yaml
+RUN chmod -R a+rX /app/gateway /app/notifiers /app/config.example.yaml
+ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1
+CMD ["python", "-m", "gateway"]
diff --git a/services/notify-gateway/README.md b/services/notify-gateway/README.md
new file mode 100644
index 0000000..537701b
--- /dev/null
+++ b/services/notify-gateway/README.md
@@ -0,0 +1,46 @@
+# notify-gateway
+
+独立、可复用的 HTTP 通知网关。调用方必须显式指定渠道；Telegram 还必须指定命名 BOT。
+当前支持 Telegram、企业微信群 Webhook，并保留默认关闭的 OneBot QQ 适配器。
+
+## API
+
+```http
+POST /send
+Authorization: Bearer $NOTIFY_GATEWAY_TOKEN
+Content-Type: application/json
+```
+
+```json
+{
+ "title": "通知标题",
+ "text": "正文",
+ "channels": ["telegram", "wecom_webhook"],
+ "telegram_bot": "default",
+ "idempotency_key": "caller:event:123"
+}
+```
+
+可选 `images` 接受最多 4 张 JPEG/PNG Base64 图片，原始数据合计不超过 50 MB：
+
+```json
+{"images":[{"filename":"photo.jpg","content_type":"image/jpeg","data_base64":"..."}]}
+```
+
+- Telegram：尺寸允许时直接以照片显示，过大时只发送压缩照片预览。
+- 企业微信：尺寸允许时直接显示，过大时只发送不超过平台限制的压缩预览。
+- 网关不再把原图作为文件附件发送，也不接受远程图片 URL。
+- 空标题表示原样纯文本模式，不自动添加 emoji 或分割线。
+- `channels` 必须显式指定且不能为空。
+- 请求 Telegram 时必须提供配置中存在的 `telegram_bot`。
+- 建议始终提供稳定的 `idempotency_key`，成功渠道在重试时会跳过。
+- `GET /health` 无需鉴权。
+
+## 启动
+
+1. 复制 `.env.example` 为 `.env` 并填写所需渠道变量。
+2. 复制 `config.example.yaml` 为 `config/config.yaml`，按需启用渠道和命名 BOT。
+3. 创建共享网络：`docker network create shared-services`，或设置 `SHARED_NETWORK_NAME`。
+4. 运行 `docker compose up -d --build`。
+
+Python 调用客户端见 `notifiers/client.py`。
diff --git a/services/notify-gateway/compose.yaml b/services/notify-gateway/compose.yaml
new file mode 100644
index 0000000..ecffccc
--- /dev/null
+++ b/services/notify-gateway/compose.yaml
@@ -0,0 +1,38 @@
+services:
+ notify-gateway:
+ build: .
+ image: notify-gateway:local
+ container_name: notify-gateway
+ restart: unless-stopped
+ environment:
+ NOTIFY_GATEWAY_CONFIG: /app/config/config.yaml
+ NOTIFY_GATEWAY_TOKEN: ${NOTIFY_GATEWAY_TOKEN:?set NOTIFY_GATEWAY_TOKEN in .env}
+ TELEGRAM_BOT_TOKEN: ${TELEGRAM_BOT_TOKEN:-}
+ TELEGRAM_NOTIFY_CHAT_ID: ${TELEGRAM_NOTIFY_CHAT_ID:-}
+ WECOM_WEBHOOK_URLS: ${WECOM_WEBHOOK_URLS:-}
+ TZ: ${TZ:-Asia/Singapore}
+ volumes:
+ - ./config:/app/config:ro
+ - ./data:/app/data
+ - ./logs:/app/logs
+ ports:
+ - "127.0.0.1:8770:8770"
+ networks:
+ - class-monitor
+ - shared-services
+ read_only: true
+ tmpfs:
+ - /tmp:size=16m,mode=1777
+ user: "10000:10000"
+ security_opt:
+ - no-new-privileges:true
+ cap_drop:
+ - ALL
+
+networks:
+ class-monitor:
+ external: true
+ name: class-monitor
+ shared-services:
+ external: true
+ name: ${SHARED_NETWORK_NAME:-shared-services}
diff --git a/services/notify-gateway/config.example.yaml b/services/notify-gateway/config.example.yaml
new file mode 100644
index 0000000..7c0bb2b
--- /dev/null
+++ b/services/notify-gateway/config.example.yaml
@@ -0,0 +1,24 @@
+timezone: Asia/Singapore
+listen_host: 0.0.0.0
+listen_port: 8770
+state_db: /app/data/deliveries.db
+api_token: ""
+channels:
+ telegram:
+ enabled: true
+ bots:
+ default:
+ token_env: TELEGRAM_BOT_TOKEN
+ chat_ids_env: TELEGRAM_NOTIFY_CHAT_ID
+ # lottery:
+ # token_env: TELEGRAM_LOTTERY_BOT_TOKEN
+ # chat_ids_env: TELEGRAM_LOTTERY_CHAT_ID
+ wecom_webhook:
+ enabled: false
+ # WECOM_WEBHOOK_URLS: semicolon-separated, one URL per internal group.
+ urls: []
+ qq:
+ enabled: false
+ onebot_http: ""
+ access_token: ""
+ targets: []
diff --git a/services/notify-gateway/gateway/__init__.py b/services/notify-gateway/gateway/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/services/notify-gateway/gateway/__main__.py b/services/notify-gateway/gateway/__main__.py
new file mode 100644
index 0000000..37c2a2d
--- /dev/null
+++ b/services/notify-gateway/gateway/__main__.py
@@ -0,0 +1,4 @@
+from .app import main
+
+if __name__ == "__main__":
+ main()
diff --git a/services/notify-gateway/gateway/app.py b/services/notify-gateway/gateway/app.py
new file mode 100644
index 0000000..62c7bf5
--- /dev/null
+++ b/services/notify-gateway/gateway/app.py
@@ -0,0 +1,180 @@
+import asyncio
+import base64
+import hashlib
+import hmac
+import logging
+import os
+from pathlib import Path
+
+import yaml
+from aiohttp import web
+
+from notifiers.base import ImageAttachment
+from notifiers.factory import build_notifier
+from .delivery_store import DeliveryStore
+
+LOG = logging.getLogger("notify-gateway")
+MAX_IMAGES = 4
+MAX_IMAGE_BYTES = 50_000_000
+MAX_TOTAL_IMAGE_BYTES = 50_000_000
+
+
+def parse_images(raw: object) -> list[ImageAttachment]:
+ if raw in (None, []):
+ return []
+ if not isinstance(raw, list) or len(raw) > MAX_IMAGES:
+ raise web.HTTPBadRequest(reason="images must be a list with at most 4 items")
+ images: list[ImageAttachment] = []
+ total_bytes = 0
+ for item in raw:
+ if not isinstance(item, dict):
+ raise web.HTTPBadRequest(reason="invalid image item")
+ content_type = str(item.get("content_type") or "")
+ if content_type not in {"image/jpeg", "image/png"}:
+ raise web.HTTPBadRequest(reason="unsupported image content type")
+ try:
+ data = base64.b64decode(str(item.get("data_base64") or ""), validate=True)
+ except Exception:
+ raise web.HTTPBadRequest(reason="invalid image base64")
+ magic_ok = (
+ content_type == "image/jpeg" and data.startswith(bytes.fromhex("ffd8ff"))
+ ) or (
+ content_type == "image/png"
+ and data.startswith(bytes.fromhex("89504e470d0a1a0a"))
+ )
+ if not magic_ok or not data or len(data) > MAX_IMAGE_BYTES:
+ raise web.HTTPBadRequest(reason="invalid image data or size")
+ total_bytes += len(data)
+ if total_bytes > MAX_TOTAL_IMAGE_BYTES:
+ raise web.HTTPBadRequest(reason="total image size exceeds 50 MB")
+ filename = Path(str(item.get("filename") or "image")).name
+ images.append(ImageAttachment(filename, content_type, data))
+ return images
+
+
+class Gateway:
+ def __init__(self, cfg: dict):
+ self.cfg = cfg
+ self.expected = os.getenv("NOTIFY_GATEWAY_TOKEN", "") or str(cfg.get("api_token", ""))
+ self.channels_cfg = cfg.get("channels") or {}
+ self.notifiers = {}
+ self.telegram_bots = {}
+ for name, channel_cfg in self.channels_cfg.items():
+ if not channel_cfg.get("enabled"):
+ continue
+ if name == "telegram":
+ bots_cfg = channel_cfg.get("bots") or {}
+ if not isinstance(bots_cfg, dict) or not bots_cfg:
+ raise RuntimeError("Telegram bots are not configured")
+ self.telegram_bots = {
+ bot_name: build_notifier("telegram", bot_cfg)
+ for bot_name, bot_cfg in bots_cfg.items()
+ }
+ else:
+ self.notifiers[name] = build_notifier(name, channel_cfg)
+ self.store = DeliveryStore(str(cfg.get("state_db", "/app/data/deliveries.db")))
+
+ def _check(self, request: web.Request) -> None:
+ if not self.expected:
+ raise web.HTTPForbidden(reason="gateway token not configured")
+ auth = request.headers.get("Authorization", "")
+ if not hmac.compare_digest(auth, f"Bearer {self.expected}"):
+ raise web.HTTPUnauthorized()
+
+ async def health(self, request: web.Request):
+ channels = {
+ name: (bool(self.telegram_bots) if name == "telegram" else name in self.notifiers)
+ for name in self.channels_cfg
+ }
+ return web.json_response({
+ "ok": True,
+ "channels": channels,
+ "telegram_bots": sorted(self.telegram_bots),
+ })
+
+ async def send(self, request: web.Request):
+ self._check(request)
+ try:
+ data = await request.json()
+ except Exception:
+ raise web.HTTPBadRequest(reason="invalid json")
+ title = str(data.get("title", ""))
+ text = str(data.get("text", ""))
+ images = parse_images(data.get("images"))
+ wanted = data.get("channels")
+ if not text and not title and not images:
+ raise web.HTTPBadRequest(reason="empty message")
+ if not isinstance(wanted, list) or not wanted or not all(isinstance(x, str) and x for x in wanted):
+ raise web.HTTPBadRequest(reason="channels must explicitly select at least one channel")
+ if len(set(wanted)) != len(wanted):
+ raise web.HTTPBadRequest(reason="channels must not contain duplicates")
+ if len(title) > 500 or len(text) > 20000:
+ raise web.HTTPRequestEntityTooLarge(max_size=20500, actual_size=len(title) + len(text))
+ telegram_bot = data.get("telegram_bot")
+ if "telegram" in wanted:
+ if not isinstance(telegram_bot, str) or not telegram_bot.strip():
+ raise web.HTTPBadRequest(reason="telegram_bot is required for Telegram delivery")
+ telegram_bot = telegram_bot.strip()
+ if telegram_bot not in self.telegram_bots:
+ raise web.HTTPBadRequest(reason="unknown telegram_bot")
+ elif telegram_bot is not None:
+ raise web.HTTPBadRequest(reason="telegram_bot is only valid with the Telegram channel")
+ key = str(data.get("idempotency_key") or "").strip()
+ if not key:
+ digest_input = bytearray((title + "\0" + text).encode("utf-8"))
+ for image in images:
+ digest_input.extend(b"\0")
+ digest_input.extend(hashlib.sha256(image.data).digest())
+ digest = hashlib.sha256(digest_input).hexdigest()
+ key = f"content:{digest}"
+ if len(key) > 200:
+ raise web.HTTPBadRequest(reason="idempotency_key too long")
+ results: dict[str, str] = {}
+ for name in wanted:
+ notifier = self.telegram_bots.get(telegram_bot) if name == "telegram" else self.notifiers.get(name)
+ if not notifier:
+ results[name] = "disabled"
+ continue
+ delivery_target = f"telegram:{telegram_bot}" if name == "telegram" else name
+ if await self.store.delivered(key, delivery_target):
+ results[name] = "already_delivered"
+ continue
+ try:
+ await notifier.send(title, text, images)
+ await self.store.record(key, delivery_target, True)
+ results[name] = "ok"
+ except Exception as exc:
+ LOG.exception("gateway send failed channel=%s", name)
+ await self.store.record(key, delivery_target, False, repr(exc))
+ results[name] = "error"
+ ok = bool(results) and all(v in {"ok", "already_delivered"} for v in results.values())
+ return web.json_response({"ok": ok, "results": results})
+
+ async def run_app(self):
+ await self.store.init()
+ site = web.Application(client_max_size=70 * 1024 * 1024)
+ site.router.add_get("/health", self.health)
+ site.router.add_post("/send", self.send)
+ runner = web.AppRunner(site)
+ await runner.setup()
+ await web.TCPSite(runner, self.cfg.get("listen_host", "0.0.0.0"),
+ int(self.cfg.get("listen_port", 8770))).start()
+ LOG.info("notify-gateway ready channels=%s",
+ [k for k, v in (self.cfg.get("channels") or {}).items() if v.get("enabled")])
+ await asyncio.Event().wait()
+
+
+def load_config():
+ with open(os.environ.get("NOTIFY_GATEWAY_CONFIG", "/app/config/config.yaml"),
+ encoding="utf-8") as f:
+ return yaml.safe_load(f)
+
+
+def main():
+ Path("/app/logs").mkdir(parents=True, exist_ok=True)
+ logging.basicConfig(level=logging.INFO,
+ format="%(asctime)s %(levelname)s %(name)s %(message)s")
+ # httpx logs full request URLs; Telegram and WeCom put credentials in URLs.
+ logging.getLogger("httpx").setLevel(logging.WARNING)
+ logging.getLogger("httpcore").setLevel(logging.WARNING)
+ asyncio.run(Gateway(load_config()).run_app())
diff --git a/services/notify-gateway/gateway/delivery_store.py b/services/notify-gateway/gateway/delivery_store.py
new file mode 100644
index 0000000..6e31685
--- /dev/null
+++ b/services/notify-gateway/gateway/delivery_store.py
@@ -0,0 +1,48 @@
+import aiosqlite
+
+
+SCHEMA = """
+PRAGMA journal_mode=WAL;
+PRAGMA synchronous=FULL;
+CREATE TABLE IF NOT EXISTS deliveries (
+ idempotency_key TEXT NOT NULL,
+ channel TEXT NOT NULL,
+ delivered INTEGER NOT NULL DEFAULT 0,
+ attempts INTEGER NOT NULL DEFAULT 0,
+ last_error TEXT,
+ updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+ PRIMARY KEY (idempotency_key, channel)
+);
+"""
+
+
+class DeliveryStore:
+ def __init__(self, path: str) -> None:
+ self.path = path
+
+ async def init(self) -> None:
+ async with aiosqlite.connect(self.path) as db:
+ await db.executescript(SCHEMA)
+ await db.commit()
+
+ async def delivered(self, key: str, channel: str) -> bool:
+ async with aiosqlite.connect(self.path) as db:
+ row = await (await db.execute(
+ "SELECT delivered FROM deliveries WHERE idempotency_key=? AND channel=?",
+ (key, channel),
+ )).fetchone()
+ return bool(row and row[0])
+
+ async def record(self, key: str, channel: str, ok: bool, error: str = "") -> None:
+ async with aiosqlite.connect(self.path) as db:
+ await db.execute(
+ """INSERT INTO deliveries(idempotency_key,channel,delivered,attempts,last_error)
+ VALUES(?,?,?,?,?)
+ ON CONFLICT(idempotency_key,channel) DO UPDATE SET
+ delivered=MAX(deliveries.delivered, excluded.delivered),
+ attempts=deliveries.attempts+1,
+ last_error=CASE WHEN excluded.delivered=1 THEN NULL ELSE excluded.last_error END,
+ updated_at=CURRENT_TIMESTAMP""",
+ (key, channel, int(ok), 1, error[:500] or None),
+ )
+ await db.commit()
diff --git a/services/notify-gateway/notifiers/__init__.py b/services/notify-gateway/notifiers/__init__.py
new file mode 100644
index 0000000..a92e761
--- /dev/null
+++ b/services/notify-gateway/notifiers/__init__.py
@@ -0,0 +1,3 @@
+from .factory import build_notifier, build_notifiers
+
+__all__ = ["build_notifier", "build_notifiers"]
diff --git a/services/notify-gateway/notifiers/base.py b/services/notify-gateway/notifiers/base.py
new file mode 100644
index 0000000..37f8f72
--- /dev/null
+++ b/services/notify-gateway/notifiers/base.py
@@ -0,0 +1,18 @@
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class ImageAttachment:
+ filename: str
+ content_type: str
+ data: bytes
+
+
+class Notifier(ABC):
+ name = "base"
+
+ @abstractmethod
+ async def send(self, title: str, text: str,
+ images: list[ImageAttachment] | None = None) -> None:
+ """Deliver one notification or raise so the caller can retry."""
diff --git a/services/notify-gateway/notifiers/client.py b/services/notify-gateway/notifiers/client.py
new file mode 100644
index 0000000..deb52b5
--- /dev/null
+++ b/services/notify-gateway/notifiers/client.py
@@ -0,0 +1,36 @@
+"""HTTP client for external programs calling notify-gateway."""
+import httpx
+
+
+class NotifyClient:
+ def __init__(self, base_url: str, token: str, timeout: float = 20.0) -> None:
+ self.base_url = base_url.rstrip("/")
+ self.token = token
+ self.timeout = timeout
+
+ async def send(self, title: str, text: str, channels: list[str],
+ telegram_bot: str = "",
+ idempotency_key: str = "",
+ images: list[dict] | None = None) -> dict:
+ if not channels:
+ raise ValueError("channels must explicitly select at least one channel")
+ if "telegram" in channels and not telegram_bot.strip():
+ raise ValueError("telegram_bot is required for Telegram delivery")
+ payload = {"title": title, "text": text, "channels": channels}
+ if "telegram" in channels:
+ payload["telegram_bot"] = telegram_bot.strip()
+ if images:
+ payload["images"] = images
+ if idempotency_key:
+ payload["idempotency_key"] = idempotency_key
+ async with httpx.AsyncClient(timeout=self.timeout, trust_env=False) as client:
+ r = await client.post(
+ f"{self.base_url}/send",
+ headers={"Authorization": f"Bearer {self.token}"},
+ json=payload,
+ )
+ r.raise_for_status()
+ data = r.json()
+ if not data.get("ok"):
+ raise RuntimeError(f"notify-gateway delivery incomplete: {data.get('results', {})}")
+ return data
diff --git a/services/notify-gateway/notifiers/factory.py b/services/notify-gateway/notifiers/factory.py
new file mode 100644
index 0000000..b6514d9
--- /dev/null
+++ b/services/notify-gateway/notifiers/factory.py
@@ -0,0 +1,51 @@
+from .base import ImageAttachment, Notifier
+from .qq import QQNotifier
+from .telegram import TelegramNotifier
+from .wecom_webhook import WecomWebhookNotifier
+
+
+def build_notifier(name: str, cfg: dict | None = None) -> Notifier:
+ cfg = cfg or {}
+ if name == "telegram":
+ return TelegramNotifier(
+ token=cfg.get("token", ""),
+ chat_ids=cfg.get("chat_ids"),
+ token_env=cfg.get("token_env", "TELEGRAM_BOT_TOKEN"),
+ chat_ids_env=cfg.get("chat_ids_env", "TELEGRAM_NOTIFY_CHAT_ID"),
+ )
+ if name == "wecom_webhook":
+ return WecomWebhookNotifier(urls=cfg.get("urls"))
+ if name == "qq":
+ return QQNotifier(
+ onebot_http=cfg.get("onebot_http", ""),
+ access_token=cfg.get("access_token", ""),
+ targets=cfg.get("targets", []))
+ raise ValueError(f"Unknown notification channel: {name}")
+
+
+def build_notifiers(names: list[str], channels_cfg: dict | None = None) -> "MultiNotifier":
+ channels_cfg = channels_cfg or {}
+ return MultiNotifier([build_notifier(n, channels_cfg.get(n, {})) for n in names])
+
+
+class MultiNotifier(Notifier):
+ """Fan-out to N channels; raises only if ALL fail."""
+
+ name = "multi"
+
+ def __init__(self, notifiers: list[Notifier]):
+ self.notifiers = notifiers
+
+ async def send(self, title: str, text: str,
+ images: list[ImageAttachment] | None = None) -> None:
+ errors: dict[str, str] = {}
+ for n in self.notifiers:
+ try:
+ await n.send(title, text, images)
+ except Exception as exc:
+ errors[n.name] = repr(exc)
+ if errors and len(errors) == len(self.notifiers):
+ raise RuntimeError(f"All notifiers failed: {errors}")
+ if errors:
+ import logging
+ logging.getLogger("notify-gateway").warning("partial notify failure: %s", errors)
diff --git a/services/notify-gateway/notifiers/image_utils.py b/services/notify-gateway/notifiers/image_utils.py
new file mode 100644
index 0000000..696e493
--- /dev/null
+++ b/services/notify-gateway/notifiers/image_utils.py
@@ -0,0 +1,52 @@
+import io
+from pathlib import Path
+
+from PIL import Image, ImageOps
+
+from .base import ImageAttachment
+
+
+INLINE_IMAGE_MAX_BYTES = 2_000_000
+PREVIEW_TARGET_BYTES = 1_900_000
+WECOM_FILE_MAX_BYTES = 20_000_000
+TELEGRAM_PHOTO_MAX_BYTES = 10_000_000
+TELEGRAM_FILE_MAX_BYTES = 50_000_000
+
+
+def compressed_preview(image: ImageAttachment) -> ImageAttachment:
+ if len(image.data) <= INLINE_IMAGE_MAX_BYTES:
+ return image
+
+ with Image.open(io.BytesIO(image.data)) as source:
+ source.load()
+ working = ImageOps.exif_transpose(source)
+ if working.mode in {"RGBA", "LA"}:
+ background = Image.new("RGB", working.size, "white")
+ alpha = working.getchannel("A")
+ background.paste(working.convert("RGB"), mask=alpha)
+ working = background
+ elif working.mode != "RGB":
+ working = working.convert("RGB")
+
+ current = working
+ while True:
+ for quality in (90, 82, 74, 66, 58, 50, 42, 34):
+ output = io.BytesIO()
+ current.save(
+ output,
+ format="JPEG",
+ quality=quality,
+ optimize=True,
+ progressive=True,
+ )
+ data = output.getvalue()
+ if len(data) <= PREVIEW_TARGET_BYTES:
+ name = f"{Path(image.filename).stem}-preview.jpg"
+ return ImageAttachment(name, "image/jpeg", data)
+ width, height = current.size
+ if min(width, height) <= 128:
+ raise RuntimeError("image cannot be compressed below preview limit")
+ current = current.resize(
+ (max(1, int(width * 0.8)), max(1, int(height * 0.8))),
+ Image.Resampling.LANCZOS,
+ )
diff --git a/services/notify-gateway/notifiers/qq.py b/services/notify-gateway/notifiers/qq.py
new file mode 100644
index 0000000..cd2f04c
--- /dev/null
+++ b/services/notify-gateway/notifiers/qq.py
@@ -0,0 +1,37 @@
+"""QQ notifier (reserved). Transport: NapCat OneBot HTTP API."""
+import os
+
+import httpx
+
+from .base import ImageAttachment, Notifier
+
+
+class QQNotifier(Notifier):
+ name = "qq"
+
+ def __init__(self, onebot_http: str = '', access_token: str = '',
+ targets: list[dict] | None = None) -> None:
+ self.onebot_http = (onebot_http or os.getenv("QQ_ONEBOT_HTTP", "")).rstrip("/")
+ self.access_token = access_token or os.getenv("ONEBOT_ACCESS_TOKEN", "")
+ self.targets = targets if targets is not None else []
+ if not self.onebot_http:
+ raise RuntimeError("QQ OneBot HTTP endpoint not configured (reserved)")
+
+ async def send(self, title: str, text: str,
+ images: list[ImageAttachment] | None = None) -> None:
+ if not self.targets:
+ raise RuntimeError("QQ targets not configured (reserved)")
+ headers = {}
+ if self.access_token:
+ headers["Authorization"] = f"Bearer {self.access_token}"
+ body = f"{title}\n{text}" if title else text
+ async with httpx.AsyncClient(timeout=20, trust_env=False) as client:
+ for t in self.targets:
+ if t.get("group_id"):
+ await client.post(f"{self.onebot_http}/send_group_msg",
+ headers=headers,
+ json={"group_id": t["group_id"], "message": body})
+ elif t.get("user_id"):
+ await client.post(f"{self.onebot_http}/send_private_msg",
+ headers=headers,
+ json={"user_id": t["user_id"], "message": body})
diff --git a/services/notify-gateway/notifiers/telegram.py b/services/notify-gateway/notifiers/telegram.py
new file mode 100644
index 0000000..16ff2c6
--- /dev/null
+++ b/services/notify-gateway/notifiers/telegram.py
@@ -0,0 +1,72 @@
+import html
+import os
+
+import httpx
+
+from .base import ImageAttachment, Notifier
+from .image_utils import (
+ TELEGRAM_PHOTO_MAX_BYTES,
+ compressed_preview,
+)
+
+
+def esc(value: object) -> str:
+ return html.escape(str(value or ""))
+
+
+class TelegramNotifier(Notifier):
+ name = "telegram"
+
+ def __init__(self, token: str = '', chat_ids: list[str] | None = None,
+ token_env: str = "TELEGRAM_BOT_TOKEN",
+ chat_ids_env: str = "TELEGRAM_NOTIFY_CHAT_ID") -> None:
+ self.token = token or os.environ.get(token_env, "")
+ raw = ""
+ if chat_ids:
+ raw = ",".join(chat_ids)
+ else:
+ raw = os.getenv(chat_ids_env) or (
+ os.getenv("TELEGRAM_HOME_CHANNEL") if chat_ids_env == "TELEGRAM_NOTIFY_CHAT_ID" else ""
+ ) or ""
+ # FIX vs old code: fan-out to ALL chat ids, not just split(',',1)[0].
+ self.chat_ids = [c.strip() for c in raw.split(",") if c.strip()]
+ if not self.token:
+ raise RuntimeError("Telegram token is not configured")
+ if not self.chat_ids:
+ raise RuntimeError("Telegram notification target is not configured")
+ self.proxy = os.getenv("TELEGRAM_PROXY") or None
+
+ async def send(self, title: str, text: str,
+ images: list[ImageAttachment] | None = None) -> None:
+ body = f"{esc(title)}\n{text}" if title else text
+ failures = 0
+ async with httpx.AsyncClient(timeout=90, proxy=self.proxy, trust_env=False) as client:
+ for chat_id in self.chat_ids:
+ try:
+ response = await client.post(
+ f"https://api.telegram.org/bot{self.token}/sendMessage",
+ json={"chat_id": chat_id, "text": body[:4096], "parse_mode": "HTML", "disable_web_page_preview": True},
+ )
+ ok = response.status_code < 400 and response.json().get("ok") is True
+ if ok:
+ for image in images or []:
+ upload = (
+ image if len(image.data) <= TELEGRAM_PHOTO_MAX_BYTES
+ else compressed_preview(image)
+ )
+ response = await client.post(
+ f"https://api.telegram.org/bot{self.token}/sendPhoto",
+ data={"chat_id": chat_id},
+ files={"photo": (
+ upload.filename, upload.data, upload.content_type
+ )},
+ )
+ if response.status_code >= 400 or response.json().get("ok") is not True:
+ ok = False
+ break
+ except Exception:
+ ok = False
+ if not ok:
+ failures += 1
+ if failures:
+ raise RuntimeError(f"Telegram delivery failed for {failures} target(s)")
diff --git a/services/notify-gateway/notifiers/wecom_webhook.py b/services/notify-gateway/notifiers/wecom_webhook.py
new file mode 100644
index 0000000..1d3ad85
--- /dev/null
+++ b/services/notify-gateway/notifiers/wecom_webhook.py
@@ -0,0 +1,126 @@
+import base64
+import hashlib
+import html
+import os
+import re
+from urllib.parse import parse_qs, urlparse
+
+import httpx
+
+from .base import ImageAttachment, Notifier
+from .image_utils import compressed_preview
+
+
+def _truncate_utf8(value: str, limit: int) -> str:
+ data = value.encode("utf-8")
+ if len(data) <= limit:
+ return value
+ return data[:limit - 3].decode("utf-8", errors="ignore") + "..."
+
+
+def _format_plain_text(title: str, text: str) -> str:
+ plain = re.sub(r"", "\n", text, flags=re.IGNORECASE)
+ plain = re.sub(r"<[^>]+>", "", plain)
+ plain = html.unescape(plain)
+ if not title:
+ return plain.strip()
+
+ field_icons = {
+ "发送人：": "👤",
+ "类别：": "🗂",
+ "待行动：": "✅",
+ "截止：": "⏰",
+ }
+ section_icons = {
+ "摘要：": "📌",
+ "原因：": "📝",
+ "原文：": "💬",
+ }
+ output: list[str] = []
+ for line in plain.splitlines():
+ stripped = line.strip()
+ matched = False
+ for label, icon in field_icons.items():
+ if stripped.startswith(label):
+ output.append(f"{icon} {stripped}")
+ matched = True
+ break
+ if matched:
+ continue
+ for label, icon in section_icons.items():
+ if stripped.startswith(label):
+ if output and output[-1] != "":
+ output.append("")
+ output.append(f"{icon} {label[:-1]}")
+ value = stripped[len(label):].strip()
+ if value:
+ output.append(value)
+ matched = True
+ break
+ if not matched:
+ output.append(line)
+
+ body = "\n".join(output).strip()
+ if not title:
+ return body
+ icon = "🔴" if "重要" in title else "🔔"
+ return f"{icon} {title}\n━━━━━━━━━━━━\n\n{body}" if body else f"{icon} {title}"
+
+
+class WecomWebhookNotifier(Notifier):
+ name = "wecom_webhook"
+
+ def __init__(self, urls: list[str] | None = None) -> None:
+ configured = urls or []
+ if not configured:
+ configured = [part.strip() for part in os.getenv(
+ "WECOM_WEBHOOK_URLS", ""
+ ).split(";") if part.strip()]
+ self.urls = [self._validate_url(url) for url in configured]
+ if not self.urls:
+ raise RuntimeError("WECOM_WEBHOOK_URLS is not configured")
+
+ @staticmethod
+ def _validate_url(value: str) -> str:
+ parsed = urlparse(value.strip())
+ key = parse_qs(parsed.query).get("key", [])
+ if (parsed.scheme != "https" or parsed.hostname != "qyapi.weixin.qq.com"
+ or parsed.path != "/cgi-bin/webhook/send" or not key or not key[0]):
+ raise RuntimeError("invalid Enterprise WeChat webhook URL")
+ return value.strip()
+
+ async def send(self, title: str, text: str,
+ images: list[ImageAttachment] | None = None) -> None:
+ content = _truncate_utf8(_format_plain_text(title, text), 2048)
+ failures = 0
+ async with httpx.AsyncClient(timeout=90, trust_env=False) as client:
+ for url in self.urls:
+ try:
+ response = await client.post(
+ url,
+ json={"msgtype": "text", "text": {"content": content}},
+ )
+ data = response.json()
+ ok = response.status_code < 400 and data.get("errcode") == 0
+ if ok:
+ for image in images or []:
+ preview = compressed_preview(image)
+ encoded = base64.b64encode(preview.data).decode("ascii")
+ digest = hashlib.md5(preview.data).hexdigest()
+ response = await client.post(
+ url,
+ json={
+ "msgtype": "image",
+ "image": {"base64": encoded, "md5": digest},
+ },
+ )
+ data = response.json()
+ if response.status_code >= 400 or data.get("errcode") != 0:
+ ok = False
+ break
+ except Exception:
+ ok = False
+ if not ok:
+ failures += 1
+ if failures:
+ raise RuntimeError(f"Enterprise WeChat webhook failed for {failures} target(s)")
diff --git a/services/notify-gateway/requirements.txt b/services/notify-gateway/requirements.txt
new file mode 100644
index 0000000..ae481e4
--- /dev/null
+++ b/services/notify-gateway/requirements.txt
@@ -0,0 +1,5 @@
+aiohttp==3.12.15
+aiosqlite==0.21.0
+httpx==0.28.1
+Pillow==12.3.0
+PyYAML==6.0.2
diff --git a/services/notify-gateway/tests/test_plain_format.py b/services/notify-gateway/tests/test_plain_format.py
new file mode 100644
index 0000000..0a6c355
--- /dev/null
+++ b/services/notify-gateway/tests/test_plain_format.py
@@ -0,0 +1,32 @@
+import inspect
+
+from notifiers.telegram import TelegramNotifier
+from notifiers.wecom_webhook import WecomWebhookNotifier, _format_plain_text
+
+
+def test_titleless_message_is_plain_and_undecorated():
+ text = "【小学群】 发送人：家长\n摘要：发送了一张图片\n准备物品：课外书"
+ result = _format_plain_text("", text)
+ assert result == text
+ assert "🔔" not in result
+ assert "📌" not in result
+ assert "━━━━━━━━━━━━" not in result
+
+
+def test_titled_messages_keep_existing_decoration():
+ result = _format_plain_text("系统通知", "发送人：测试")
+ assert result.startswith("🔔 系统通知\n━━━━━━━━━━━━")
+ assert "👤 发送人：测试" in result
+
+
+def test_original_image_file_attachments_are_disabled():
+ telegram_source = inspect.getsource(TelegramNotifier.send)
+ wecom_source = inspect.getsource(WecomWebhookNotifier.send)
+ assert "sendDocument" not in telegram_source
+ assert "upload_media" not in wecom_source
+ assert '"msgtype": "file"' not in wecom_source
+
+
+for name in sorted(value for value in globals() if value.startswith("test_")):
+ globals()[name]()
+ print(f"{name}=PASS")