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

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