272 lines
10 KiB
Python
272 lines
10 KiB
Python
import json
|
||
import requests
|
||
import pandas as pd
|
||
import numpy as np
|
||
|
||
from DBHelper.MongoHelper import MongoHelper
|
||
from Test.TestObj import CompanyUser, RatingRecord
|
||
|
||
|
||
class CompanyAuthImpl(CompanyUser):
|
||
"""企业相关类"""
|
||
db = MongoHelper("tfse_v0.21")
|
||
|
||
def company_register(self):
|
||
"""
|
||
新注册企业
|
||
1.调用发送注册邮箱，获取验证码
|
||
2.使用邮箱、密码、验证码进行注册
|
||
3.使用认证接口进行认证、初始化企业数据
|
||
"""
|
||
|
||
def send_mail():
|
||
"""获取验证码"""
|
||
url = "https://api.fecribd.com/app/test/verify_email"
|
||
secret = 'EZgo9ykxrYuBMYnYmmKIh'
|
||
headers = {'secret': secret, "Content-Type": "application/json;charset=UTF-8"}
|
||
param = {"email": self.dict_to_save(columns=['邮箱'])['邮箱']}
|
||
insert = json.dumps(param)
|
||
res = requests.post(url, headers=headers, data=insert)
|
||
result = json.loads(res.text)
|
||
if result['info'] == '邮件已发送':
|
||
code = self.db.find_single_data(
|
||
'应用端',
|
||
'邮箱验证码记录',
|
||
{'邮箱': self.email},
|
||
['验证码']
|
||
)
|
||
return code['验证码'] if code else False
|
||
else:
|
||
return '邮箱已被注册'
|
||
|
||
def register():
|
||
"""注册"""
|
||
email_code = send_mail()
|
||
if email_code and email_code != '邮箱已被注册':
|
||
data = dict()
|
||
data['email'] = self.dict_to_save(columns=['邮箱'])['邮箱']
|
||
data['code'] = email_code
|
||
url = "https://api.fecribd.com/app/test/register"
|
||
secret = 'EZgo9ykxrYuBMYnYmmKIh'
|
||
headers = {'secret': secret, "Content-Type": "application/json;charset=UTF-8"}
|
||
insert = json.dumps(data)
|
||
res = requests.post(url, headers=headers, data=insert)
|
||
result = json.loads(res.text)
|
||
if result['info'] == '注册成功':
|
||
cid = self.db.find_single_data(
|
||
'应用端',
|
||
'企业用户',
|
||
{'邮箱': self.email},
|
||
['企业ID']
|
||
)
|
||
return cid['企业ID'] if cid else False
|
||
elif email_code == '邮箱已被注册':
|
||
return {"info": "邮箱已被注册"}, 210
|
||
else:
|
||
return {"info": "验证码发送失败"}, 210
|
||
|
||
return register()
|
||
|
||
def enterprise_certification(self):
|
||
"""企业认证（测试用）"""
|
||
insert_data = self.dict_to_save(columns=['认证信息'])
|
||
self.db.update_single_data(
|
||
"应用端",
|
||
"企业用户",
|
||
{"企业ID": self.cid},
|
||
insert_data
|
||
)
|
||
|
||
def certification_interface():
|
||
"""调用应用端认证接口"""
|
||
url = "https://api.fecribd.com/app/test/company_verify?cid={}".format(self.cid)
|
||
secret = 'EZgo9ykxrYuBMYnYmmKIh'
|
||
headers = {'secret': secret, "Content-Type": "application/json;charset=UTF-8"}
|
||
res = requests.get(url, headers=headers)
|
||
return json.loads(res.text)
|
||
|
||
return certification_interface()
|
||
|
||
def change_verified(self):
|
||
"""企业认证状态修改"""
|
||
verify_info = self.VerifyInfo()
|
||
verify_info.name = None
|
||
verify_info.code = None
|
||
verify_info.legal_person = None
|
||
insert_data = verify_info.dict_to_save()
|
||
self.db.update_single_data(
|
||
"应用端",
|
||
"企业用户",
|
||
{"企业ID": self.cid},
|
||
{"已认证": "否", "认证信息": insert_data}
|
||
)
|
||
|
||
|
||
class CompanyIndexImpl(RatingRecord):
|
||
"""企业主页"""
|
||
db = MongoHelper("tfse_v0.21")
|
||
|
||
def generate_homepage(self):
|
||
"""生成主页数据"""
|
||
self.cid = self.db.find_single_data(
|
||
'企业数据',
|
||
'评价记录',
|
||
{"评价ID": self.rid},
|
||
['企业ID']
|
||
)['企业ID']
|
||
data = self.dict_to_save(columns=['企业ID', '评价项目'])
|
||
|
||
def generate_interface():
|
||
"""调用打分接口"""
|
||
url = "https://api.fecribd.com/app/test/general_rating"
|
||
secret = 'EZgo9ykxrYuBMYnYmmKIh'
|
||
headers = {'secret': secret, "Content-Type": "application/json;charset=UTF-8"}
|
||
insert = json.dumps(data)
|
||
res = requests.post(url, headers=headers, data=insert)
|
||
return json.loads(res.text)
|
||
|
||
return generate_interface()
|
||
|
||
|
||
class DeleteCompanyDataImpl(RatingRecord):
|
||
"""删除企业相关数据"""
|
||
db = MongoHelper("tfse_v0.21")
|
||
|
||
def form_rid_delete(self, record, data):
|
||
"""根据rid删除相关记录"""
|
||
|
||
def delete_all_records():
|
||
"""根据rid删除评价数据库相匹配数据"""
|
||
if '财指结果' in data:
|
||
database = '综信评价数据'
|
||
database_01 = '综信报告'
|
||
database_02 = '综信证书'
|
||
else:
|
||
database = 'ESG评价数据'
|
||
database_01 = 'ESG报告'
|
||
database_02 = 'ESG证书'
|
||
for name in data:
|
||
if name == '财指结果':
|
||
for i in range(3):
|
||
self.db.delete_single_data(database, name, {"企业ID": self.rid})
|
||
else:
|
||
self.db.delete_single_data(database, name, {"企业ID": self.rid})
|
||
# 文件相关
|
||
report_fid = records['报告fid']
|
||
certificate_fid = records['证书fid']
|
||
self.db.delete_file('文件', database_01, report_fid)
|
||
self.db.delete_file('文件', database_02, certificate_fid)
|
||
return '删除成功'
|
||
|
||
def delete_part_records():
|
||
"""根据rid删除评价数据库相匹配数据"""
|
||
change = {'进行状态': '进行', '评价结果': None, '开始填报时间': None, '提交填报时间': None, '评价完成时间': None,
|
||
'评价完成日期': None, '报告fid': None, '证书fid': None}
|
||
self.db.update_single_data(
|
||
'企业数据',
|
||
'评价记录',
|
||
{'评价ID': self.rid},
|
||
change
|
||
)
|
||
return '状态更改成功'
|
||
|
||
records = self.db.find_single_data(
|
||
'企业数据',
|
||
'评价记录',
|
||
{"评价ID": self.rid},
|
||
['进行状态', '报告fid', '证书fid']
|
||
)
|
||
if records:
|
||
if records['进行状态'] == '完成':
|
||
if record == '删除记录':
|
||
info = delete_all_records()
|
||
else:
|
||
info = delete_all_records()
|
||
if info == '删除成功':
|
||
info = delete_part_records()
|
||
return info
|
||
return info
|
||
else:
|
||
return '不能删除正在进行的评价记录'
|
||
|
||
else:
|
||
return '未查询到此rid相关记录'
|
||
|
||
def form_cid_delete(self, data):
|
||
"""根据cid删除相关记录"""
|
||
for name in data:
|
||
self.db.delete_single_data('企业数据', name, {"企业ID": self.cid})
|
||
return '企业相关数据删除成功'
|
||
|
||
|
||
class GenerateFillInData(RatingRecord):
|
||
"""生成填报数据"""
|
||
db = MongoHelper('tfse_v0.21')
|
||
|
||
def insert_esg_data(self):
|
||
"""自动生成esg问卷填报数据"""
|
||
def search_data():
|
||
"""根据rid查询填报数据"""
|
||
fill = self.db.find_single_data(
|
||
'ESG评价数据',
|
||
'填报数据',
|
||
{'评价ID': self.rid},
|
||
[]
|
||
)
|
||
return fill
|
||
|
||
data = search_data()
|
||
|
||
def e_script():
|
||
"""环境问卷"""
|
||
e_data = data['环境问卷']['近三年公司数据']
|
||
name = list()
|
||
for k in e_data[0].keys():
|
||
if k != '年份':
|
||
name.append(k)
|
||
year = [e_data[0]['年份'], e_data[1]['年份'], e_data[2]['年份']]
|
||
df = pd.DataFrame(np.random.randint(200, 500, size=(3, 13)), columns=name)
|
||
df.insert(loc=0, column='年份', value=year)
|
||
num = df.shape[0]
|
||
result = list()
|
||
for n in range(num):
|
||
result.append(df.iloc[n].to_dict())
|
||
return result
|
||
|
||
def s_script():
|
||
"""社会问卷"""
|
||
s_data = data['社会问卷']['近三年公司数据']
|
||
year = [s_data[0]['年份'], s_data[1]['年份'], s_data[2]['年份']]
|
||
df = pd.DataFrame(year, columns=['年份'])
|
||
df['员工总数（年底）'] = np.random.randint(150, 250, size=3)
|
||
df['当年离职人数'] = np.random.randint(10, 25, size=3)
|
||
df['人均薪酬水平（元/月）'] = np.random.randint(7500, 8500, size=3)
|
||
df['劳动合同要求工作时长（每周小时数）'] = np.random.randint(40, 50, size=3)
|
||
df['劳资纠纷次数'] = np.random.randint(0, 5, size=3)
|
||
df['安全事故发生次数'] = [0, 0, 0]
|
||
df['组织培训次数'] = np.random.randint(5, 15, size=3)
|
||
df['公司从前3大供应商拿货占全部供应商比例（%）'] = [round(i * 100, 2) for i in np.random.rand(3)]
|
||
df['公司前3大客户销售额占全部销售比例（%）'] = [round(i * 100, 2) for i in np.random.rand(3)]
|
||
df['返修、退回、投诉产品对应销售额占全部销售比例（%）'] = [round(i * 100, 2) for i in np.random.rand(3)]
|
||
df['扶贫+捐赠规模（万元）'] = np.random.randint(100, 500, size=3)
|
||
num = df.shape[0]
|
||
result = list()
|
||
for n in range(num):
|
||
result.append(df.iloc[n].to_dict())
|
||
return result
|
||
|
||
def save_data():
|
||
"""保存数据"""
|
||
self.db.update_single_data(
|
||
'ESG评价数据',
|
||
'填报数据',
|
||
{'评价ID': self.rid},
|
||
{
|
||
'环境问卷.近三年公司数据': e_script(),
|
||
'社会问卷.近三年公司数据': s_script()
|
||
}
|
||
)
|
||
return '生成成功'
|
||
|
||
return save_data()
|