限流、违规报警优化

app/domains/analysis_task/rate_limited_article_filter/_utils.py

@@ -2,6 +2,7 @@ from collections import defaultdict
 from typing import Dict, List
 
 from app.core.observability import LogService
+from app.infra.external import feishu_robot
 
 
 class RateLimitedArticleUtils:
@@ -125,3 +126,10 @@ class RateLimitedArticleUtils:
             )
 
         return results
+
+    @staticmethod
+    async def feishu_alert(title, detail: Dict, mention: bool = False):
+        return await feishu_robot.bot(
+            title=title, detail=detail, mention=mention, env="prod"
+        )
+

app/domains/analysis_task/rate_limited_article_filter/entrance.py

@@ -6,6 +6,7 @@ from typing import Dict
 from app.core.config import GlobalConfigSettings
 from app.core.database import DatabaseManager
 from app.core.observability import LogService
+
 from app.infra.internal import delete_illegal_gzh_articles
 from app.infra.shared import run_tasks_with_asyncio_task_group
 
@@ -30,23 +31,26 @@ class RateLimitedArticleFilter(RateLimitedArticleMapper):
         """处理单个文章的异步任务"""
         title = data["title"]
         title_md5 = hashlib.md5(title.encode("utf-8")).hexdigest()
-        remark = json.dumps(
-            {
-                "发文数量": data["publish_count"],
-                "限流数量": data["low_read_count"],
-                "限流比例": data["low_read_ratio"],
-                "周期": data["days"],
-                "触发规则": data["trigger_rules"],
-                "执行日期": datetime.datetime.today().strftime("%Y-%m-%d"),
-            },
-            ensure_ascii=False,
-        )
+        remark = {
+            "文章标题": title,
+            "发文数量": data["publish_count"],
+            "限流数量": data["low_read_count"],
+            "限流比例": data["low_read_ratio"],
+            "统计周期": data["days"],
+            "触发规则": data["trigger_rules"],
+            "执行日期": datetime.datetime.today().strftime("%Y-%m-%d"),
+        }
         try:
             insert_rows = await self.save_record(
-                article_tuple=(title_md5, title, remark)
+                article_tuple=(title_md5, title, json.dumps(remark, ensure_ascii=False))
             )
             if insert_rows:
-                gh_id = data["gh_ids"][0]
+                await self.tool.feishu_alert(
+                    title="限流策略命中文章",
+                    detail=remark,
+                    mention=True
+                )
+                gh_id = data["gh_ids"][0]    # 删文是标题粒度的，删文过程中所有账号的标题都会被删除， gh_id 只作为一个参数传递即可
                 await delete_illegal_gzh_articles(
                     gh_id=gh_id, title=title, delete_flag=self.RATE_LIMITED
                 )

app/domains/monitor_tasks/gzh_article_monitor/entrance.py

@@ -49,7 +49,7 @@ class InnerArticleMonitorTask(GzhArticleMonitorConst):
                             article=(wx_sn.decode("utf-8"), title, error_detail)
                         )
                         await self.tool.feishu_alert(
-                            title="模式违规报警", detail=article_detail, mention=False
+                            title="长文——模式违规报警", detail=article_detail, mention=True
                         )
                         return
 
@@ -65,15 +65,16 @@ class InnerArticleMonitorTask(GzhArticleMonitorConst):
                     )
 
                     if insert_row:
+                        # illegal_articles 里面没有违规文章、首次出现报警
+                        await self.tool.feishu_alert(
+                            title="长文——文章违规告警", detail=article_detail, mention=True
+                        )
                         # 判断文章是否删过
                         title_md5 = self.tool.title_to_md5(title)
                         if await self.mapper.whether_title_unsafe(title_md5=title_md5):
-                            # 说明文章已经删过，无需处理和报警
+                            # 说明文章已经删过，无需再删文处理
                             return
 
-                        await self.tool.feishu_alert(
-                            title="文章违规告警", detail=article_detail, mention=False
-                        )
                         await self.tool.delete_illegal_articles(gh_id, title)
 
                 case _:

app/infra/external/feishu.py

@@ -25,6 +25,14 @@ class Feishu:
     # rank_bot
     rank_monitor_bot = "https://open.feishu.cn/open-apis/bot/v2/hook/f9dae7ba-decf-436b-b438-41994c35af1e"
 
+    # 用户名与手机号映射
+    name_phone_dict = {
+        "卓异": "18624010360",
+        "俊辉": "18801281360",
+        "范军": "15200827642",
+        "林强": "15810236123",
+    }
+
     def __init__(self):
         self.token = None
         self.headers = {"Content-Type": "application/json"}
@@ -46,6 +54,36 @@ class Feishu:
         tenant_access_token = response["tenant_access_token"]
         self.token = tenant_access_token
 
