Merge pull request #212 from Haibersut/feat-baiducloud

增加百度云内容审核

pkg/qqbot/filter.py

@@ -1,19 +1,71 @@
 # 敏感词过滤模块
 import re
+import requests
+import json
+import logging
 
 
 class ReplyFilter:
-
     sensitive_words = []
 
+    # 默认值( 兼容性考虑 )
+    baidu_check = False
+    baidu_api_key = ""
+    baidu_secret_key = ""
+    inappropriate_message_tips = "[百度云]请珍惜机器人，当前返回内容不合规"
+
     def __init__(self, sensitive_words: list):
         self.sensitive_words = sensitive_words
+        import config
+        if hasattr(config, 'baidu_check') and hasattr(config, 'baidu_api_key') and hasattr(config, 'baidu_secret_key'):
+            self.baidu_check = config.baidu_check
+            self.baidu_api_key = config.baidu_api_key
+            self.baidu_secret_key = config.baidu_secret_key
+            self.inappropriate_message_tips = config.inappropriate_message_tips
 
     def process(self, message: str) -> str:
+
+        # 本地关键词屏蔽
         for word in self.sensitive_words:
             match = re.findall(word, message)
             if len(match) > 0:
                 for i in range(len(match)):
                     message = message.replace(match[i], "*" * len(match[i]))
 
+        # 百度云审核
+        if self.baidu_check:
+
+            # 百度云审核URL
+            baidu_url = "https://aip.baidubce.com/rest/2.0/solution/v1/text_censor/v2/user_defined?access_token=" + \
+                str(requests.post("https://aip.baidubce.com/oauth/2.0/token", 
+                      params={"grant_type": "client_credentials", 
+                              "client_id": self.baidu_api_key,
+                              "client_secret": self.baidu_secret_key}).json().get("access_token"))
+
+            # 百度云审核
+            payload = "text=" + message
+            logging.info("向百度云发送:" + payload)
+            headers = {'Content-Type': 'application/x-www-form-urlencoded', 'Accept': 'application/json'}
+
+            if isinstance(payload, str):
+                payload = payload.encode('utf-8')
+
+            response = requests.request("POST", baidu_url, headers=headers, data=payload)
+            response_dict = json.loads(response.text)
+
+            if "error_code" in response_dict:
+                error_msg = response_dict.get("error_msg")
+                logging.warning(f"百度云判定出错，错误信息：{error_msg}")
+                conclusion = f"百度云判定出错，错误信息：{error_msg}\n以下是原消息：{message}"
+            else:
+                conclusion = response_dict["conclusion"]
+                if conclusion in ("合规"):
+                    logging.info(f"百度云判定结果：{conclusion}")
+                    return message
+                else:
+                    logging.warning(f"百度云判定结果：{conclusion}")
+                    conclusion = self.inappropriate_message_tips
+            # 返回百度云审核结果
+            return conclusion
+        
         return message