data_quality_server/tables/fields/NoField.py

# coding:utf-8

from tables import CatchContentObject, fsc
from util.sensitive_word import AcAutomation
from docs.config import amount_config
from docs.config import budget_config
from docs.config import DEBUG
from docs.config import abnormal_config
import csv

class NoFieldChecker(object):
    """
        无字段或空值检查
    """
    def __init__(self):
        self.errors_tables = {
            "title": self.check_title,
            "projectname": self.check_projectname,
            "buyer":self.check_buyer,
            "winner": self.check_winner,
            "budget": self.check_budget,
            "bidamount": self.check_bidamount,
            "area":self.check_region,
            "projectcode": self.check_projectcode,
            "multipackage":self.check_subpackage,
        }

    def check_bidamount(self,obj,catch_content: CatchContentObject) -> bool:
        """
        中标金额为空检测
        :param obj:代表一个item
        :return:返回true 代表异常
        """
        self.check_bidamount_ac = AcAutomation()
        with open(amount_config["table_field_config"]["path"], "r") as f:
            reads = csv.reader(f)
            [self.check_bidamount_ac.add_word(w[0]) for w in reads]

        detail = obj.get("detail", "")
        attach_text = obj.get("attach_text", {})
        subtype = obj.get("subtype", "")
        if subtype in ["中标", "成交","合同","验收"]:
            contents = catch_content.public_attachment_catch(detail, platform="html", document_id="公告") #返回值是字典
            content = "\n".join(contents) #字典处理成字符串
            if self.check_bidamount_ac.search(content):
                return True

            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)
                            content = "\n".join(contents)
                            if self.check_bidamount_ac.search(content):
                                    return True
            return False
        return False
        # 处理正文
        # 检查因素
        # 是否返回 0000

    def check_winner(self,obj, catch_content: CatchContentObject) -> bool:
        """
        中标单位名称为空检测，除中标类型的标讯，其他类型标讯不检查这个字段是否为空
        :param obj:代表一个item
        :return:返回true 代表异常
        """
        subtype = obj.get("subtype", "")
        if subtype in ["中标", "成交", "合同", "验收"]:
            winner = obj.get("winner", "")
            if winner:
                return False
            return True
        return  False
        # 处理正文
        # 检查因素
        # 是否返回 0000

    def check_buyer(self,obj,catch_content: CatchContentObject) -> bool:
        """
        采购单位名称是否为空检测
        :param buyer:采购单位，多个逗号分割
        :param obj:代表一个item
        :return:返回true 代表异常
        """
        buyer = obj.get("buyer", "")
        if buyer :
            return False
        return True

        # 处理正文
        # 检查因素
        # 是否返回 0000

    def check_budget(self,obj, catch_content: CatchContentObject) -> bool:
        """
        预算为空检测
        :param obj:代表一个item
        :return:返回true 代表异常
        """
        self.check_budget_ac = AcAutomation()
        with open(budget_config["table_field_config"]["path"],"r") as f :
            reads=csv.reader(f)
            [self.check_budget_ac.add_word(w[0]) for w in reads ]

        detail = obj.get("detail", "")
        attach_text = obj.get("attach_text", {})
        subtype = obj.get("subtype", "")
        if subtype not in ["中标", "成交", "合同", "验收"]:
            contents = catch_content.public_attachment_catch(detail, platform="html", document_id="公告")  # 返回值是字典
            content = "\n".join(contents)  # 字典处理成字符串
            if self.check_budget_ac.search(content):
                return True

            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)
                            content = "\n".join(contents)
                            if self.check_budget_ac.search(content):
                                return True
            return False
        return False
        # 处理正文
        # 检查因素
        # 是否返回 0000

    def check_region(self,obj, catch_content: CatchContentObject) -> bool:
        """
        区域为空检测
        :param obj:代表一个item
        :return:返回true 代表异常
        """
        pass
        # 处理正文
        # 检查因素
        # 是否返回 0000

    def check_title(self,obj, catch_content: CatchContentObject) -> bool:
        """
        :param obj:代表一个item
        :return:返回true 代表异常
        """
        title = obj.get("title", "")
        if title :
            return False
        return True
        # 处理正文
        # 检查因素
        # 是否返回 0000

    def check_projectname(self,obj, catch_content: CatchContentObject) -> bool:
        """
        :param obj:代表一个item
        :return:返回true 代表异常
        """
        projectname = obj.get("projectname", "")
        if projectname :
            return False
        return True
        # 处理正文
        # 检查因素
        # 是否返回 0000

    def check_projectcode(self,obj, catch_content: CatchContentObject) -> bool:
        """
        项目编号为空检测
        :param obj:代表一个item
        :return:返回true 代表异常
        """
        self.check_projectcode_ac = AcAutomation()
        with open(abnormal_config["table_field_config"]["path4"], "r") as f:
            reads = csv.reader(f)
            [self.check_projectcode_ac.add_word(w[0]) for w in reads]

        projectcode = obj.get("projectcode", "")
        detail = obj.get("detail", "")
        attach_text = obj.get("attach_text", {})
        if projectcode == "":
            contents = catch_content.public_attachment_catch(detail, platform="html", document_id="公告") #返回值是字典
            content = "\n".join(contents) #字典处理成字符串
            if self.check_projectcode_ac.search(content):
                return True

            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)
                            content = "\n".join(contents)
                            if self.check_projectcode_ac.search(content):
                                    return True
            return False
        return False
        # 处理正文
        # 检查因素
        # 是否返回 0000
    def check_subpackage(self,obj, catch_content: CatchContentObject) -> bool:
        """
        公司名称检测
        :param obj:代表一个item
        :return:返回true 代表异常
        """
        pass
        # 处理正文
        # 检查因素
        # 是否返回 0000