import json import math from datetime import datetime, timedelta 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()} business_type = {row["问题"]: row["类型"] for _, row in business_data.iterrows()} for k, v in business_dict.items(): if pd.isna(v): business_dict[k] = None if business_dict[k] is None and business_type[k] == "文本": business_dict[k] = "--" # 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}) if not 工商信息: raise HTTPException(status_code=404, detail="无公司数据") 产品信息, _ = 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 - 2000}", "月份": f"{datetime.now().month}", "报告编号": "中小商协[2023]0111号", "信用级别": rate_res.data.评级结果.级别, "证书编码": f"{datetime.now.strftime('%y%m%d')}{str(rate_id).rjust(6, '0')}", "有效日期": f"{item.rate_time.strftime('%Y%m%d')}至{(datetime.fromtimestamp(item.rate_time.timestamp()).replace(year=item.rate_time.year + 3) - timedelta(days=1)).strftime('%Y%m%d')}", "通知书日期": item.rate_time.strftime('%Y%m%d'), "申明日期": item.rate_time.strftime('%Y%m%d'), "所属国民经济行业": 工商信息['industry'], "企业中文名称": company_name, "英文名称": 工商信息["property3"], "工商注册号": 工商信息['regNumber'], "组织机构代码": 工商信息['orgNumber'], "统一社会信用代码": 工商信息['creditCode'], "注册资本": 工商信息['regCapital'], "企业类型": 工商信息['companyOrgType'], "注册地址": 工商信息['regLocation'], "成立日期": 工商信息['estiblishTime'], "核准日期": 工商信息['approvedTime'], "营业期限终止日期": "-", "法定代表人": 工商信息['legalPersonName'] if 工商信息 else '--', "经营地址": 工商信息['regLocation'], "邮政编码": 工商信息['reportList'][0]['postcode'] if 工商信息 and 工商信息['reportList'] else '--', "联系人": man_data['联系人'], "联系电话": 工商信息["phoneNumber"] if 工商信息 else '--', "传真号码": 工商信息['reportList'][0]['phoneNumber'] if 工商信息 and 工商信息['reportList'] else '--', "电子邮箱": 工商信息['email'] if 工商信息['reportList'] else '--', "企业网址": 工商信息["website"], "经营范围": 工商信息['businessScope'], "主营业务": man_data['主营业务'], "主要产品": "、".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\t\t".join([f"{item['name']}认缴出资{item['amount']}，出资比例{item['percent']}" for item in 股权结构图['structure']['children']]) if 股权结构图 else '--', "销售人数": man_data["销售人员人数"], "销售占比": str(int(man_data["销售人员人数"] / man_data["年末员工总数"] * 100))+'%', "技术服务人数": "0", "技术服务占比": "0%", "生产人数": man_data.get("生产人数","--"), "生产占比": str(int(man_data.get("生产人数",0) / 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["年末员工总数"]), "合计占比": "100%", "高管构成": "--", "人民币开户银行名称": 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