28-----使用工作流推送通知

今天药忘吃喽~ 2023-09-27 19:38 2阅读 0赞

工作流是一种简单但功能强大的构造，它接受可调用元素并返回元素。他们正在流式传输并批量处理数据，从而可以高效地处理大量数据。

本文将展示如何使用工作流在某些事件触发时推送通知。使用此方法，可以创建内容的活动提要。

安装依赖  
安装txtai和所有依赖项。  
pip install txtai\[pipeline,workflow\]  
创建工作流通知操作  
工作流运行一系列任务来转换和处理数据。此部分创建可用作工作流操作的可调用对象。

下面的操作将事件推送到Slack。虽然这里使用了 Slack，但任何通知服务都可以很容易地替换为（Zapier、IFTT等）。

假定已安装并准备好使用 Slack 工作区和应用程序。有关设置新的 Slack 应用程序和通过 API 发布消息的更多信息，请参阅此综合示例。

可以从 Slack Web 界面找到频道 ID。登录 Slack 并单击要发布消息的频道。是channel idURL 的最后一部分。  
https://app.slack.com/client//  
import logging  
import requests

## Uncomment and set. The following are dummy parameters. Your parameters should not be publicly shared! ##

## AUTH = “xoxb-not-a-real-token-this-will-not-work” ##

## CHANNEL = “C0XXXXXXXXX” ##

## URL = “https://slack.com/api/chat.postMessage” ##

## Logging configuration ##

logger = logging.getLogger(**name**)  
logger.setLevel(logging.INFO)

class Slack:  
def **init**(self):  
self.alerts = set()

def __call__(self, elements):
        for alert in elements:
            uid, text, _ = self.extract(alert)
            if uid not in self.alerts:
                logger.info("Sending alert: %s", alert)
                self.alerts.add(uid)
    
                headers = {
                    "Content-type": "application/json",
                    "Authorization": f"Bearer {AUTH}"
                }
    
                response = requests.post(URL, headers=headers, json={
                    "channel": CHANNEL,
                    "text": f"{text} {uid}"
                }).json()
    
                if not response["ok"]:
                    logger.error(response)
    
        return elements
    
    def extract(self, alert):
        if isinstance(alert, dict):
            return (alert["id"], alert["text"], None)
    
        return alert

构建语义通知工作流程  
接下来我们将创建一个通知工作流。下面的示例为最热门的 Hac