Books / Python Automation in Practice: Replace Repetitive Work with Scripts
Chapter 11

Notifications — Slack, DingTalk, WeChat, and Desktop Alerts

Chapter 11: Messaging Automation — Feishu, WeCom, Telegram Bots

Email is for formal communication. Alerts need to arrive fast and in-context. Feishu, WeCom, and Telegram webhook bots are the ideal channels for pushing program status — errors trigger instantly, daily stats get broadcast automatically, with no human monitoring required. This chapter covers all major IM platforms and ends with a clean unified Notifier interface for one-line channel switching.

Feishu (Lark) Bot

Setup

import httpx

FEISHU_WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/your-token"

def feishu_send_text(text: str, webhook: str = FEISHU_WEBHOOK) -> bool:
    resp = httpx.post(webhook, json={"msg_type": "text", "content": {"text": text}}, timeout=10)
    return resp.json().get("StatusCode") == 0

def feishu_send_card(title: str, content: str, color: str = "red",
                     webhook: str = FEISHU_WEBHOOK) -> bool:
    """color: red / green / orange / blue"""
    payload = {
        "msg_type": "interactive",
        "card": {
            "config": {"wide_screen_mode": True},
            "header": {"title": {"tag": "plain_text", "content": title}, "template": color},
            "elements": [{"tag": "div", "text": {"tag": "lark_md", "content": content}}],
        },
    }
    resp = httpx.post(webhook, json=payload, timeout=10)
    return resp.json().get("StatusCode") == 0

WeCom (Enterprise WeChat) Bot

import httpx

WECOM_WEBHOOK = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=your-key"

def wecom_send_markdown(content: str, webhook: str = WECOM_WEBHOOK) -> bool:
    resp = httpx.post(webhook, json={"msgtype": "markdown", "markdown": {"content": content}}, timeout=10)
    return resp.json().get("errcode") == 0

def wecom_send_text(content: str, mention_all: bool = False) -> bool:
    payload = {
        "msgtype": "text",
        "text": {
            "content": content,
            "mentioned_list": ["@all"] if mention_all else [],
        }
    }
    resp = httpx.post(WECOM_WEBHOOK, json=payload, timeout=10)
    return resp.json().get("errcode") == 0

Telegram Bot

Creating a Bot

import httpx, os

def telegram_send(text: str, chat_id: str = None, token: str = None) -> bool:
    token = token or os.getenv("TELEGRAM_BOT_TOKEN")
    chat_id = chat_id or os.getenv("TELEGRAM_CHAT_ID")
    resp = httpx.post(
        f"https://api.telegram.org/bot{token}/sendMessage",
        json={"chat_id": chat_id, "text": text, "parse_mode": "Markdown"},
        timeout=10,
    )
    return resp.json().get("ok", False)

Bot with Command Handlers

from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes
import os

