data_quality_server/tables/fields/buyer.py

# coding:utf-8
from tables.ai import org_ner
from tables import clear_spacing
from html_table_extractor.extractor import Extractor
from tables import CatchContentObject, fsc
from util.sensitive_word import AcAutomation
from tables import match_company_index, key_value_header
from tables import tfc_object
from docs.config import ai2config
from docs.config import DEBUG
from util.get_region import get_city_info
import csv
import re
from docs.config import abnormal_config

pattern = r',|。|\?|!|;'


class BuyerChecker(object):
    """
        中标字段检查
    """

    def __init__(self):
        """
        采购单位0101判断不准确，备用
        """
        self.errors_tables = {
            # "0101": {
            #     "name": "实体识别",
            #     "parent_name": "名称错误",
            #     "parent_code": "01",
            #     "checkFn": self.check0101
            # },
            # "0201": {
            #     "name": "看数据的标签是不是采购单位",
            #     "parent_name": "数据标签错误",
            #     "parent_code": "02",
            #     "checkFn": self.check0201
            # },
            "0103": {
                "name": "包含叠词，异常词汇，特殊词汇",
                "parent_name": "名称错误",
                "parent_code": "01",
                "checkFn": self.check0103
            },
            "0104": {
                "name": "名称不完整",
                "parent_name": "名称错误",
                "parent_code": "01",
                "checkFn": self.check0104
            },
            "0301": {
                "name": "采购单位名称长度<3",
                "parent_name": "名称长度异常错误",
                "parent_code": "03",
                "checkFn": self.check0301
            },
            "0105": {
                "name": "采购单位与中标单位一致",
                "parent_name": "名称错误",
                "parent_code": "01",
                "checkFn": self.check0105
            }
        }
        #
        self.buyer_ac = AcAutomation()
        with open(ai2config["table_field_config"]["corpus_path"], "r") as f:
            reads = csv.reader(f)
            [self.buyer_ac.add_word(d[0].strip()) for d in reads if "采购单位" in d[1] and d[0].strip()]

    def intention_check(self, header):
        """
        意图结果检测
        :param header:
        :return:
        """
        if header in self.buyer_ac.catch:
            if DEBUG:
                print(f"采购单位意图:::>>   **{header}**==>    [采购单位]")
            return True
        tags = tfc_object.predict([header])
        if tags:
            if "采购单位" in tags[0]:
                if DEBUG:
                    print(f"采购单位意图:::>>   **{header}**==>    {tags}")
                return True

    def buyer_intention_table(self, tables: list, companies):
        """
        表格意图检测
        :param tables:
        :param companies:
        :return:
        """

        for row_ind, row in enumerate(tables):
            for col_ind, column in enumerate(row):
                extract_companies = re.findall("|".join(companies), column)
                if extract_companies:
                    if col_ind > 0:
                        status = self.intention_check(row[col_ind - 1])
                        if status:
                            for company in companies:
                                companies.remove(company)
                            if not companies:
                                return False
                    if row_ind > 0 and len(tables[row_ind - 1]) > col_ind:
                        status = self.intention_check(tables[row_ind - 1][col_ind])
                        if status:
                            for company in companies:
                                companies.remove(company)
                            if not companies:
                                return False
                    if self.buyer_ac.search(column):
                        companies = self.buyer_intention_content(companies, column)
                    if not companies:
                        return False
        return companies

    def buyer_intention_content(self, companies, column):
        """
        文本意图检测
        :param companies:
        :param column:
        :return:
        """
        # 公司名称的下标
        indexes = match_company_index(companies, column)
        if not indexes:
            return companies

        # 实体提取的head字段
        start_ind = 0
        for r in indexes:
            start, end, company_name = r
            if company_name not in companies:
                start_ind = end
                continue
            start_ind = start_ind if start_ind > start - 10 else start - 10
            text_ = column[start_ind:end + 10]
            start_ind = end
            head = key_value_header(text_)
            for val, ind in head:
                if self.intention_check(val):
                    if company_name in companies:
                        companies.pop(company_name)
                if not companies:
                    return False
        return companies

    @staticmethod
    def check_company_name(contents: list, companies: list):
        """
        公司名称检测
        :param contents:正文段落分割后
        :param companies: 公司list
        :return:返回False结束流程,list继续流程
        """
        new_content_list = []
        # 合并文本
        for ind, con in enumerate(contents):
            if "<table" in con:
                table = Extractor(con).parse()
                _tables = table.return_list()
                _tables = clear_spacing(_tables)
                for text in _tables:
                    new_content_list.extend(text)
            else:
                # 一段文本
                new_content_list.append(con.replace(" ", ""))

        # 开始判断公司名称
        for text in new_content_list:
            p = r"|".join(companies)
            repatten = p.replace(")", "\)").replace("(", "\(").replace(".", "\.")
            s = re.split(pattern, text)
            for t in s:
                if re.search(repatten, t):
                    al_result = org_ner(t)
                    for company in al_result:
                        if company in companies:
                            if DEBUG:
                                print(f"采购单位实体识别:::>>    **{text}**==>    {company}")
                            companies.remove(company)
                if not companies:
                    return False
        return companies

    def check_intention(self, contents: list, companies: list):
        """
        意图检测
        :param contents:正文段落分割后
        :param companies: 公司list
        :return:返回False结束流程,list继续流程
        """
        for ind, content in enumerate(contents):
            if "<table" in content:
                # 表格处理
                table = Extractor(content).parse()
                _tables = table.return_list()
                _tables = clear_spacing(_tables)
                _table_str = str(_tables)
                if re.search("|".join(companies), _table_str):
                    companies = self.buyer_intention_table(_tables, companies)
                if not companies:
                    return False
                continue
            # 非表格处理
            companies = self.buyer_intention_content(companies, content)
            if not companies:
                return False
        return companies

    def check0101(self, buyer: str, detail: str, attach_text: dict, catch_content: CatchContentObject) -> bool:
        """
        公司名称检测
        :param buyer:采购单位
        :param detail: 公告
        :param attach_text: 附件解析结果
        :param catch_content: 单挑数据缓存
        :return:返回true 代表异常
        """
        try:
            companies = [buyer]
            contents = catch_content.public_attachment_catch(detail, platform="html", document_id="公告")
            companies = self.check_company_name(contents, companies)
            if not companies:
                return False
            for attach_index, attach_content in attach_text.items():
                if attach_content:
                    for topic_index, topic_detail in attach_content.items():
                        # oss地址
                        attach_url = topic_detail.get("attach_url", "")
                        if attach_url:
                            # 获取附件内容
                            st, content = fsc.download_text_content(attach_url)

                            # 下载成功
                            # 超长文本不处理，暂定30万字
                            if st and content.strip():
                                if len(content) > 300000:
                                    continue
                            # 开始检测
                            contents = catch_content.public_attachment_catch(content, platform="attach",
                                                                             document_id=attach_url)
                            companies = self.check_company_name(contents, companies)
                            if not companies:
                                return False
        except Exception as e:
            print(e)
        return True

    def check0201(self, buyer: str, detail: str, attach_text: dict, catch_content: CatchContentObject) -> bool:
        """
        公司名称检测
        :param buyer:采购单位
        :param detail: 公告
        :param attach_text: 附件解析结果
        :param catch_content: 单挑数据缓存
        :return:返回true 代表异常
        """
        companies = [buyer]  # 多中标人

        # 公告意图检测
        contents = catch_content.public_attachment_catch(detail, platform="html", document_id="公告")
        companies = self.check_intention(contents, companies)
        if not companies:
            return False

        # 附件意图检测
        for attach_index, attach_content in attach_text.items():
            if attach_content:
                for topic_index, topic_detail in attach_content.items():
                    # oss地址
                    attach_url = topic_detail.get("attach_url", "")
                    if attach_url:
                        # 获取附件内容
                        st, content = fsc.download_text_content(attach_url)

                        # 下载成功
                        # 超长文本不处理，暂定30万字
                        if st and content.strip():
                            if len(content) > 300000:
                                continue
                        # 开始检测
                        contents = catch_content.public_attachment_catch(content, platform="attach",
                                                                         document_id=attach_url)
                        companies = self.check_intention(contents, companies)
                        if not companies:
                            return False
        return True

    def check0103(self, buyer: str):
        """
        return  True 代表异常
        """
        # 采购单位名称以异常词开始
        with open(abnormal_config["table_field_config"]["path1"], "r") as f:
            reads = csv.reader(f)
            for n in reads:
                p1 = re.compile("^" + n[0])
                if p1.match(buyer):
                    return True

        # 采购单位名称中包含异常词
        self.check_abnormal_ac = AcAutomation()
        with open(abnormal_config["table_field_config"]["path2"], "r") as f:
            reads = csv.reader(f)
            for k in reads:
                if k[0] in (buyer):
                    return True

        # 采购单位名称以异常词结尾
        with open(abnormal_config["table_field_config"]["path3"], "r") as f:
            reads = csv.reader(f)
            for m in reads:
                p2 = re.compile(f"{m[0]}$")
                if p2.search(buyer):
                    return True
        return False

    def check0301(self, buyer: str):
        """
        return  True 代表异常
        """
        if len(buyer) < 3:
            return True
        return False

    def check0104(self, buyer: str, buyerclass: str):
        """
        #如果采购单位类型in ("学校","教育","卫健委","医疗","政府办","政务中心")，则采购单位名称中一般都含有地名
        """
        if buyerclass in ("学校", "教育", "卫健委", "医疗", "政府办", "政务中心"):
            province, city, district = get_city_info(buyer)
            if province == None and city == None and district == None:
                return True
        return False

    def check0105(self, buyer: str, s_winner: str):
        if buyer and s_winner:
            if buyer == s_winner:
                return True
        return False