wd-smebiz/mods/smebiz_rate/company_rate/rate_utils/utils.py

513 lines
22 KiB
Python
Raw 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 json
import math
from datetime import datetime
from typing import List
from fastapi import HTTPException
from sqlalchemy.orm import Session
from context.common import tianyancha_api
from mods.smebiz_rate.company_rate.models import CompanyRate
from mods.smebiz_rate.company_rate.rate_utils.config import 合同纠纷案由, 劳动仲裁案由
import pandas as pd
import requests as rq
from context.common import conf
from utils.data_utils import JsDict
class RateRes:
message: str
class data:
class 指标数值与得分:
指标: str
数值: float
单位: str
权重: float
得分: float
class 评级结果:
级别: str
总分: float
指标数值与得分: List[指标数值与得分]
def nan_to_none(data):
if pd.isna(data):
return None
else:
return data
def load_question_excel(excel_path):
# Load “经营问卷” data from Excel
business_data = pd.read_excel(excel_path, sheet_name="经营问卷")
business_data = business_data.where(pd.notna(business_data), None)
business_dict = {row["问题"]: row["填写(文本类无则不填写)"] for _, row in business_data.iterrows()}
for k, v in business_dict.items():
if pd.isna(v):
business_dict[k] = None
# Load “财务问卷” data from Excel
finance_data = pd.read_excel(excel_path, sheet_name="财务问卷")
finance_data = finance_data.where(pd.notna(finance_data), None)
# Convert multi-year data into a nested dictionary format
finance_report_dict = {
"2020年": {},
"2021年": {},
"2022年": {},
}
report = None
for i, item in enumerate(finance_data['报表']):
if item in ["单位", "是否审计", "会计事务所"]:
if item in ["单位"]:
continue
if item in ["是否审计", "会计事务所"]:
finance_report_dict[item] = nan_to_none(finance_data['科目'][i])
continue
if item is not None:
report = item
finance_report_dict[report] = {}
subject = finance_data['科目'][i]
# finance_report_dict[subject] = {
# # finance_report_dict[report][subject] = {
# "2020年": nan_to_none(finance_data['2020年'][i]),
# "2021年": nan_to_none(finance_data['2021年'][i]),
# "2022年": nan_to_none(finance_data['2022年'][i]),
# }
if subject:
finance_report_dict["2020年"][subject] = nan_to_none(finance_data['2020年'][i])
finance_report_dict["2021年"][subject] = nan_to_none(finance_data['2021年'][i])
finance_report_dict["2022年"][subject] = nan_to_none(finance_data['2022年'][i])
if pd.isna(subject):
continue
data_dict = {
"经营问卷": business_dict,
"财务问卷": finance_report_dict
}
return data_dict
def filter_bg(data: [], company_name):
new_data = []
for item in data:
add = False
for i in item['defendant']:
if company_name in i['name']:
add = True
if add:
new_data.append(item)
return new_data
def load_api_data(company_name):
司法风险, _ = tianyancha_api.get("司法风险", {"keyword": company_name})
# 资质证书, 资质证书数量 = tianyancha_api.get("资质证书", {"keyword": company_name})
# 历史经营异常数量 = tianyancha_api.get_total("历史经营异常", {"keyword": company_name})
双随机抽查, 双随机抽查数量 = tianyancha_api.get("双随机抽查", {"keyword": company_name})
双随机抽查详情 = [tianyancha_api.get("双随机抽查详情", {"businessId": item['taskList'][0]['businessId']})[0] for
item in 双随机抽查]
质押比例, _ = tianyancha_api.get("质押比例", {"keyword": company_name})
# 破产重整数量 = tianyancha_api.get_total("破产重整", {"keyword": company_name})
# 司法拍卖数量 = tianyancha_api.get_total("司法拍卖", {"keyword": company_name})
企业基本信息, _ = tianyancha_api.get("企业基本信息", {"keyword": company_name})
# 企业专利信息数量 = tianyancha_api.get_total("企业专利信息", {"keyword": company_name})
# 知识产权出质, _ = tianyancha_api.get("知识产权出质", {"keyword": company_name})
税务评级, _ = tianyancha_api.get("税务评级", {"keyword": company_name})
进出口信用, _ = tianyancha_api.get("进出口信用", {"keyword": company_name})
税收违法数量 = tianyancha_api.get_total("税收违法", {"keyword": company_name})
欠税公告数量 = tianyancha_api.get_total("欠税公告", {"keyword": company_name})
# 行政许可数量 = tianyancha_api.get_total("行政许可", {"keyword": company_name})
行政处罚数量 = tianyancha_api.get_total("行政处罚", {"keyword": company_name})
经营异常数量 = tianyancha_api.get_total("经营异常", {"keyword": company_name})
严重违法数量 = tianyancha_api.get_total("严重违法", {"keyword": company_name})
# 终本案件数量 = tianyancha_api.get_total("终本案件", {"keyword": company_name})
失信被执行人数量 = tianyancha_api.get_total("失信被执行人", {"keyword": company_name})
被执行人数量 = tianyancha_api.get_total("被执行人", {"keyword": company_name})
限制消费令数量 = tianyancha_api.get_total("限制消费令", {"keyword": company_name})
股权变更, _ = tianyancha_api.get("股权变更", {"keyword": company_name})
新闻舆情, _ = tianyancha_api.get("新闻舆情", {"name": company_name}, page=1)
行政处罚, _ = tianyancha_api.get("行政处罚", {"keyword": company_name})
成立年限 = datetime.now().year - datetime.fromtimestamp(
企业基本信息['estiblishTime'] / 1000).year if 企业基本信息 else 0
企业类型 = '有限责任公司' if 企业基本信息 and '有限责任' in 企业基本信息['companyOrgType'].split("(")[
0] else '股份制公司'
开庭公告_劳动争议 = len(
[item["caseReason"] for item in filter_bg(司法风险['ktAnnouncementList'], company_name) if
item['caseReason'] in 劳动仲裁案由]) if 司法风险 else 0
开庭公告_被告_合同纠纷 = len(
[item["caseReason"] for item in 司法风险['ktAnnouncementList'] if
item["caseReason"] and item["caseReason"] in 合同纠纷案由]) if 司法风险 else 0
纳税信用等级 = 税务评级[0]['grade'] if 税务评级 else "M"
双随机抽查结果 = '合格'
for item in 双随机抽查详情:
for child in item:
if child['checkResult'] != '未发现问题':
双随机抽查结果 = '批评和处罚'
break
now = datetime.now()
近三年股东变更次数 = len([item for item in 股权变更 if
datetime.fromtimestamp(item['change_time'] / 1000) > now.replace(year=now.year - 3)])
最近10条企业舆情负面占比 = len([item for item in 新闻舆情 if item['emotion'] == -1]) / 10 * 100
行政处罚_警告 = len([item for item in 行政处罚 if (item['type'] in ["警告", "通报批评", "罚款"] or any(
map(lambda s: s in item['content'], ["警告", "通报批评", "罚款"]))) and '没收' not in item['content']])
行政处罚_没收 = len(
[item for item in 行政处罚 if item['type'] in ["没收"] or any(map(lambda s: s in item['content'], ["没收"]))])
data = {
"成立年限": 成立年限,
"企业类型": 企业类型,
# "股东性质": None,
"行政处罚(警告、通报批评、罚款)": 行政处罚_警告,
"行政处罚(没收违法所得、没收非法财务...": 行政处罚_没收,
"近三年股东变更次数": 近三年股东变更次数,
"最近10条企业舆情负面占比": 最近10条企业舆情负面占比,
"开庭公告(被告-合同纠纷、劳动争议)": 开庭公告_劳动争议 + 开庭公告_被告_合同纠纷,
"纳税信用等级": 纳税信用等级,
"双随机抽查结果": 双随机抽查结果,
"经营异常": 经营异常数量,
"欠税公告": 欠税公告数量,
"被执行人": 被执行人数量,
"失信被执行人": 失信被执行人数量,
"税收违法": 税收违法数量,
"严重违法": 严重违法数量
}
return data
def load_report_need_data(db: Session, rate_id):
item: CompanyRate = db.query(CompanyRate).filter(CompanyRate.id == rate_id).first()
if not item.rate_level:
raise HTTPException(status_code=303, detail="未评级,无法下载报告")
man_data = json.loads(item.man_data)['经营问卷']
财务问卷 = json.loads(item.man_data)['财务问卷']
rate_res: RateRes = JsDict(json.loads(item.rate_data))
company_name = item.company_name
工商信息, _ = tianyancha_api.get("工商信息", {"keyword": company_name})
产品信息, _ = tianyancha_api.get("产品信息", {"keyword": company_name})
资质证书, _ = tianyancha_api.get("资质证书", {"name": company_name})
股权结构图, _ = tianyancha_api.get("股权结构图", {"keyword": company_name})
知识产权, _ = tianyancha_api.get("知识产权", {"keyword": company_name})
供应商, _ = tianyancha_api.get("供应商", {"keyword": company_name})
企业招投标信息, _ = tianyancha_api.get("企业招投标信息", {"keyword": company_name}, page=1)
# data = {
# "企业中文名称": company_name,
# "所属国民经济行业": 工商信息['industry'],
# "英文名称": 工商信息['property3'],
# "工商注册号": 工商信息['regNumber'],
# "组织机构代码": 工商信息['orgNumber'],
# "统一社会信用代码": 工商信息['creditCode'],
# "注册资本": 工商信息['regCapital'],
# "企业类型": 工商信息['companyOrgType'],
# "注册地址": 工商信息['regLocation'],
# "成立日期": 工商信息['estiblishTime'],
# "核准日期": 工商信息['approvedTime'],
# "营业期限终止日期": '',
# "法定代表人": 工商信息['punishList'][0]['legalPersonName'] if 工商信息['punishList'] else '',
# "经营地址": '',
# "邮政编码": 工商信息['reportList'][0]['postcode'] if 工商信息['reportList'] else '',
# "联系电话": 工商信息['reportList'][0]['phoneNumber'] if 工商信息['reportList'] else '',
# "传真号码": 工商信息['reportList'][0]['phoneNumber'] if 工商信息['reportList'] else '',
# "电子邮箱": 工商信息['reportList'][0]['email'] if 工商信息['reportList'] else '',
# "企业网址": '',
# "经营范围": 工商信息['businessScope'],
# "主要产品": '',
# "历史沿革": '',
# "股权结构": '',
# "高管构成": '',
# "企业资质": '',
# "企业商标": '',
# "企业专利": '',
# "企业软件著作权": '',
# "企业供应商": '',
# "企业招投标情况": ''
# }
data = {
"企业名称": company_name,
"年份": f"{datetime.now().year}",
"月份": f"{datetime.now().month}",
"报告编号": "",
"信用级别": rate_res.data.评级结果.级别,
"证书编码": "",
"有效日期": "",
"通知书日期": "",
"申明日期": "",
"所属国民经济行业": 工商信息['industry'],
"企业中文名称": company_name,
"英文名称": 工商信息["property3"],
"工商注册号": 工商信息['regNumber'],
"组织机构代码": 工商信息['orgNumber'],
"统一社会信用代码": 工商信息['creditCode'],
"注册资本": 工商信息['regCapital'],
"企业类型": 工商信息['companyOrgType'],
"注册地址": 工商信息['regLocation'],
"成立日期": 工商信息['estiblishTime'],
"核准日期": 工商信息['approvedTime'],
"营业期限终止日期": "-",
"法定代表人": 工商信息['legalPersonName'] if 工商信息 else '--',
"经营地址": 工商信息['regLocation'],
"邮政编码": 工商信息['reportList'][0]['postcode'] if 工商信息 and 工商信息['reportList'] else '--',
"联系人": 工商信息['reportList'][0]['operatorName'] if 工商信息 and 工商信息['reportList'] else '--',
"联系电话": 工商信息["phoneNumber"] if 工商信息 else '--',
"传真号码": 工商信息['reportList'][0]['phoneNumber'] if 工商信息 and 工商信息['reportList'] else '--',
"电子邮箱": 工商信息['email'] if 工商信息['reportList'] else '--',
"企业网址": 工商信息["website"],
"经营范围": 工商信息['businessScope'],
"主营业务": "",
"主要产品": "".join([item['filterName'] for item in 产品信息]),
"历史沿革": (("\n".join(
["\t" + datetime.strptime(item['changeTime'], '%Y-%m-%d').strftime('%Y年%m月%d') + '\n\t\t' + item[
'changeItem'] + f':由"{item["contentBefore"]}"变更为"{item["contentAfter"]}"' for item in
工商信息['changeList'] if
item['changeItem'] and item['contentAfter'] and item['contentBefore']])) + " ")[1:] if 工商信息 else "--",
"股权结构": "\n".join([f"{item['name']}认缴出资{item['amount']},出资比例{item['percent']}" for item in
股权结构图['structure']['children']]) if 股权结构图 else '--',
"销售人数": str(int(man_data["销售人员人数"] / man_data["年末员工总数"] * 100)),
"销售占比": "--",
"技术服务人数": "--",
"技术服务占比": "--",
"生产人数": "--",
"生产占比": "--",
"管职人数": str(man_data["管理人员人数"] + man_data["职能人员人数"]),
"管职占比": str(
int((man_data["管理人员人数"] + man_data["职能人员人数"]) / man_data["年末员工总数"] * 100)) + '%',
"本科及以上人数": str(man_data["本科及以上人员人数"]),
"本科及以上占比": str(int(man_data["本科及以上人员人数"] / man_data["年末员工总数"] * 100)) + '%',
"专科人数": man_data["专科及以下人员人数"],
"专科占比": str(int(man_data["专科及以下人员人数"] / man_data["年末员工总数"] * 100)) + '%',
"合计人数": str(man_data["年末员工总数"]),
"合计占比": "--",
"高管构成": "--",
"人民币开户银行名称": man_data["人民币开户银行名称"],
"人民币开户银行账号": man_data["人民币开户银行账号"],
"外币开户银行名称": man_data["外币开户银行名称"],
"外币开户银行账号": man_data["外币开户银行账号"],
"贷款卡编号": man_data["贷款卡编号"],
"经营场所建筑面积": man_data["经营场所建筑面积"],
"经营场所位置": man_data["经营场所位置"],
"经营场所权属关系": man_data["经营场所权属关系"],
"工商信用记录": man_data["工商信用记录"],
"海关信用记录": man_data["海关信用记录"],
"税务信用记录": "--",
"银行信用记录": man_data["银行信用记录"],
"法院信用记录": man_data["法院信用记录"],
"生产安全信息": man_data["生产安全信息"],
"社会责任实施": man_data["社会责任实施"],
"企业资质": "".join({item['certificateName'] for item in 资质证书}),
"商标信息": [
{"申请日期": item["appDate"], "商标": item["tmPic"], "商标名称": item["tmName"], "注册号": item["regNo"],
"类别": item["intCls"], "流程状态": item["status"]} for item in 知识产权['tmList']] if 知识产权 else [],
"专利信息": [
{
"申请日": item["applicationTime"],
"专利名称": item["patentName"],
"专利类型": item["patentType"],
"专利状态": item["lawStatus"],
"申请号": item["appNumber"],
"公开号": item["applicationPublishNum"],
"公开日": item["applicationPublishTime"],
"发明人": item["inventor"]
} for item in 知识产权['patentList']] if 知识产权 else [],
"软件著作权": [
{
"批准日期": item["regTime"],
"软件全称": item["fullName"],
"软件简称": item["simpleName"],
"登记号": item["regNum"],
"分类号": item["catNum"],
"版本号": item["version"],
} for item in 知识产权['copyRegList']] if 知识产权 else [],
"主要供应商情况": [
{
"供应商名称": item["supplier_name"],
"主工采购产品种类": "--",
"年平均采购额": item["amt"],
} for item in 供应商['pageBean']['result']
] if 供应商 else [],
"招投标情况": [
{
"发布时间": item["publishTime"],
"标题": item["title"],
"采购人": item["purchaser"],
} for item in 企业招投标信息
] if 企业招投标信息 else [],
"主要财务数据": {
"2022年": 财务问卷['2022年'],
"2021年": 财务问卷['2021年'],
"2020年": 财务问卷['2020年'],
}
}
for key, v in data.items():
if v is None:
data[key] = "--"
return data
class 商标信息:
申请日期: str # appDate
商标: str # tmPic
商标名称: str # tmName
注册号: str # regNo
类别: str # intCls
流程状态: str # status
class 专利信息:
申请日: str
专利名称: str
专利类型: str
专利状态: str
申请号: str
公开号: str
公开日: str
发明人: str
class 软件著作权:
批准日期: str
软件全称: str
软件简称: str
登记号: str
分类号: str
版本号: str
class 主要供应商情况:
供应商名称: str
主工采购产品种类: str
年平均采购额: str
class 招投标情况:
发布时间: str
标题: str
采购人: str
def pre_process_data(经营问卷, 财务问卷, 背调接口):
"""
问卷数据清洗为打分接口所需数据
"""
经营问卷['对外服务总次数'] = 经营问卷["对外服务总次数(包括电话咨询、客户支持、产品售后等)"]
背调接口["开庭公告被告合同纠纷劳动争议"] = 背调接口["开庭公告(被告-合同纠纷、劳动争议)"]
背调接口["行政处罚没收违法所得没收非法财务"] = 背调接口["行政处罚(没收违法所得、没收非法财务..."]
背调接口["行政处罚警告通报批评罚款"] = 背调接口["行政处罚(警告、通报批评、罚款)"]
data = {
"经营问卷": 经营问卷,
"财务问卷": {
"是否审计": None,
"会计事务所": None
},
"背调接口": 背调接口
}
for key in ["2022年", "2021年", "2020年"]:
for item in 财务问卷[key]:
if not data['财务问卷'].get(item):
data['财务问卷'][item] = []
data['财务问卷'][item].append(财务问卷[key][item])
data['财务问卷']['累计折旧'] = data['财务问卷']['减:累计折旧']
data['财务问卷']['是否审计'] = 财务问卷['是否审计']
data['财务问卷']['会计事务所'] = 财务问卷['会计事务所']
return data
def sme_rate_interface(data):
"""
打分
"""
res = rq.post(f'{conf["rate_utils"]["host"]}/api/model_score/model_score_reusult', json.dumps(
data), timeout=8)
data = res.json()
radar_data = make_radar_data(data)
data['radar_data'] = radar_data
# raise HTTPException(status_code=303, detail="评级接口错误")
if data and data['message'] == '计算成功':
return data
else:
print(data)
raise HTTPException(status_code=303, detail="评级接口错误")
def make_radar_data(rate_data):
radar = {
"财务状况": [
"营业收入",
"净利率",
"资产负债率",
"流动比率",
"速动比率",
"EBIT利息保障倍数",
"总资产周转次数",
"资产总额",
"股东支持力度"
],
"经营能力": [
"经营异常",
"双随机抽查结果",
"行政处罚(警告、通报批评、罚款)",
"负面舆情占比",
"高管平均从业年限",
"经营年限"
],
"内部治理": [
"管理制度文件类别数",
"管理人员比例",
"近三年股东变更次数",
"股东性质",
"董事会情况",
"监事会情况",
"设置有财务管理制度",
"财务报表审计状态"
],
"社会责任": [
"因公伤亡人次",
"发生过客户隐私泄露",
"产品退回率",
"员工离职率",
"员工每周工作时长",
"培训人数占比",
"服务投诉率",
"投诉响应时间",
"投诉解决率",
],
"守法守规": [
"被执行人",
"失信被执行人",
"纳税信用等级",
"欠税公告",
"税收违法",
"行政处罚(没收违法所得、没收非法财务...",
"严重违法",
"开庭公告(被告-合同纠纷、劳动争议)"
]
}
index_dic = {item['指标']: {"scope": item['得分'], "weight": item['权重']} for item in
rate_data["data"]["指标数值与得分"]}
radar_data = []
for key, items in radar.items():
scope = sum([index_dic[item]['scope'] for item in items])
weight = sum([index_dic[item]['weight'] for item in items])
radar_data.append({
"label": key,
"value": round(scope / weight * 100, 2)
})
print("radar_data", radar_data)
return radar_data