tfse-app-api-v0.2/company/company_impl.py

import json
import pandas as pd


from company.company_db import FIND_DATA_IN_COMPANY, FIND_DATA_IN_MACRO_INDUSTRY_ANALYSIS, \
    FIND_SINGLE_DATA_BY_REQ_WITH_SORT, FIND_SINGLE_DATA_BY_REQ
from company.company_utils import judge_industry


def find_company_data_scripts(table, cid):
    return FIND_DATA_IN_COMPANY(table, {"企业ID": cid})


def get_head_info_impl(cid):
    # company_info: 企业信息
    company_info = FIND_SINGLE_DATA_BY_REQ("应用端", "企业用户", {"企业ID": cid}, ["企业名称", "已认证"])

    # cc_rating: 最近一期综信评价结果
    cc_rating = FIND_SINGLE_DATA_BY_REQ_WITH_SORT("综信评价数据", "评价结果", {"企业ID": cid}, ["信用评分", "信用等级", "评价时间"], {"评价年度": -1})
    # esg_rating: 最近一期ESG评价结果
    esg_rating = FIND_SINGLE_DATA_BY_REQ_WITH_SORT("ESG评价数据", "评价结果", {"企业ID": cid}, ["ESG得分", "评价等级", "评价时间"], {"评价年度": -1})

    # head_info: 公司主要头部信息
    head_info = {
        "企业名称": company_info['企业名称'],
        "企业认证": "已认证" if company_info['已认证'] == "是" else "未认证",
        "综合信用评价": cc_rating if cc_rating else {"信用等级": "N/A", "信用评分": "N/A", "评价时间": "N/A"},
        "ESG评价": esg_rating if esg_rating else {"信用等级": "N/A", "信用评分": "N/A", "评价时间": "N/A"}
    }

    return head_info


def find_financial_index_detail_scripts(table, cid):

    index_detail = FIND_DATA_IN_COMPANY(table, {"企业ID": cid})
    if not index_detail:
        return False

    df = pd.DataFrame(index_detail).sort_values('年报期', ascending=False)
    cols = [
        '年报期',
        '净资产收益率',
        '总资产报酬率',
        '总资产周转率',
        '应收账款周转率',
        '存货周转率',
        '资产负债率',
        '已获利息倍数',
        '速动比率',
        '营业增长率',
        '总资产增长率',
        '技术投入比率'
    ]
    unit_cols = [
        '净资产收益率',
        '总资产报酬率',
        '总资产周转率',
        '存货周转率',
        '资产负债率',
        '速动比率',
        '营业增长率',
        '总资产增长率',
        '技术投入比率'
    ]

    def percent_unit(param):
        return param.map(lambda x: '{}%'.format(x) if (param.name in unit_cols and not pd.isna(x)) else x)

    df_result = df[cols].apply(lambda x: percent_unit(x))

    return list(json.loads(df_result.T.to_json()).values())


def industry_analysis_impl(cid):
    """
    行业宏观分析数据
    Parameters:
        cid: str 企业ID
    Returns:
        Case Success:
            data: dict 行业宏观分析数据
        Case Fail:
            data: dict 空字典
    """

    industry = judge_industry(cid)

    if industry is False:
        return {}

    data = FIND_DATA_IN_MACRO_INDUSTRY_ANALYSIS(industry)

    if data is False:
        return {}

    return data


def esg_analysis_impl(cid):
    """
    ESG要素分析数据
    Parameters:
        cid: str 企业ID
    Returns:
        Case Success:
            data: dict ESG分析数据
        Case Fail:
            data: dict 空字典
    """
    data = FIND_DATA_IN_COMPANY('ESG要素分析',{"企业ID": cid})

    if data is False:
        return {}

    return data