tfse-admin-api-v0.2/Manage/Company/CompanyImpl.py

from DBHelper.MongoHelper import MongoHelper
from Manage.Company.CompanyObj import SearchCompanyApi, SearchCompanyResult, CompanyIndexApi


class SearchCompanyApiImpl(SearchCompanyApi):
    """企业查询请求接口实现类"""
    db = MongoHelper("tfse_v0.21")

    def conditions_search(self):
        """"""

        search_body = self.search.make_search_body()
        sort = self.sort.make_sort_body()
        page_size = 10 if self.page_size > 10 else self.page_size
        page_no = self.page_no

        total = self.db.find_all_data_with_count(
            "企业数据",
            "企业数据_更新汇总",
            search_body
        )

        records = self.db.find_data_by_page_with_sort(
            "企业数据",
            "企业数据_更新汇总",
            search_body,
            ["企业ID", "企业名称", "一级行业", "综信评价结果.信用等级", "ESG评价结果.评价等级"],
            sort,
            page_size,
            page_no
        )

        table_data = list()
        if records:
            for record in records:
                company = SearchCompanyResult()
                company.cid = record['企业ID']
                company.name = record['企业名称']
                company.industry = record['一级行业']

                # if record.__contains__('综信评价结果'):
                try:
                    company.credit_level = record['综信评价结果']['信用等级']
                # else:
                except KeyError:
                    company.credit_level = None

                # if record.__contains__('ESG评价结果'):
                try:
                    company.esg_level = record['ESG评价结果']['评价等级']
                # else:
                except KeyError:
                    company.esg_level = None

                data = company.dict_to_show(columns=["企业ID", "企业名称", "一级行业", "信用等级", "ESG等级"])

                table_data.append(data)

        result = {
            "records": table_data,
            "total": total
        }

        return result


class CompanyIndexApiImpl(CompanyIndexApi):
    """企业主页接口实现类"""

    db = MongoHelper("tfse_v0.21")

    head_info = CompanyIndexApi.HeadInfo()
    update_time = CompanyIndexApi.UpdateTime()
    basic_info = CompanyIndexApi.BasicInfo()
    financial_data = CompanyIndexApi.FinancialData()
    operating_risk = CompanyIndexApi
    financial_analysis = CompanyIndexApi
    cc_rating = CompanyIndexApi
    esg_rating = CompanyIndexApi

    def get_head_info(self):
        """获取头部信息"""

        self.head_info.email = self.db.find_single_column(
            "应用端",
            "企业用户",
            {"企业ID": self.head_info.cid},
            "邮箱"
        )

        result = self.db.find_single_data(
            "企业数据",
            "企业数据_更新汇总",
            {"企业ID": self.head_info.cid},
            ["企业名称", "一级行业"]
        )

        self.head_info.name = result['企业名称']
        self.head_info.industry = result['一级行业']

    def get_update_time(self):
        """"""

        results = self.db.find_all_data_with_single_sort(
            "企业数据",
            self.update_time.table_map[self.update_time.table],
            {"企业ID": self.update_time.cid},
            ["更新日期"],
            {"更新日期": -1}
        )

        self.update_time.update_time = [result['更新日期'] for result in results]

    def get_basic_info(self):
        """获取基本工商信息"""
        if self.basic_info.__dict__.__contains__('update_time'):
            find_body = {"企业ID": self.basic_info.cid, "更新时间.工商信息": self.basic_info.update_time}
        else:
            find_body = {"企业ID": self.basic_info.cid}

        result = self.db.find_single_data(
            "企业数据",
            "企业数据_更新汇总",
            find_body,
            ["工商信息", "股东信息", "主要成员"]
        )

        if not result:
            result = {'工商信息': {}, "股东信息": {}, "主要成员": {}}

        self.basic_info.business_info = self.dict_to_set(instance=self.basic_info.BusinessInfo, data=result['工商信息'])
        self.basic_info.share_holder = self.instance_list_to_set(instance_list_type=self.basic_info.ShareHolder, data=result['股东信息'])
        self.basic_info.main_member = self.instance_list_to_set(instance_list_type=self.basic_info.MainMember, data=result['主要成员'])

    def get_financial_data(self):
        """获取财务数据"""
        result = self.db.find_single_data(
            "企业数据",
            "基本财务数据",
            {"企业ID": self.financial_data.cid, "报告期": self.financial_data.report_date},
            ["报告期", "资产负债表", "利润表", "补充数据表", "财务指标"]
        )

        if not result:
            result = {'报告期': None, "资产负债表": {}, "利润表": {}, "补充数据表": {}, "财务指标": {}}

        self.financial_data.report_date = result['报告期']
        self.financial_data.balance_sheet = self.dict_to_set(instance=self.financial_data.BalanceSheet, data=result['资产负债表'])
        self.financial_data.profit_sheet = self.dict_to_set(instance=self.financial_data.ProfitSheet, data=result['利润表'])
        self.financial_data.appendix_data = self.dict_to_set(instance=self.financial_data.AppendixDataSheet, data=result['补充数据表'])
        self.financial_data.financial_index = self.dict_to_set(instance=self.financial_data.FinancialIndex, data=result['财务指标'])