async def cmd_start(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text("Commands:\n/stats — today's data\n/help — show this menu")

async def cmd_stats(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
    # Query your database here
    await update.message.reply_text("*Today's Stats*\nOrders: 342\nRevenue: $12,800", parse_mode="Markdown")

def run_bot():
    app = Application.builder().token(os.getenv("TELEGRAM_BOT_TOKEN")).build()
    app.add_handler(CommandHandler("start", cmd_start))
    app.add_handler(CommandHandler("stats", cmd_stats))
    app.run_polling()

if __name__ == "__main__":
    run_bot()

DingTalk Bot (with Signature)

import hashlib, hmac, base64, time, urllib.parse, httpx, os

def dingtalk_send(title: str, content: str) -> bool:
    webhook = os.getenv("DINGTALK_WEBHOOK")
    secret = os.getenv("DINGTALK_SECRET")
    timestamp = str(round(time.time() * 1000))
    sign_str = f"{timestamp}\n{secret}"
    hmac_code = hmac.new(secret.encode(), sign_str.encode(), digestmod=hashlib.sha256).digest()
    sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
    url = f"{webhook}&timestamp={timestamp}&sign={sign}"
    payload = {"msgtype": "markdown", "markdown": {"title": title, "text": f"## {title}\n\n{content}"}}
    resp = httpx.post(url, json=payload, timeout=10)
    return resp.json().get("errcode") == 0

Unified Notifier Interface

import os, logging
from abc import ABC, abstractmethod
import httpx

logger = logging.getLogger(__name__)

class BaseNotifier(ABC):
    @abstractmethod
    def send(self, title: str, content: str, level: str = "info") -> bool: ...

class FeishuNotifier(BaseNotifier):
    COLOR_MAP = {"info": "blue", "warn": "orange", "error": "red"}
    def __init__(self, webhook: str): self.webhook = webhook
    def send(self, title: str, content: str, level: str = "info") -> bool:
        payload = {"msg_type": "interactive", "card": {
            "config": {"wide_screen_mode": True},
            "header": {"title": {"tag": "plain_text", "content": title},
                       "template": self.COLOR_MAP.get(level, "blue")},
            "elements": [{"tag": "div", "text": {"tag": "lark_md", "content": content}}],
        }}
        try:
            return httpx.post(self.webhook, json=payload, timeout=10).json().get("StatusCode") == 0
        except Exception as e:
            logger.error(f"Feishu failed: {e}"); return False

class TelegramNotifier(BaseNotifier):
    ICONS = {"info": "ℹ️", "warn": "⚠️", "error": "🚨"}
    def __init__(self, token: str, chat_id: str): self.token, self.chat_id = token, chat_id
    def send(self, title: str, content: str, level: str = "info") -> bool:
        text = f"{self.ICONS.get(level, 'ℹ️')} *{title}*\n\n{content}"
        try:
            resp = httpx.post(f"https://api.telegram.org/bot{self.token}/sendMessage",
                              json={"chat_id": self.chat_id, "text": text, "parse_mode": "Markdown"}, timeout=10)
            return resp.json().get("ok", False)
        except Exception as e:
            logger.error(f"Telegram failed: {e}"); return False

class Notifier:
    def __init__(self): self._channels: list[BaseNotifier] = []
    def add_channel(self, n: BaseNotifier) -> "Notifier": self._channels.append(n); return self
    def notify(self, title: str, content: str, level: str = "info") -> dict:
        return {type(ch).__name__: ch.send(title, content, level) for ch in self._channels}

def build_notifier_from_env() -> Notifier:
    n = Notifier()
    if w := os.getenv("FEISHU_WEBHOOK"): n.add_channel(FeishuNotifier(w))
    if (t := os.getenv("TELEGRAM_BOT_TOKEN")) and (c := os.getenv("TELEGRAM_CHAT_ID")):
        n.add_channel(TelegramNotifier(t, c))
    return n

Project: Multi-Channel Alert System

import logging, traceback, functools
from datetime import datetime

logging.basicConfig(
    level=logging.INFO,
    handlers=[logging.FileHandler("app.log", encoding="utf-8"), logging.StreamHandler()]
)
logger = logging.getLogger("app")
notifier = build_notifier_from_env()

def alert_on_error(func):
    """Decorator: auto-catch exceptions and push notifications."""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            tb = traceback.format_exc()
            title = f"Error in {func.__name__}"
            content = (
                f"**Type**: {type(e).__name__}\n"
                f"**Message**: {str(e)}\n"
                f"**Time**: {datetime.now():%Y-%m-%d %H:%M:%S}\n"
                f"**Traceback** (last 3 lines):\n```\n{chr(10).join(tb.strip().splitlines()[-3:])}\n```"
            )
            logger.error(f"{title}\n{tb}")
            notifier.notify(title, content, level="error")
            raise
    return wrapper

@alert_on_error
def run_daily_job():
    logger.info("Starting daily job...")
    # Your business logic here — any exception triggers an alert automatically
    pass

if __name__ == "__main__":
    try:
        run_daily_job()
    except Exception:
        pass

Previous

Next
Chapter 12: AI API Integration
Rate this chapter
4.8  / 5  (26 ratings)

💬 Comments