jd-tb-yaofang-1/commons/feishu_webhook.py

import requests
import json
import time
from commons.config import FEISHU_WEBHOOK_URL

WEBHOOK_URL = FEISHU_WEBHOOK_URL


def send_text(text):
    data = {
        "msg_type": "text",
        "content": {"text": text},
    }
    headers = {"Content-Type": "application/json"}
    try:
        response = requests.post(
            WEBHOOK_URL,
            headers=headers,
            data=json.dumps(data),
            timeout=5
        )
    except Exception as e:
        pass


def send_error_card(task_name, err_msg, mention_all=False):
    """发送异常红色卡片"""
    content = (
        f"**任务**：{task_name}\n"
        f"**状态**：❌ 失败\n"
        f"**错误**：{err_msg}\n"
        f"**时间**：{time.strftime('%Y-%m-%d %H:%M:%S')}"
    )
    if mention_all:
        content = f"<at id=all></at>\n{content}"

    data = {
        "msg_type": "interactive",
        "card": {
            "config": {"wide_screen_mode": True},
            "header": {
                "template": "red",
                "title": {"tag": "plain_text", "content": "异常告警"},
            },
            "elements": [
                {
                    "tag": "div",
                    "text": {"tag": "lark_md", "content": content},
                }
            ],
        },
    }
    headers = {"Content-Type": "application/json"}
    response = requests.post(
        WEBHOOK_URL,
        headers=headers,
        data=json.dumps(data),
        timeout=5,
    )
    print(response.json())


# 使用示例
if __name__ == "__main__":
    account1 = "aaaaa"
    plat_form = "京东"
    drug = "999小儿感冒颗粒24粒"
    drug_count = 128
    text = f"**重要通知** {str(time.strftime("%Y-%m-%d %H:%M:%S"))}\n 账号{account1}, {plat_form} 采集 {drug} 数据 {drug_count} 条"
    send_text(text)