易盾gevent调度检测

api/utils/sensitive.py

+"""敏感词等反垃圾相关。"""
+
+
+from engine.rpc import get_current_rpc_invoker
+from libs.spawn_task import SpawnTask, Task
+
+
+class DRIVERS(object):
+    """后续可能接入多个服务"""
+
+    YIDUN = "1"  # "易盾"
+
+
+class ENDPOINT(object):
+    """请求入口。"""
+
+    # 所有请求接入到 antispam 服务中
+    endpoint = {
+        DRIVERS.YIDUN: "antispam/yidun",
+    }
+
+    def __getitem__(self, key):
+        try:
+            return self.endpoint[key]
+        except KeyError:
+            raise Exception("没有对应的检测平台")
+
+
+class Sensitive(object):
+    """敏感词等反垃圾相关入口。"""
+
+    endpoint = ENDPOINT()
+
+    @classmethod
+    def _check_job(cls, endpoint, text):
+        """内容检测"""
+
+        rpc = get_current_rpc_invoker()
+        try:
+            res = rpc[endpoint](text=text).unwrap()
+        except:
+            res = None
+
+        return res
+
+    @classmethod
+    def check(cls, text_list, driver=DRIVERS.YIDUN):
+        """检测"""
+
+        endpoint = cls.endpoint[driver]
+
+        tasks = [Task(cls._check_job, endpoint, text) for text in text_list]
+        spawn = SpawnTask(tasks)
+
+        spawn.run()
+
+        data = []
+        if driver == DRIVERS.YIDUN:
+            for res in spawn.result:
+                data.append(cls._yidun_paraser(res))
+
+        return data
+
+    @classmethod
+    def _yidun_paraser(cls, data):
+        """易盾结果解析"""
+        # {
+        #   'code': 200,
+        #   'msg': 'ok',
+        #   'result': {
+        #       'taskId': 'd28e036d95874361916a26d406ea7db9',
+        #       'action': 2,
+        #       'labels': [
+        #           {
+        #               'label': 600,
+        #               'level': 2,
+        #               'details': {
+        #                   'hint': ['傻逼']
+        #               }
+        #           }
+        #       ]
+        #   }
+        # }
+        hints = []
+        if not data or "detail" in data:  # 程序出错啦
+            return
+
+        action = data["result"]["action"]
+        if action == 0:  # "通过"
+            return []
+
+        lables = data.get("result", {}).get("labels", [])
+        for lable in lables:
+            hints.extend(lable.get('details', {}).get("hint"))
+
+        return hints

libs/spawn_task.py

+"""利用gevent并发网络请求"""
+
+import gevent
+
+from gevent import monkey
+monkey.patch_socket()
+
+
+class Task(object):
+
+    def __init__(self, serve, *args, **kwargs):
+        self.serve = serve
+        self.args = args
+        self.kwargs = kwargs
+
+
+class SpawnTask(object):
+
+    TIMEOUT = 5
+
+    def __init__(self, tasks):
+        """tasks：任务列表，其中每个task包含任务以及对应的参数。
+
+        tasks = [
+            (get_sensitive, text1, )),
+            (get_sensitive, text2, )),
+            (get_sensitive, text3, )),
+        ]
+        """
+
+        self.tasks = tasks
+        self.jobs = []
+
+    def run(self):
+        """开始任务"""
+        # 每个任务的时间在
+
+        self._spawn()
+        self._joinall()
+
+    @property
+    def result(self):
+        return [job.value for job in self.jobs]
+
+    def _spawn(self):
+
+        self.jobs = [
+            gevent.spawn(task.serve, *task.args, **task.kwargs)
+            for task in self.tasks
+        ]
+
+    def _joinall(self):
+        gevent.joinall(self.jobs, timeout=self.TIMEOUT)