data_quality_server/quality_server.py

# coding:utf-8

from a2s.a2s_server import simple_params, watch
from a2s.tools import json_deserialize, json_serialize
from app import check
from docs.config import ReluMongodb
from util.mogodb_helper import MongoDBInterface
import time

ReluClient = MongoDBInterface(ReluMongodb)

Rules = {}
ClearTime = 0


def clear_timeout_rules():
    """
    清理超时的规则
    """
    global Rules
    p_clear = []
    for rules_id, r in Rules.items():
        if time.time() - r["time"] > 5:
            p_clear.append(rules_id)
    for rules_id in p_clear:
        Rules.pop(rules_id)


def get_rule(rules_id):
    """
    获取规则
    """
    global Rules
    global ClearTime
    if time.time() - ClearTime > 5:
        clear_timeout_rules()
        ClearTime = time.time()
    if rules_id in Rules:
        return Rules[rules_id]["relus"]
    else:
        rule_row = ReluClient.find_relus(ReluMongodb["col"], rules_id)
        if rule_row:
            rules = rule_row.get("rules", {})
            Rules[rules_id] = {}
            Rules[rules_id]["time"] = int(time.time())
            Rules[rules_id]["relus"] = rules
            return rules
    return None


@simple_params
@watch
def server_start(data: bytes, *args, **kwargs):
    data = json_deserialize(data)
    row = data.get("data", {})
    rules_id = data.get("rules_id", None)
    rules = get_rule(rules_id)
    if not rules or not row:
        data = {
            "code": 500,
            "msg": "失败：规则与检测数据不能为空",
            "data": {}
        }
    else:
        result = check(row, rules)
        data = {
            "code": 200,
            "msg": "成功",
            "data": result
        }
    return json_serialize(data)


if __name__ == '__main__':
    server_start()
    # 本地启动服务
    # python -u quality_server.py -host 192.168.3.240 -port 19090 -subject test_quality_bid -queue main -manager_subject monitorRelease -server_host 127.0.0.1
    # server_start(host="192.168.3.240",port=19090,subject="test_quality_bid",queue="main")