import copy import time from rating.db import find_tyc_db, find_tfse_db from rating.static.province_map import p_map import pandas as pd def modify_unit(param): """ 金额单位从元改为万元 Parameters: param: 金额数据 Returns: result: 单位为万元的金额数据 """ money = round(param/10000, 2) return str(money) def judge_list(param): """ 判断列表中是否有None， 如果有置为- Parameters: param: list列表 Returns: result: 没有None的list """ df = pd.DataFrame(param) df = df.fillna('-') df = df[0].apply(lambda x: round(x/10000, 2) if isinstance(x, float) else x) df = df.to_list() df = [str(x) for x in df] return df def calculate_growth_rate(param1, param2): """ 计算增长率 传入两年财务数据字典项，算出年增长率 Parameters: param1: 财务指标1 param2: 财务指标2 Returns: 增长率 """ param1 = float(param1) param2 = float(param2) rate = round((param1 - param2) / param2, 2) return rate def calculate_amount(param): """ 计算数据之和 清洗数据库数据，根据需要返回 注意： 货币单位不同，现在存在美元、人民币两种，汇率暂用6.5 Parameters: param: 数据库数据 Returns: 数据合计 """ result = dict() result['amount'] = 0 result['industry'] = list() result['base'] = list() for item in param: if item['amountSuffix'] == '万人民币': try: result['amount'] += item['amount'] except TypeError: result['amount'] += 0 elif item['amountSuffix'] == '万美元': try: result['amount'] += item['amount'] * 6.5 except TypeError: result['amount'] += 0 result['industry'].append(item['category']) result['base'].append(item['base']) return result def calculate_list(param): """ 计算列表中所有元素出现的次数 Parameters: param: list数据 Returns: 列表元素出现次数对象 """ result = dict() for item in param: result[item] = param.count(item) return result def time_stamp(param): """ 时间戳转换 Parameters: param: 时间戳数据 Returns: 转换后时间数据 """ if param is None: return '-' else: estiblish_time = param / 1000 put_date = time.localtime(estiblish_time) date = time.strftime("%Y-%m-%d", put_date) return date def province_conversion(param): """ 省份转换 Parameters: param: 省份简称 Returns: 转换后省份全称 """ res = [v for k, v in p_map.items() if k == param][0] return res def industry_processing(param): """ 将行业数组转为字符串 Parameters: param list 行业数组 Returns: result str 行业字符串 """ return param[0] + '-' + param[1] # 获取数据库数据 def get_tfse_data(param): """ 获取数据库数据 Parameters: param: 企业名称、企业ID、评价ID Returns: report_data: 数据库数据 """ company = param['company'] rid = param['rid'] cid = param['cid'] report_data = dict() # 评价结果 es = find_tfse_db(db='评价', col='评价结果', query={'评价ID': rid})[0] # 综合评价填报 cef = find_tfse_db(db='评价', col='综合评价填报', query={'评价ID': rid})[0] # 指标明细 fid = find_tfse_db(db='企业', col='指标明细', query={'企业名称': company}) # 基本信息 cbd = find_tyc_db(db='公司背景', col='基本信息', query={'企业名称': company})[0] # 企业股东 cs = find_tyc_db(db='公司背景', col='企业股东', query={'企业名称': company})[0] # 对外投资 oi = find_tyc_db(db='公司背景', col='对外投资', query={'企业名称': company})[0] # 进出口信用 ip = find_tyc_db(db='经营状况', col='进出口信用', query={'企业名称': company})[0] # 专利 ip_02 = find_tyc_db(db='知识产权', col='专利', query={'企业名称': company})[0] # 软件著作权 sc = find_tyc_db(db='知识产权', col='软件著作权', query={'企业名称': company})[0] # 资质证书 qh = find_tyc_db(db='经营状况', col='资质证书', query={'企业名称': company})[0] # 融资历史 fh = find_tyc_db(db='公司发展', col='融资历史', query={'企业名称': company})[0] # 风险数据 rd = find_tfse_db(db='评价', col='风险数据', query={'企业ID': cid})[0] # 周边风险 pr = find_tyc_db(db='天眼风险', col='周边风险', query={'企业名称': company})[0] report_data['评价结果'] = es report_data['综合评价填报'] = cef report_data['指标明细'] = fid report_data['基本信息'] = cbd report_data['企业股东'] = cs report_data['对外投资'] = filter_oi_data(oi) report_data['进出口信用'] = ip report_data['专利'] = ip_02 report_data['软件著作权'] = sc report_data['资质证书'] = qh report_data['融资历史'] = fh report_data['风险数据'] = rd report_data['周边风险'] = pr return report_data def filter_oi_data(param): """ 对外投资 清洗掉状态为吊销、注销、迁出的企业 Parameters: param: 对外投资数据 Returns: 清洗后的数据 """ result = list() total = 0 data = param['对外投资']['result'] for item in data: if item['regStatus'] != '注销' and item['regStatus'] != '迁出' and item['regStatus'] != '吊销': result.append(item) total += 1 param['对外投资']['result'] = result param['对外投资']['total'] = total return param def shareholder_strength_data(param): """ 股东数据 清洗数据库数据，根据需要返回 Parameters: param: 企业名称 Returns: 股东数据 """ data = dict() # 基本信息 cbd = find_tyc_db(db='公司背景', col='基本信息', query={'企业名称': param})[0] # 企业股东 cs = find_tyc_db(db='公司背景', col='企业股东', query={'企业名称': param})[0] data['基本信息'] = cbd data['企业股东'] = cs return data def person_subsidiaries(param): """ 人员控股数据 清洗数据库数据，根据需要返回 Parameters: param: 人员id Returns: 人员控股数据 """ data = dict() # 控股数据 ps = find_tyc_db(db='公司背景', col='人员控股企业', query={'人员ID': param})[0] data['人员控股企业'] = ps return data def get_esg_data(param): """ 获取数据库数据并按要求格式返回 Parameters: param dict 企业名称、企业ID、评价ID Returns: result dict 符合要求需要清洗的数据 """ company = param['company'] rid = param['rid'] cid = param['cid'] # ESG评价结果 es = find_tfse_db(db='评价', col='ESG评价结果', query={'评价ID': rid})[0] # ESG风险数据 rd = find_tfse_db(db='评价', col='ESG风险数据', query={'企业ID': cid})[0] # ESG评价填报 ef = find_tfse_db(db='评价', col='ESG评价填报', query={'评价ID': rid})[0] rd.pop('企业ID') rd.pop('评价ID') rd.pop('更新日期') report_data = copy.deepcopy(ef) report_data['评价等级'] = es['评价等级'] report_data['风险数据'] = rd return report_data