tfse-etl-api-v0.2/RatingData/scripts/esg/social.py

249 lines
7.4 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import operator
def social_turnover_salary(param):
"""
离职率/人均薪酬水平
Parameters:
param dict 数据库数据
Returns:
result dict 清洗好的数据
"""
# Params
three_year_data = sorted(param['社会问卷']['近三年公司数据'], key=operator.itemgetter('年份'), reverse=False)
# Returns
result = dict()
def date_range():
"""
评价周期年度区间
Parameters:
param dict 环境问卷
Returns:
result str 区间数据
"""
year_begin = three_year_data[0]['年份']
year_end = three_year_data[2]['年份']
res = year_begin + '-' + year_end
return res
def turnover_ratio(year_param):
"""
每年离职比例
Parameters:
param dict 年度社会问卷数据
Returns:
result type 该年度离职比例
"""
# 离职率
total = year_param['员工总数(年底)']
turnover = year_param['当年离职人数']
rate = round(turnover / total * 100, 2)
return str(rate) + '%'
def calculate_three_year_data():
"""
计算三年数据并转为字符串
Parameters:
'-'
Returns:
result str 三年数据用顿号隔开
"""
res = dict()
year_turnover = list()
yart_salary_level = list()
for item in three_year_data:
turnover_res = turnover_ratio(item)
year_turnover.append(turnover_res)
yart_salary_level.append(item['人均薪酬水平(元/月)'])
# 列表中元素类型float转str
yart_salary_level_new = [str(x) for x in yart_salary_level]
# Returns-part1
res['part_01'] = dict()
res['part_01']['date_range'] = date_range()
res['part_01']['turnover_ratio'] = ''.join(year_turnover)
res['part_01']['year_03'] = three_year_data[2]['年份']
res['part_01']['industry_indicators'] = None
res['part_01']['industry_indicators_average'] = None
res['part_01']['industry_indicators_median'] = None
res['part_01']['median_comparison'] = None
# part2
res['part_02'] = dict()
res['part_02']['date_range'] = date_range()
res['part_02']['salary_level'] = ''.join(yart_salary_level_new)
res['part_02']['year_02'] = three_year_data[1]['年份']
res['part_02']['year_03'] = three_year_data[2]['年份']
res['part_02']['year01_on_year02'] = round((yart_salary_level[1] / yart_salary_level[0] - 1) * 100, 2)
res['part_02']['year02_on_year03'] = round((yart_salary_level[2] / yart_salary_level[1] - 1) * 100, 2)
res['part_02']['industry_indicators'] = None
res['part_02']['industry_indicators_average'] = None
res['part_02']['industry_indicators_median'] = None
res['part_02']['median_comparison'] = None
return res
result['part_01'] = calculate_three_year_data()['part_01']
result['part_02'] = calculate_three_year_data()['part_02']
return result
def social_labor_disputes(param):
"""
劳动纠纷
Parameters:
param dict 数据库数据
Returns:
result dict 清洗好的数据
"""
data = param['社会问卷']
# Returns
result = dict()
for item in data['近三年公司数据']:
if item['劳资纠纷次数'] > 0:
result['describe'] = '劳动纠纷——公司在近三年出现了劳动纠纷,这影响了公司的社会得分。'
break
else:
result['describe'] = False
return result
def social_safety_accident(param):
"""
安全事故
Parameters:
param dict 数据库数据
Returns:
result dict 清洗好的数据
"""
data = param['社会问卷']
# Returns
result = dict()
for item in data['近三年公司数据']:
if item['安全事故发生次数'] > 0:
result['describe'] = '安全事故——公司在近三年出现了安全事故,这影响了公司的社会得分。。'
break
else:
result['describe'] = False
return result
def social_training_return(param):
"""
提供培训/产品返修、退还、投诉
Parameters:
param dict 数据库数据
Returns:
result dict 清洗好的数据
"""
# Params
three_year_data = sorted(param['社会问卷']['近三年公司数据'], key=operator.itemgetter('年份'), reverse=False)
# Returns
result = dict()
def date_range():
"""
评价周期年度区间
Parameters:
param dict 环境问卷
Returns:
result str 区间数据
"""
year_begin = three_year_data[0]['年份']
year_end = three_year_data[2]['年份']
res = year_begin + '-' + year_end
return res
def calculate_three_year_data():
"""
计算三年数据并转为字符串
Parameters:
'-'
Returns:
result str 三年数据用顿号隔开
"""
res = dict()
year_training = list()
year_return = list()
for item in three_year_data:
year_training.append(str(item['组织培训次数']))
year_return.append(str(item['返修、退回、投诉产品对应销售额占全部销售比例(%']))
# Returns-part1
res['part_01'] = dict()
res['part_01']['date_range'] = date_range()
res['part_01']['training_number'] = ''.join(year_training)
res['part_01']['year_03'] = three_year_data[2]['年份']
res['part_01']['industry_indicators'] = None
# part2
res['part_02'] = dict()
res['part_02']['date_range'] = date_range()
res['part_02']['salary_level'] = ''.join(year_return)
res['part_02']['year_03'] = three_year_data[2]['年份']
res['part_02']['industry_indicators'] = None
return res
result['part_01'] = calculate_three_year_data()['part_01']
result['part_02'] = calculate_three_year_data()['part_02']
return result
def socail_poverty_alleviation(param):
"""
扶贫和捐赠
Parameters:
param dict 数据库数据
Returns:
result dict 清洗好的数据
"""
data = param['社会问卷']
# Returns
result = dict()
for item in data['近三年公司数据']:
if item['扶贫+捐赠规模(万元)'] > 0:
result['describe'] = '扶贫和捐赠——公司在近三年有过扶贫、捐赠行为,这提高了公司社会得分。'
break
else:
result['describe'] = False
return result
def socail_legal_proceedings(param):
"""
司法风险
Parameters:
param dict 数据库数据
Returns:
result dict 清洗好的数据
"""
# Parameter
data = param['风险数据']
# Returns
result = dict()
for value in data.values():
if value > 0:
result['describe'] = '法律诉讼作为被告、被执行人、失信人、行政处罚、税收违法、严重违法、经营异常——经查公司存在1项或多项的法律诉讼作为被告、被执行人、失信人、行政处罚、税收违法、严重违法、经营异常情况这影响了公司社会分数。'
break
else:
result['describe'] = False
return result