import copy import time from RatingData.db import FIND_DATA_N_TYC, FIND_SINGLE_DATA_BY_REQ, FIND_ALL_DATA_BY_REQ from RatingData.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_SINGLE_DATA_BY_REQ('综信评价数据', '评价结果', {'评价ID': rid}, ["企业名称", "评价年度", "财务评分", "经营评分", "信用等级", "评价时间"]) # 综合评价填报 cef = FIND_SINGLE_DATA_BY_REQ('综信评价数据', '填报数据', {'评价ID': rid}, ["财务填报", "行业选择"]) # 指标明细 fid = FIND_ALL_DATA_BY_REQ('综信评价数据', '财指结果', {'企业ID': cid}, ['报告期', '净资产收益率', '存货周转率', '已获利息倍数', '应收账款周转率', '总资产周转率', '总资产增长率', '总资产报酬率', '技术投入比率', '营业增长率', '资产负债率', '速动比率']) # 基本信息 cbd = FIND_DATA_N_TYC(db='公司背景', col='基本信息', query={'企业名称': company})[0] if '上市' in cbd['基本信息']['companyOrgType'] and '非上市' not in cbd['基本信息']['companyOrgType']: # 企业股东 cs = FIND_DATA_N_TYC(db='公司背景', col='十大股东', query={'企业名称': company})[0] else: cs = FIND_DATA_N_TYC(db='公司背景', col='企业股东', query={'企业名称': company})[0] # 对外投资 oi = FIND_DATA_N_TYC(db='公司背景', col='对外投资', query={'企业名称': company})[0] # 进出口信用 ip = FIND_DATA_N_TYC(db='经营状况', col='进出口信用', query={'企业名称': company})[0] # 专利 ip_02 = FIND_DATA_N_TYC(db='知识产权', col='专利', query={'企业名称': company})[0] # 软件著作权 sc = FIND_DATA_N_TYC(db='知识产权', col='软件著作权', query={'企业名称': company})[0] # 资质证书 qh = FIND_DATA_N_TYC(db='经营状况', col='资质证书', query={'企业名称': company})[0] # 融资历史 fh = FIND_DATA_N_TYC(db='公司发展', col='融资历史', query={'企业名称': company})[0] # 风险数据 rd = FIND_SINGLE_DATA_BY_REQ('综信评价数据', '风险数据', {'评价ID': rid}, ["合规风险", "经营风险", "关联风险"]) # 周边风险 pr = FIND_DATA_N_TYC(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_DATA_N_TYC(db='公司背景', col='基本信息', query={'企业名称': param})[0] # 企业股东 cs = FIND_DATA_N_TYC(db='公司背景', col='企业股东', query={'企业名称': param})[0] data['基本信息'] = cbd data['企业股东'] = cs return data def ten_shareholder_data(param): """ 十大股东数据 清洗数据库数据，根据需要返回 Parameters: param: 企业名称 Returns: 股东数据 """ data = dict() # 基本信息 cbd = FIND_DATA_N_TYC(db='公司背景', col='基本信息', query={'企业名称': param})[0] data['基本信息'] = cbd if '上市' in cbd['基本信息']['companyOrgType']: cs = FIND_DATA_N_TYC(db='公司背景', col='十大股东', query={'企业名称': param})[0] data['十大股东'] = cs else: cs = FIND_DATA_N_TYC(db='公司背景', col='企业股东', query={'企业名称': param})[0] data['企业股东'] = cs return data def person_subsidiaries(param): """ 人员控股数据 清洗数据库数据，根据需要返回 Parameters: param: 人员id Returns: 人员控股数据 """ data = dict() # 控股数据 ps = FIND_DATA_N_TYC(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_SINGLE_DATA_BY_REQ('ESG评价数据', '评价结果', {'评价ID': rid}, ['评价等级']) # ESG风险数据 rd = FIND_SINGLE_DATA_BY_REQ('ESG评价数据', '风险数据', {'企业ID': cid}, ["严重违法", "失信人", "法律诉讼", "税收违法", "经营异常", "行政处罚", "被执行人"]) # ESG评价填报 ef = FIND_SINGLE_DATA_BY_REQ('ESG评价数据', '填报数据', {'评价ID': rid}, ["评价ID", "企业ID", "企业名称", "所属行业", "评价年度", "公司当年收入（万元）", "环境问卷", "社会问卷", "治理问卷"]) report_data = copy.deepcopy(ef) report_data['评价等级'] = es['评价等级'] report_data['风险数据'] = rd return report_data