mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-07-22 04:16:07 +00:00
根据建议修改
This commit is contained in:
+1
-1
@@ -116,7 +116,7 @@ baidu_api_key = ""
|
|||||||
baidu_secret_key =""
|
baidu_secret_key =""
|
||||||
|
|
||||||
#不合规消息自定义返回
|
#不合规消息自定义返回
|
||||||
illgalmessage = "[百度云]请珍惜机器人,当前返回内容不合规"
|
inappropriate_message_tips = "[百度云]请珍惜机器人,当前返回内容不合规"
|
||||||
|
|
||||||
# 启动时是否发送赞赏码
|
# 启动时是否发送赞赏码
|
||||||
# 仅当使用量已经超过2048字时发送
|
# 仅当使用量已经超过2048字时发送
|
||||||
|
|||||||
+40
-26
@@ -2,44 +2,64 @@
|
|||||||
import re
|
import re
|
||||||
import requests
|
import requests
|
||||||
import json
|
import json
|
||||||
from config import baidu_check, baidu_api_key, baidu_secret_key, illgalmessage
|
import importlib
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
|
# 默认值( 兼容性考虑 )
|
||||||
|
|
||||||
# 然后可以通过config.check, config.baidu_api_key等方式来使用这些变量。
|
baidu_check = False
|
||||||
|
baidu_api_key = ""
|
||||||
|
baidu_secret_key = ""
|
||||||
|
inappropriate_message_tips = "[百度云]请珍惜机器人,当前返回内容不合规"
|
||||||
|
|
||||||
def get_access_token():
|
# 初始化
|
||||||
"""
|
|
||||||
使用 AK,SK 生成鉴权签名(Access Token)
|
|
||||||
:return: access_token,或是None(如果错误)
|
|
||||||
"""
|
|
||||||
url = "https://aip.baidubce.com/oauth/2.0/token"
|
|
||||||
params = {"grant_type": "client_credentials", "client_id": baidu_api_key,
|
|
||||||
"client_secret": baidu_secret_key}
|
|
||||||
return str(requests.post(url, params=params).json().get("access_token"))
|
|
||||||
|
|
||||||
|
|
||||||
# 百度云审核URL
|
|
||||||
baidu_url = "https://aip.baidubce.com/rest/2.0/solution/v1/text_censor/v2/user_defined?access_token=" \
|
|
||||||
+ get_access_token()
|
|
||||||
|
|
||||||
|
config = importlib.import_module('config')
|
||||||
|
baidu_check = config.baidu_check
|
||||||
|
baidu_api_key = config.baidu_api_key
|
||||||
|
baidu_secret_key = config.baidu_secret_key
|
||||||
|
inappropriate_message_tips = config.inappropriate_message_tips
|
||||||
|
|
||||||
class ReplyFilter:
|
class ReplyFilter:
|
||||||
sensitive_words = []
|
sensitive_words = []
|
||||||
|
|
||||||
def __init__(self, sensitive_words: list):
|
def __init__(self, sensitive_words: list):
|
||||||
self.sensitive_words = sensitive_words
|
self.sensitive_words = sensitive_words
|
||||||
|
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
|
||||||
|
|
||||||
def process(self, message: str) -> str:
|
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 baidu_check:
|
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": baidu_api_key,
|
||||||
|
"client_secret": baidu_secret_key}).json().get("access_token"))
|
||||||
|
|
||||||
# 百度云审核
|
# 百度云审核
|
||||||
payload = "text=" + message
|
payload = "text=" + message
|
||||||
logging.info("向百度云发送:" + payload)
|
logging.info("向百度云发送:" + payload)
|
||||||
headers = {'Content-Type': 'application/x-www-form-urlencoded', 'Accept': 'application/json'}
|
headers = {'Content-Type': 'application/x-www-form-urlencoded', 'Accept': 'application/json'}
|
||||||
response = requests.request("POST", baidu_url, headers=headers, data=payload.encode('utf-8'))
|
|
||||||
|
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)
|
response_dict = json.loads(response.text)
|
||||||
# 处理百度云审核结果
|
|
||||||
if "error_code" in response_dict:
|
if "error_code" in response_dict:
|
||||||
error_msg = response_dict.get("error_msg")
|
error_msg = response_dict.get("error_msg")
|
||||||
logging.info(f"百度云判定出错,错误信息:{error_msg}")
|
logging.info(f"百度云判定出错,错误信息:{error_msg}")
|
||||||
@@ -51,14 +71,8 @@ class ReplyFilter:
|
|||||||
return message
|
return message
|
||||||
else:
|
else:
|
||||||
logging.info(f"百度云判定结果:{conclusion}")
|
logging.info(f"百度云判定结果:{conclusion}")
|
||||||
conclusion = illgalmessage
|
conclusion = inappropriate_message_tips
|
||||||
# 返回百度云审核结果
|
# 返回百度云审核结果
|
||||||
return conclusion
|
return conclusion
|
||||||
|
|
||||||
# 本地关键词屏蔽
|
|
||||||
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]))
|
|
||||||
return message
|
return message
|
||||||
|
|||||||
Reference in New Issue
Block a user