import logging import requests logger = logging.getLogger(__name__) class FeishuNotifier: def __init__(self, webhook_url): self.webhook_url = webhook_url def send(self, payload): try: resp = requests.post( self.webhook_url, json=payload, timeout=5 ) if resp.status_code != 200 or resp.json().get('StatusCode') != 0: raise Exception(f'status_code: {resp.status_code}, resp: {resp.json()}') return True except Exception as e: logger.error(f'send message to lark failed, payload: {payload}, {e}') return False def send_markdown(self, title, content): return self.send({ "msg_type": "interactive", "card": { "config": { "wide_screen_mode": True, "enable_forward": True }, "header": { "title": { "tag": "plain_text", "content": title }, "template": "red" }, "elements": [{ "tag": "div", "text": { "content": content, "tag": "lark_md" } }], } })