|
|
|
|
#!/usr/bin/env python
|
|
|
|
|
# -*- encoding: utf-8 -*-
|
|
|
|
|
'''
|
|
|
|
|
@Date:2022/07/19 13:54:49
|
|
|
|
|
'''
|
|
|
|
|
|
|
|
|
|
import sys
|
|
|
|
|
sys.path.append(".")
|
|
|
|
|
|
|
|
|
|
import random
|
|
|
|
|
from sqlalchemy import or_, desc
|
|
|
|
|
from common.utils import Utils
|
|
|
|
|
from db_logic.db_base import Repository
|
|
|
|
|
from models.warning_models import EntityWarning
|
|
|
|
|
#预警信息业务逻辑类
|
|
|
|
|
class BllWarning(Repository):
|
|
|
|
|
def __init__(self, entityType=EntityWarning):
|
|
|
|
|
return super().__init__(entityType)
|
|
|
|
|
|
|
|
|
|
#获取预警列表
|
|
|
|
|
def getWarningList(self,customerId,pageParam,keyWord=''):
|
|
|
|
|
keyWord='%'+keyWord+'%'
|
|
|
|
|
orm_query= self.findList().filter(
|
|
|
|
|
EntityWarning.customer_id==customerId
|
|
|
|
|
).filter(
|
|
|
|
|
or_(EntityWarning.object_name.like(keyWord),
|
|
|
|
|
EntityWarning.warning_user_name.like(keyWord))
|
|
|
|
|
).order_by(desc(EntityWarning.warning_date))
|
|
|
|
|
return self.queryPage(orm_query,pageParam)
|
|
|
|
|
|
|
|
|
|
#获取预警类型数量
|
|
|
|
|
def getWarningCount(self):
|
|
|
|
|
recordList=self.findList().all()
|
|
|
|
|
# 过期预警列表
|
|
|
|
|
gqList = [record for record in recordList if record.object_type == '2']
|
|
|
|
|
# 余量预警列表
|
|
|
|
|
ylList = [record for record in recordList if record.object_type == '3']
|
|
|
|
|
return (len(recordList),len(gqList),len(ylList))
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def get_waring_type_classify(self):
|
|
|
|
|
sql_all = """
|
|
|
|
|
SELECT count(warning_id) type_number, object_type FROM `rms_warning` GROUP BY object_type
|
|
|
|
|
"""
|
|
|
|
|
data_list = self.execute(sql_all).fetchall()
|
|
|
|
|
all_number = sum([i[0] for i in data_list])
|
|
|
|
|
data_list = Utils.msyql_table_model(data_list)
|
|
|
|
|
new_data_list = []
|
|
|
|
|
for i in data_list:
|
|
|
|
|
new_dic = {
|
|
|
|
|
"ratio": str(round(i["type_number"] / all_number, 2) * 100) + "%"
|
|
|
|
|
}
|
|
|
|
|
new_dic.update(**i)
|
|
|
|
|
new_data_list.append(new_dic)
|
|
|
|
|
return new_data_list, all_number
|
|
|
|
|
|
|
|
|
|
def create_bluk_waring(self):
|
|
|
|
|
inster_list = []
|
|
|
|
|
for i in range(1000):
|
|
|
|
|
inster_list.append(
|
|
|
|
|
EntityWarning(
|
|
|
|
|
customer_id='1c39cb2a-07f8-11ed-972d-f47b094925e1',
|
|
|
|
|
object_type=random.randint(1, 7),
|
|
|
|
|
object_id='1c39cb24-07f8-11ed-abd4-f47b094925e1',
|
|
|
|
|
object_name=f'测试对象名称{i}',
|
|
|
|
|
warning_content=f'测试预警数据{i}',
|
|
|
|
|
warning_date=Utils.get_str_datetime()
|
|
|
|
|
)
|
|
|
|
|
)
|
|
|
|
|
return self.insert_many(inster_list)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# if __name__ == '__main__':
|
|
|
|
|
# a = BllWarning().get_waring_type_classify()
|
|
|
|
|
# base_ratio = 0
|
|
|
|
|
# base_number = 0
|
|
|
|
|
# for i in a:
|
|
|
|
|
# print(i)
|
|
|
|
|
# # base_ratio += i["ratio"]
|
|
|
|
|
# base_number += i['type_number']
|
|
|
|
|
# print(base_ratio)
|
|
|
|
|
# print(base_number)
|