+    async def get_user_open_id(self, username):
+        """
+        根据用户名获取飞书 open_id
+
+        :param username: 用户名
+        :return: open_id
+        """
+        if not self.token:
+            await self.fetch_token()
+
+        mobile = self.name_phone_dict.get(username)
+        if not mobile:
+            return None
+
+        url = "https://open.feishu.cn/open-apis/user/v1/batch_get_id"
+        headers = {
+            "Authorization": f"Bearer {self.token}",
+            "Content-Type": "application/json; charset=utf-8",
+        }
+        params = {"mobiles": [mobile]}
+
+        async with AsyncHttpClient() as client:
+            response = await client.get(url=url, params=params, headers=headers)
+
+        try:
+            open_id = response["data"]["mobile_users"][mobile][0]["open_id"]
+            return open_id
+        except (KeyError, IndexError):
+            return None
+
 
 class FeishuSheetApi(Feishu):
     async def prepend_value(self, sheet_token, sheet_id, ranges, values):
@@ -87,6 +125,24 @@ class FeishuSheetApi(Feishu):
 
 
 class FeishuBotApi(Feishu):
+    async def create_mention_text(self, usernames):
+        """
+        创建 @ 用户的文本
+
+        :param usernames: 用户名列表
+        :return: @ 用户的 markdown 文本
+        """
+        if not usernames:
+            return ""
+
+        mention_parts = []
+        for username in usernames:
+            open_id = await self.get_user_open_id(username)
+            if open_id:
+                mention_parts.append(f"<at id={open_id}></at>")
+
+        return " ".join(mention_parts) + "\n" if mention_parts else ""
+
     @classmethod
     def create_feishu_columns_sheet(
         cls,
@@ -151,14 +207,30 @@ class FeishuBotApi(Feishu):
                 }
 
     # 表格形式
-    def create_feishu_table(self, title, columns, rows, mention):
+    async def create_feishu_table(self, title, columns, rows, mention, mention_users=None):
+        """
+        创建飞书表格消息
+
+        :param title: 标题
+        :param columns: 列定义
+        :param rows: 行数据
+        :param mention: 是否 @ 所有人
+        :param mention_users: 要 @ 的具体用户列表，例如 ["mark"]
+        """
+        mention_element = self.not_mention
+        if mention:
+            mention_element = self.mention_all
+        elif mention_users:
+            mention_text = await self.create_mention_text(mention_users)
+            mention_element = {"content": mention_text, "tag": "lark_md"}
+
         table_base = {
             "header": {
                 "template": "blue",
                 "title": {"content": title, "tag": "plain_text"},
             },
             "elements": [
-                self.mention_all if mention else self.not_mention,
+                mention_element,
                 {
                     "tag": "table",
                     "page_size": len(rows) + 1,
@@ -178,15 +250,27 @@ class FeishuBotApi(Feishu):
         }
         return table_base
 
-    def create_feishu_bot_obj(self, title, mention, detail):
+    async def create_feishu_bot_obj(self, title, mention, detail, mention_users=None):
         """
         create feishu bot object
+
+        :param title: 标题
+        :param mention: 是否 @ 所有人
+        :param detail: 详细内容
+        :param mention_users: 要 @ 的具体用户列表，例如 ["luojunhui"]
         """
+        mention_element = self.not_mention
+        if mention:
+            mention_element = self.mention_all
+        elif mention_users:
+            mention_text = await self.create_mention_text(mention_users)
+            mention_element = {"content": mention_text, "tag": "lark_md"}
+
         return {
             "elements": [
                 {
                     "tag": "div",
-                    "text": self.mention_all if mention else self.not_mention,
+                    "text": mention_element,
                 },
                 {
                     "tag": "div",
@@ -201,8 +285,24 @@ class FeishuBotApi(Feishu):
 
     # bot
     async def bot(
-        self, title, detail, mention=True, table=False, env="long_articles_task"
+        self,
+        title,
+        detail,
+        mention=True,
+        table=False,
+        env="long_articles_task",
+        mention_users=None,
     ):
+        """
+        发送飞书机器人消息
+
+        :param title: 标题
+        :param detail: 详细内容
+        :param mention: 是否 @ 所有人
+        :param table: 是否为表格形式
+        :param env: 环境，决定发送到哪个机器人
+        :param mention_users: 要 @ 的具体用户列表，例如 ["luojunhui"]
+        """
         match env:
             case "dev":
                 url = self.long_articles_bot_dev
@@ -223,15 +323,16 @@ class FeishuBotApi(Feishu):
 
         headers = {"Content-Type": "application/json"}
         if table:
-            card = self.create_feishu_table(
+            card = await self.create_feishu_table(
                 title=title,
                 columns=detail["columns"],
                 rows=detail["rows"],
                 mention=mention,
+                mention_users=mention_users,
             )
         else:
-            card = self.create_feishu_bot_obj(
-                title=title, mention=mention, detail=detail
+            card = await self.create_feishu_bot_obj(
+                title=title, mention=mention, detail=detail, mention_users=mention_users
             )
 
         data = {"msg_type": "interactive", "card": card}