api-datamanager/Modules/Models/Score/ScoreImpl.py

162 lines
5.9 KiB
Python
Raw Normal View History

2022-07-05 10:46:00 +08:00
import time
from DBHelper.MongoHelperInstance import DB_TEST
from Modules.Models.Score.ScoreObj import ScoreModelObj
class ScoreModelImpl(object):
@staticmethod
def search_score_model(**kwargs):
"""搜索打分模型"""
total = DB_TEST.find_all_data_with_count(
'模型数据',
'打分模型',
{'模型名称': {"$regex": kwargs['model_name']}}
)
records = DB_TEST.find_all_data_with_single_sort(
'模型数据',
'打分模型',
{'模型名称': {"$regex": kwargs['model_name']}},
['模型名称', '作者', '日期', '标签'],
{'日期': -1},
kwargs['page_size'],
kwargs['page_no']
)
result = {
"total": total,
"records": records
}
return result
@staticmethod
def new_score_model(**kwargs):
"""构建打分模型"""
data = kwargs['data']
case = DB_TEST.find_single_column(
'模型数据',
'打分模型',
{'模型ID': data['model_id']},
'模型ID'
)
if case:
return '模型ID已存在'
model = ScoreModelObj()
model.model_id = data['model_id']
model.model_name = data['model_name']
model.author = data['author']
model.tags = data['tags']
model.status = 'design'
model.date = time.strftime('%Y-%m-%d', time.localtime())
model.dimension = list()
model.level_setting = list()
# 级别设置
for level in data['level_setting']:
setting = model.LevelSetting()
setting.level = level['level']
setting.score = level['score']
setting.position = level['position']
model.level_setting.append(setting)
# 维度
for dim in data['dimention']:
dimension = model.Dimension()
dimension.name = dim['dimention_name']
dimension.first_level_index = list()
for first in dim['first_index']:
first_level = dimension.FirstIndex()
first_level.name = first['index_name']
first_level.secondary_index = list()
for second in first['second_index']:
second_level = first_level.SecondaryIndex()
second_level.name = second['index_name']
second_level.standard_score = second['standard_score']
for score_set in second['score_setting']:
score_setting = second_level.ScoreSetting()
score_setting.pattern = score_set['pattern']
score_setting.interval = score_set['interval']
score_setting.score_position = score_set['score_position']
second_level.score_setting = score_setting
for index in second['bind_index']:
bind_index = second_level.BindIndex()
bind_index.index_name = index['index_name']
bind_set = bind_index.BindSet()
bind_set.name = index['bind_set']['name']
bind_set.type = index['bind_set']['type']
params = bind_set.Params
params.param = index['bind_set']['params']['param']
params.describe = index['bind_set']['params']['describe']
data_bind = params.DataBind()
data_bind.data_base = index['bind_set']['params']['data_bind']['data_base']
data_bind.data_sheet = index['bind_set']['params']['data_bind']['data_sheet']
data_bind.data_field = index['bind_set']['params']['data_bind']['data_field']
data_bind.dispose_none_string = index['bind_set']['params']['data_bind']['dispose_none_string']
data_bind.dispose_none_value = index['bind_set']['params']['data_bind']['dispose_none_value']
params.data_bind = data_bind
bind_set.params = params
bind_index.bind_set = bind_set
second_level.bind_index = bind_index
first_level.secondary_index.append(second_level)
dimension.first_level_index.append(first_level)
model.dimension.append(dimension)
return model.fields_toggle()
@staticmethod
def save_score_model(**kwargs):
model = kwargs['model']
DB_TEST.upsert_single_data(
'模型数据',
'打分模型',
{'模型名称': model['模型名称']},
model
)
return '构建成功'
@staticmethod
def compute_index(**kwargs):
"""计算指标"""
# 企业名称
name = kwargs['name']
# 级别设置
level_setting = kwargs['level_setting']
# 维度
dimension = kwargs['dimension']
result = dict()
result['企业名称'] = name
result['维度得分'] = list()
result['合计'] = 0
result['级别'] = None
def dimension_score():
"""维度得分"""
2022-07-05 16:00:35 +08:00
# for dim in dimension:
# dim_score = dict()
# dim_score['维度名称'] = dim['dimention_name']
# dim_score['一级指标'] = list()
# for first_index in dim['first_index']:
2022-07-05 10:46:00 +08:00
def check_score_model(self, **kwargs):
data = kwargs['data']
company = kwargs['company']
model_name = data['model_name']
level_setting = data['level_setting']
dimension = data['dimension']
test_result = list()
for name in company:
result = self.compute_index(name=name, level_setting=level_setting, dimension=dimension)
test_result.append(result)