BIDI_ML_INFO_EXTRACTION/BiddingKG/dl/interface/get_label_dic.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
@author: bidikeji
@time: 2024/7/23 14:45
"""

from BiddingKG.dl.common.Utils import getUnifyMoney
import math
import re

def chinese_to_arabic(s):
    # 中文数字到阿拉伯数字的映射
    num_map = {'零': 0, '一': 1, '二': 2, '两': 2, '三': 3, '四': 4,
               '五': 5, '六': 6, '七': 7, '八': 8, '九': 9}
    # 单位到倍数的映射
    unit_map = {'十': 10, '百': 100}

    # 初始化结果和当前数值
    result = 0
    current_num = 0
    has_unit = False

    if s.startswith('十'):
        result = 10
    # 遍历字符串
    for char in s:
        if char in num_map:
            # 如果是数字，则进行处理
            if has_unit:
                # 如果之前已经有单位了，则需要将当前数字乘以前面的单位
                result += current_num * unit_map[last_unit]
                current_num = num_map[char]
                has_unit = False
            else:
                # 如果之前没有单位，则直接累加
                current_num = current_num * 10 + num_map[char]
        elif char in unit_map:
            # 如果是单位，则标记为已有单位，并保存最后一个单位
            last_unit = char
            has_unit = True

            # 处理字符串末尾的数字（如果没有单位，则直接加上）
    if current_num != 0:
        if has_unit:
            result += current_num * unit_map[last_unit]
        else:
            result += current_num

    return result

def get_all_label(title, content):
    def is_direct_procurement():
        # 企业直采
        if re.search('询比价|询比|竞价|议价|报价', title) or re.search('我要报价|竞价起止时间|报价起止时间', content) or \
                (re.search('公司|集团|企业', content) and re.search('招标|中标|投标', content) == None):
            return 1
        return 0

    def is_target_small():
        # 专门面向中小企业
        if re.search('专门面向中小微?企业', content):
            if re.search('(非|不属于|不|是/否)）?专门面向(中小微?企业)?|部分面向中小微?企业|专门面向中小企业采购：否', content):
                return 0
            return 1
        elif re.search('仅面向小微企业|专门面向.{,30}中小企业采购|是否专门面向中小微?企业(采购)?：是|本项目为中小型企业预留项目|专门面向中小微?企业', content):
            return 1
        elif re.search('落实政府采购政策需满足的资格要求.{,30}供应商为中小企业', content):
            if re.search('(非|不属于|不|是/否)）?专门面向中小微?企业|部分面向中小微?企业',content):
                return 0
            return 1
        return 0

    def registered_years():
        # 注册年限
        ser = None
        if re.search('禁止.{,5}注册未满(?P<num>([一二三四五六七八九十]+|\d+))(?P<unit>(年|个?月))', content):
            ser = re.search('禁止.{,5}注册未满(?P<num>([一二三四五六七八九十]+|\d+))(?P<unit>(年|个?月))', content)
        elif re.search('(成立|注册)时间：?.{,10}(不[低少]于|大于(等于)?|需满)(?P<num>([一二三四五六七八九十]+|\d+))(?P<unit>(年|个?月))', content):
            ser = re.search('(成立|注册)时间：?.{,10}(不[低少]于|大于(等于)?|需满)(?P<num>([一二三四五六七八九十]+|\d+))(?P<unit>(年|个?月))',
                            content)
        elif re.search('(成立|注册)时间：?.{,10}(?P<num>([一二三四五六七八九十]+|\d+))(?P<unit>(年|个?月)[或及]?以上)', content):
            ser = re.search('(成立|注册)时间：?.{,10}(?P<num>([一二三四五六七八九十]+|\d+))(?P<unit>(年|个?月)[或及]?以上)', content)
        elif re.search('(成立|注册)时间：?.{,10}不满(?P<num>([一二三四五六七八九十]+|\d+))(?P<unit>(年|个?月).{,5}请勿报价)', content):
            ser = re.search('(成立|注册)时间：?.{,10}不满(?P<num>([一二三四五六七八九十]+|\d+))(?P<unit>(年|个?月).{,5}请勿报价)', content)
        if ser:
            num = ser.group('num')
            unit = ser.group('unit')
            if num.isdigit():
                num = int(num)
            else:
                num = chinese_to_arabic(num)
            if '月' in unit: # 向上取整为年
                num = math.ceil(num/12)
            return num
        return 0

    def registered_capital():
        # 注册资本
        ser = None
        if re.search('注册(资本|资金)：?.{,5}(不[低少]于|大于(等于)?|≥)(?P<num>(\d+[\d.]*))(?P<unit>([万亿]?元))', content):
            ser = re.search('注册(资本|资金)：?.{,5}(不[低少]于|大于(等于)?|≥)(?P<num>(\d+[\d.]*))(?P<unit>([万亿]?元))', content)
        elif re.search('注册(资本|资金)：?.{,5}(?P<num>(\d+[\d.]*))(?P<unit>([万亿]?元)[或及]?以上)', content):
            ser = re.search('注册(资本|资金)：?.{,5}(?P<num>(\d+[\d.]*))(?P<unit>([万亿]?元)[或及]?以上)', content)
        if ser:
            num = ser.group('num')
            unit = ser.group('unit')
            return float(getUnifyMoney(num + unit))
        return 0

    def need_qualification():
        # 有资质证书要求
        if re.search('资质要求.{,150}(行业资质|证书|许可证|认证|经营范围|一级|二级|三级|甲级|乙级|丙级|特级|壹级|贰级|叁级)', content):
            return 1
        elif re.search('(提供|有|具备).{,50}(资质|认证|证书|许可证)', content):
            return 1
        elif re.search('资格）?要求：?.{,50}(甲级|丙级|乙级|一级|二级|三级|特级|壹级|贰级|叁级)', content):
            return 1
        elif re.search('认证体系要求', content):
            return 1
        elif re.search('经营范围须包含|经营范围需包含|经营范围需有|经营范围须有|经营范围有|经营范围内含|营业执照范围内包括', content):
            return 1
        return 0

    def need_ca():
        # 7 是否需要办CA
        if re.search('需要.{,20}数字证书|使用.{,20}(签章|数字证书|CA|ca)', content):
            return 1
        elif re.search('办理.{,20}(数字证书|CA|ca|密钥|签章)', content):
            if re.search('无需.{,15}办理', content):
                return 0
            return 1
        elif re.search('(数字证书|CA|ca|密钥).{,5}办理|是否要求供应商使用(CA|ca)数字证书参与：是', content):
            return 1
        if re.search('(不使用|无需).{,20}(数字证书|CA|ca)|是否要求供应商使用(CA|ca)数字证书参与：不要求', content):
            return 0
        return 0

    def need_performance():
        # 有业绩要求
        if re.search('业绩证明|业绩要求|行业业绩|相关业绩|同类项目业绩经验|业绩材料', content):
            if re.search('业绩.{,5}[/无]', content):
                return 0
            return 1
        elif re.search('类似.{,10}业绩', content):
            if re.search('业绩.{,5}如有', content):
                return 0
            return 1
        elif re.search('完成[^。；]{,100}项目', content):
            return 1
        elif re.search('(提供|有|完成|承接|具备|承担)[^。；]{,100}业绩', content):
            return 1
        return 0

    def mode_of_partipation():
        # 参与方式 1线上 2线下 0其他
        if re.search('(平台|网站|http|www|官网|网址|网页|网上中介|邮件|邮箱|客户端|采购网|系统|邮寄).{,20}(注册|报名)', content):
            return 1
        elif re.search('现场报名', content):
            if re.search(
                    '非现场报名|(在线报名|邮件|邮寄|邮箱|线上报名|网络报名).{,10}现场报名|现场报名.{,10}(在线报名|邮件|邮寄|邮箱|线上报名|网络报名)', content) == None:
                return 1
            return 2
        elif re.search('(获取采购文件|文件的获取|文件获取|获取竞价文件|获取招标文件|文件的领取|文件领取|获取投标文件).{,200}'
                       '(平台|线上|客户端|邮寄|网上获取|网站|网址|http|www|邮箱|寄送|网络获取|采购网|网络领购|系统|邮件|在线报名|网络报名|非现场报名|线上报名)', content):
            return 1
        elif re.search('(获取采购文件|文件的获取|文件获取|获取竞价文件|获取招标文件|文件的领取|文件领取|获取投标文件).{,200}'
                       '([\d一二三四五六七八九十]号|接待室|开标室|现场领取|会议室|线下购买|现场获取|X[\d一二三四五六七八九十]|办公楼|现场报名)', content):
            return 2
        elif re.search('(报价地址|报价信息|报价请点击|报价方法|报价式|报价方式|报价提交|报价地点).{,30}'
                       '(平台|网站|http|www|官网|网址|网页|网上中介|邮件|邮箱|客户端|采购网|系统|在线报价|线上报价)', content):
            return 1
        elif re.search('线下报价', content) and re.search('不接受线下报价|线下报价无效', content) == None:
            return 2
        elif re.search('(文件提交|文件递交|递交方式|文件的提交|文件送达地点|递交响应文件|证明材料的递交)', content):
            b = re.search('(文件提交|文件递交|递交方式|文件的提交|文件送达地点|递交响应文件|证明材料的递交)', content).end()
            ser = re.search('联系方式|发布公告的媒介', content[b:b + 200])
            text = content[b:b + ser.start()] if ser else content[b:b + 200]
            if re.search('平台|线上|客户端|邮寄|网站|网址|http|www|邮箱|寄送|采购网|系统|邮件|网页', text):
                return 1
            elif re.search('[\d一二三四五六七八九十]号|接待室|现场递交|开标室|会议室|[\d一二三四五六七八九十]楼|线下递交|办公楼', text):
                return 2
        if re.search('(平台|线上|客户端|网站|网址|http|www|采购网|系统|网页).{,10}递交.{,10}文件', content):
            return 1
        if re.search('(开标地点|投标地点|开标时间和地点|开标时间及地点|开标方式)', content):
            b = re.search('(开标地点|投标地点|开标时间和地点|开标时间及地点|开标方式)', content).end()
            ser = re.search('联系方式|发布公告的媒介', content[b:b + 70])
            text = content[b:b + ser.start()] if ser else content[b:b + 70]
            if re.search(
                    '平台|线上|客户端|网站|网址|http|www|线上开标|采购网|非现场开标|不见面开标|远程异地开启|系统|线上观看开标|网上开标|在线直播的方式开标|远程开标|现场开启|电子卖场|电子开标|开标现场电话联系',
                    text):
                return 1
            elif re.search('[\d一二三四五六七八九十]号|线下开标|接待室|现场递交|开标室|现场开标|会议室|[\d一二三四五六七八九十]楼|办公楼|街道', text):
                return 2
        if re.search('(平台|线上|客户端|网站|网址|http|www|采购网|系统|网页).{,10}开标', content):
            return 1
        elif re.search('不见面开标|非现场开标|远程异地开启|线上观看开标|网上开标|在线直播的方式开标|远程开标|现场开启|非公开开启|电子开标|开标现场电话联系', content):
            return 1
        elif re.search('开启.{,20}地点', content):
            b = re.search('开启.{,20}地点', content).end()
            ser = re.search('联系方式|发布公告的媒介', content[b:b + 70])
            text = content[b:b + ser.start()] if ser else content[b:b + 70]
            if re.search(
                    '平台|线上|客户端|网站|网址|http|www|线上开标|采购网|非现场开标|不见面开标|远程异地开启|系统|线上观看开标|网上开标|在线直播的方式开标|远程开标|电子卖场|电子开标|开标现场电话联系',
                    text):
                return 1
            elif re.search('[\d一二三四五六七八九十]号|线下开标|接待室|现场递交|开标室|现场开标|会议室|[\d一二三四五六七八九十]楼|办公楼|街道', text):
                return 2
        return 0

    # def suitable_small():
    #     # 适合小微企业投标
    #     if re.search('属于专门面向中小企业|有招标单位联系方式|无注册年限要求|无注册资本要求|无资质证书要求|无业绩要求', content):
    #         return 1
    #     elif re.search('属于企业直采|有招标单位联系方式|无注册年限要求|无注册资本要求|无资质证书要求|无业绩要求', content):
    #         return 2
    #     elif re.search('有招标单位联系方式|无注册年限要求|无注册资本要求|无资质证书要求|无业绩要求', content):
    #         return 3
    #     return 0

    def need_government_policy():
        # 有落实政府采购政策需满足的资格要求：1；否则 0
        if re.search('落实政府采购政策需满足的资格要求', content):
            if re.search('落实政府采购政策需满足的资格要求：无，', content) == None:
                return '否'
            return '是'
        return 0

    def consortium_permit():
        # 允许联合体：是；不允许：否；无关键词：0
        if re.search('(接受|允许|欢迎|同意)）?联合体(参与)?投标|联合体投标是合法的|联合体投标的，应|联合体各方应具备承担|投标人可以组成联合体', content):
            if re.search('不(接受|允许|欢迎|同意)）?联合体(参与)?投标|禁止联合体(参与)?投标|投标人不得为联合体|仅接受独立法人投标|投标人必须为独立法人|不得组成联合体|只有独立法人单位可以参与', content):
                return '否'
            return '是'
        elif re.search('联合(体|方|投标人)：|联合体(成员|单位)[12345一二三四五]?：|(联合体)?成员单位[12345一二三四五]?：|特殊普通合伙：|（联合(体|投标人)）|（联合体(成员|单位)方?[12345一二三四五]?）|（(联合体)?成员单位[12345一二三四五]?）|（特殊普通合伙|成员?）|[，；]成：|（成[），]|与[^，。]{6,100}联合体', content):
            return '是'
        return 0

    label_dic = {}
    is_direct_procurement = is_direct_procurement() # 是否直接采购
    is_target_small = is_target_small() # 是否面向中小企业
    mode_of_partipation = mode_of_partipation() # 参与方式
    need_ca = need_ca() # 是否需要CA
    need_performance = need_performance() # 有业绩要求
    need_qualification = need_qualification() # 资质要求
    registered_capital = registered_capital() # 注册资本
    registered_years = registered_years() # 注册年限
    # suitable_small = suitable_small() # 适合小微企业
    government_policy = need_government_policy() # 落实政府采购政策需满足的资格要求
    consortium_permit = consortium_permit()

    label_dic['is_direct_procurement'] = is_direct_procurement
    label_dic['is_target_small'] = is_target_small
    label_dic['mode_of_partipation'] = mode_of_partipation
    label_dic['need_ca'] = need_ca
    label_dic['need_performance'] = need_performance
    label_dic['need_qualification'] = need_qualification
    label_dic['registered_capital'] = registered_capital
    label_dic['registered_years'] = registered_years
    # label_dic['suitable_small'] = suitable_small
    label_dic['government_policy'] = government_policy
    label_dic['consortium_permit'] = consortium_permit

    label_dic = {k: v for k, v in label_dic.items() if v!=0}

    return label_dic

if __name__ == "__main__":
    # with open('D:\html/2.html', 'r', encoding='UTF-8') as f:
    #     html = f.read()
    # rs = get_all_label('', html)
    # print('rs: ', rs)

    import pandas as pd
    from bs4 import BeautifulSoup
    import json

    df = pd.read_csv(r'E:\导出数据\2024-03-18入库去重后所有公告_输入要素.csv')[:]
    df1 = pd.read_csv(r'E:\导出数据\2024-03-18入库去重后所有公告_html.csv')[:]
    df = df.merge(df1, how='inner', on='docid')
    df.fillna('', inplace=True)
    # df = df[:10]

    print(df.columns, len(df))
    # df.drop_duplicates(subset=['docchannel', 'web_source_name', 'exist_table'], inplace=True)
    # print(len(df))
    def get_text(html):
        soup = BeautifulSoup(html, 'lxml')
        text = soup.get_text()
        return text
    df['content'] = df['dochtmlcon'].apply(lambda x: get_text(x))
    df['标签'] = df.apply(lambda x: get_all_label(x['doctitle'], x['content']), axis=1)

    for k in ['is_direct_procurement', 'is_target_small', 'mode_of_partipation', 'need_ca', 'need_performance', 'need_qualification', 'registered_capital', 'registered_years']:
        df[k] = df['标签'].apply(lambda x: x[k])

    df['标签'] = df['标签'].apply(lambda x: json.dumps(x, ensure_ascii=False, indent=2))
    df.drop_duplicates(subset=['docchannel', 'web_source_no', 'is_direct_procurement', 'is_target_small', 'mode_of_partipation', 'need_ca', 'need_performance', 'need_qualification', 'registered_capital', 'registered_years'], inplace=True)
    df = df[['docid', 'docchannel', 'web_source_no', '标签', 'is_direct_procurement', 'is_target_small', 'mode_of_partipation', 'need_ca', 'need_performance', 'need_qualification', 'registered_capital', 'registered_years']]
    df.to_excel('E:/2024-03-18公告标签提取结果.xlsx', index=False)