BIDI_EXTRACTION_AI/BiddingKG/dl/interface/classification_process.py

import pandas as pd
import os
# 指定 Excel 文件路径
# file_path = '/Users/jinxinglin/Desktop/ICT 行业分类.xlsx'
file_path = os.path.dirname(__file__) + '/ICT行业分类_v11.xlsx'

# 1. 读取 Excel 表格
dfs = pd.read_excel(file_path, sheet_name=['标的物关键词表', '标的物规则表','业主关键词表', '业主规则表'])

# 处理各个表的数据
df_p_keyword = dfs['标的物关键词表']
# print(f"标的物关键词表数量: {len(df_p_keyword)}")

df_p_rule = dfs['标的物规则表']
# print(f"标的物规则表数量: {len(df_p_rule)}")

df_e_keyword= dfs['业主关键词表']
# print(f"业主关键词表数量: {len(df_e_keyword)}")

df_e_rule = dfs['业主规则表']
# print(f"业主规则表数量: {len(df_e_rule)}")

def process_rules(entity, df_e_rule=df_e_rule, df_e_keyword=df_e_keyword):
    # 预处理规则表
    df_e_rule = df_e_rule.fillna('')
    df_e_rule = df_e_rule[df_e_rule['status'] == 1]
    df_e_rule = df_e_rule.sort_values(by='priority', ascending=True)
    
    # 预处理关键词表：按rule_id和keyword_type分组
    keyword_groups = df_e_keyword.groupby(['rule_id', 'keyword_type'])['keyword'].agg(list).reset_index()
    keyword_dict = {}
    for _, row in keyword_groups.iterrows():
        key = (row['rule_id'], row['keyword_type'])
        keyword_dict.setdefault(key, []).extend(row['keyword'])
    
    # 使用字典存储结果标签
    result_tags = {}
    # 使用集合存储已匹配的标签
    matched_tags = set()
    
    # 维护一个清理后的实体字符串
    entity_clean = entity
    
    # 动态维护过滤标签集合
    active_filter_category_1 = set()
    active_filter_category_2 = set()
    
    # 存储所有匹配的规则信息
    rule_matches = []
    
    # 第一遍：收集所有匹配的规则
    for _, row in df_e_rule.iterrows():
        rule_id = row['id']
        category_1 = row['category_1']
        category_2 = row['category_2']
        filter_category_1 = row['filter_category_1']
        filter_category_2 = row['filter_category_2']
        contain_conditions = row['contain_conditions']
        priority = row['priority']
        
        # 获取当前规则的所有关键词类型
        exact_words = keyword_dict.get((rule_id, 'exact'), [])
        special_endings = keyword_dict.get((rule_id, 'special_ending'), [])
        exclusion_words = keyword_dict.get((rule_id, 'exclusion'), [])
        shielding_words = keyword_dict.get((rule_id, 'shielding'), [])
        assist_words = keyword_dict.get((rule_id, 'assist'), [])
        
        tag_key = f"{category_1}-{category_2}"

        # 检查当前规则是否被已激活的过滤标签排除
        skip_rule = False
        
        # 一级标签过滤
        if filter_category_1:
            filter_list = filter_category_1.split(',')
            if any(fc in active_filter_category_1 for fc in filter_list):
                skip_rule = True
        
        # 二级标签过滤
        if not skip_rule and filter_category_2:
            filter_list = filter_category_2.split(',')
            if any(f"{category_1}-{fc}" in active_filter_category_2 for fc in filter_list):
                skip_rule = True
        
        if skip_rule:
            continue
        
        # 1. 精准定位词匹配
        matched_keywords = []
        for kw in exact_words:
            if kw in entity:
                matched_keywords.append(kw)
        
        # 2. 特殊结尾词匹配
        if not matched_keywords:
            for kw in special_endings:
                if entity.endswith(kw):
                    matched_keywords.append(kw)
        
        # 3. 辅助定位词处理流程（需要检查contain_conditions）
        if not matched_keywords and assist_words:
            # 检查contain_conditions条件
            conditions_met = True
            if contain_conditions:
                # 拆分条件词
                conditions = [c.strip() for c in contain_conditions.split(',') if c.strip()]
                # 检查实体中是否包含任意一个条件词
                conditions_met = any(cond in entity for cond in conditions)
            
            # 如果条件满足，才进行辅助词匹配
            if conditions_met:
                # 过滤词检查
                if exclusion_words and any(kw in entity for kw in exclusion_words):
                    continue
                
                # 屏蔽词处理
                if shielding_words:
                    for kw in shielding_words:
                        entity_clean = entity_clean.replace(kw, "")
                
                # 辅助定位词匹配
                for kw in assist_words:
                    if kw in entity_clean:
                        matched_keywords.append(kw)
        
        # 如果有匹配的关键词
        if matched_keywords:
            # 记录匹配的规则信息
            rule_matches.append({
                'rule_id': rule_id,
                'tag_key': tag_key,
                'priority': priority,
                'keywords': matched_keywords,
                'filter_category_1': filter_category_1,
                'filter_category_2': filter_category_2
            })
            matched_tags.add(tag_key)
            
            # 激活过滤标签
            if filter_category_1:
                active_filter_category_1.update(filter_category_1.split(','))
            if filter_category_2:
                for fc in filter_category_2.split(','):
                    active_filter_category_2.add(f"{category_1}-{fc}")
    
    # 第二遍：根据优先级和排除规则处理匹配结果
    # 按优先级排序（高优先级先处理）
    rule_matches.sort(key=lambda x: x['priority'])

    for match in rule_matches:
        # tag_key = match['tag_key']
        tag_key = match['tag_key'] + "-" + str(match['rule_id'])
        keywords = match['keywords']
        filter_category_1 = match['filter_category_1']
        filter_category_2 = match['filter_category_2']
        
        # 检查是否被排除
        skip = False
        
        # 检查一级排除
        if filter_category_1:
            filter_list = filter_category_1.split(',')
            for fc in filter_list:
                if any(tag.startswith(fc + '-') for tag in matched_tags):
                    skip = True
                    break
        
        # 检查二级排除
        if not skip and filter_category_2:
            filter_list = filter_category_2.split(',')
            for fc in filter_list:
                exclude_tag = f"{tag_key.split('-')[0]}-{fc}"
                if exclude_tag in matched_tags:
                    skip = True
                    break
        
        # 如果未被排除，添加标签
        if not skip:
            # 添加当前标签
            if tag_key not in result_tags:
                result_tags[tag_key] = []
            result_tags[tag_key].extend(keywords)
            matched_tags.add(tag_key)
            
            # 检查并排除冲突标签
            if filter_category_1:
                filter_list = filter_category_1.split(',')
                for fc in filter_list:
                    # 排除所有匹配的一级标签
                    tags_to_remove = [tag for tag in matched_tags if tag.startswith(fc + '-')]
                    for tag in tags_to_remove:
                        if tag != tag_key and tag in result_tags:  # 不排除自身
                            del result_tags[tag]
                            matched_tags.remove(tag)
            
            if filter_category_2:
                filter_list = filter_category_2.split(',')
                for fc in filter_list:
                    exclude_tag = f"{tag_key.split('-')[0]}-{fc}"
                    if exclude_tag != tag_key and exclude_tag in result_tags:
                        del result_tags[exclude_tag]
                        matched_tags.remove(exclude_tag)

    return result_tags

# print(process_rules("公安厅", df_e_rule, df_e_keyword))



################  标的物 ##########################
# 新的方法，包含评分规则
import re
import pandas as pd
from itertools import product

def build_group_map(df_keyword):
    """
    构建组名到关键词集合的映射字典
    
    参数:
        df_keyword: 关键词表的DataFrame
        
    返回:
        字典，key为组名，value为该组的关键词集合
    """
    df_keyword = df_keyword[df_keyword['status'] == 1] 
    group_map = {}
    for group_name, group_df in df_keyword.groupby('group'):
        keywords = {str(kw).lower() for kw in group_df['keyword']}
        group_map[group_name] = keywords
    return group_map

def prepare_rules(df_rule):
    """
    预处理规则表
    
    参数:
        df_rule: 规则表的DataFrame
        
    返回:
        排序后的有效规则DataFrame
    """
    active_rules = df_rule[df_rule['status'] == 1].copy()
    return active_rules.sort_values('priority', ascending=True)

def parse_rule(rule_str, group_map):
    """
    解析规则字符串
    
    参数:
        rule_str: 规则字符串
        group_map: 组名到关键词集合的映射字典
        
    返回:
        tuple: (组合模式列表, 排除组集合)
    """
    tokens = re.split(r'([+-])', rule_str)
    tokens = [t.strip() for t in tokens  if t.strip()]
    combine_tokens = []
    exclude_groups = set()
    
    if tokens:
        combine_tokens.append(tokens[0])
    
    i = 1
    while i < len(tokens):
        operator = tokens[i]
        value = tokens[i+1]
        
        if operator == '+':
            combine_tokens.append(value)
        elif operator == '-':
            if value in group_map:
                exclude_groups.add(value)
        
        i += 2
    
    return combine_tokens, exclude_groups

def generate_combinations(combine_tokens, group_map):
    """
    生成所有可能的组合字符串
    
    参数:
        combine_tokens: 组合模式列表
        group_map: 组名到关键词集合的映射字典
        
    返回:
        所有可能的组合字符串集合
    """
    expanded_tokens = []
    for token in combine_tokens:
        if token in group_map:
            expanded_tokens.append(group_map[token])
        else:
            expanded_tokens.append({token})
    
    combinations = set()
    for combo in product(*expanded_tokens):
        # 使用"+"连接多个关键词
        combinations.add('+'.join(combo))
    
    return combinations

def contains_exclusion(text, exclude_groups, group_map):
    """
    检查文本中是否包含任何排除组的关键词
    
    参数:
        text: 待检查的文本
        exclude_groups: 排除组集合
        group_map: 组名到关键词集合的映射字典
        
    返回:
        bool: 如果包含任何排除关键词则返回True，否则返回False
    """
    for group_name in exclude_groups:
        if any(keyword in text for keyword in group_map[group_name]):
            return True
    return False

def find_matched_keywords(text, combinations):
    """
    在文本中查找匹配的组合关键词
    
    参数:
        text: 待匹配的文本（字符串）
        combinations: 所有可能的组合字符串集合（元素格式为"关键词"或"关键词1+关键词2"）
        
    返回:
        匹配的组合字符串集合
    """
    matched = set()
    
    for combo in combinations:
        # 分割组合关键词（支持单个或多个关键词）
        keywords = combo.split('+')
        
        # 检查所有关键词是否都存在于文本中
        if all(keyword in text for keyword in keywords):
            matched.add(combo)
            
    return matched

def extract_core_text(all_text):
    """
    从文本中提取包含特定关键词之后45个汉字以内的核心内容
    
    参数:
        all_text: 完整的文本内容
        
    返回:
        核心文本内容
    """
    # 定义需要匹配的关键词列表
    # keywords = [
    #     '项目名称', '工程名称', '采购名称', '标段名称', '项目的名称', '设备名称', '申购主题',
    #     '申购单主题', '标的', '商品名称', '二级目录', '招标内容', '项目内容', '商品清单',
    #     '标的名称', '采购内容', '集成要求', '概况介绍', '品目分类', '招标范围', '采购范围',
    #     '项目采购分', '采购合同', '招标合同'
    # ]
    keywords = [
        '项目名称','工程名称','采购名称','标段名称','项目的名称','设备名称','申购主题',
        '申购单主题','标的','商品名称','二级目录','招标内容','项目内容','商品清单',
        '标的名称','采购内容','集成要求','概况介绍','品目分类','招标范围','采购范围',
        '项目采购分','采购合同','招标合同','物料描述','物资名称','物料名称',
         '资质要求', '具备', '资格要求', '项目概况', '服务品目', '准入条件', '交易项目',
        '标项名称','规格描述','采购标的','采购需求','需求描述','项目简介','规格描述',
        '采购明细','采购条目','服务描述','服务名称'
    ]
    
    # 创建正则表达式模式，匹配任意一个关键词
    pattern = r'(' + '|'.join(re.escape(kw) for kw in keywords) + r')'
    
    # 查找所有匹配位置
    matches = list(re.finditer(pattern, all_text))
    
    if not matches:
        return ""  # 没有找到关键词
    
    # 取第一个匹配位置
    first_match = matches[0]
    start_pos = first_match.end()  # 关键词结束位置
    
    # 提取关键词之后的内容
    after_text = all_text[start_pos:]
    
    # 提取最多45个汉字
    chinese_chars = []
    count = 0
    for char in after_text:
        # 判断是否为汉字 (Unicode范围)
        if '\u4e00' <= char <= '\u9fff':
            count += 1
            if count > 45:
                break
        chinese_chars.append(char)
    
    # 将字符列表组合成字符串
    core_text = ''.join(chinese_chars).strip()
    return core_text

def normalize_whitespace(text):
    """
    将字符串中的多个空格、换行符、制表符等替换为单个空格
    
    参数:
        text: 输入的文本字符串
        
    返回:
        处理后的文本字符串
    """
    # 使用正则表达式替换所有连续空白字符为单个空格
    normalized_text = re.sub(r'\s+', ' ', text)
    return normalized_text.strip()

def calculate_keyword_position(text, keyword):
    """
    计算关键词在文本中的最后出现位置
    
    参数:
        text: 目标文本
        keyword: 要查找的关键词
        
    返回:
        关键词最后出现的位置索引（从0开始），如果未找到返回-1
    """
    last_index = -1
    start = 0
    while True:
        index = text.find(keyword, start)
        if index == -1:
            break
        last_index = index
        start = index + 1
    return last_index

def calculate_rule_score(text, matched_keywords):
    """
    计算规则的得分
    
    参数:
        text: 目标文本
        matched_keywords: 匹配的关键词集合
        
    返回:
        tuple: (匹配关键词数量, 最高位置分数)
    """
    keyword_count = len(matched_keywords)
    max_position = -1
    
    # 计算所有匹配关键词中的最高位置分数
    for combo in matched_keywords:
        # 对组合中的每个关键词计算位置
        keywords = combo.split('+')
        for keyword in keywords:
            pos = calculate_keyword_position(text, keyword)
            if pos > max_position:
                max_position = pos
    
    return keyword_count, max_position

def classify_text(title, all_text, product_text, df_p_keyword=df_p_keyword, df_p_rule=df_p_rule):
    """
    分类主函数 - 遍历所有规则并返回得分最高的规则
    
    参数:
        title: 公告标题
        all_text: 公告正文
        product: 产品词
        df_p_keyword: 关键词表的DataFrame
        df_p_rule: 规则表的DataFrame
        
    返回:
        tuple: (category_1, category_2, category_3, matched_keywords, rule_id)
        category_1: 一级分类
        category_2: 二级分类
        category_3: 三级分类
        matched_keywords: 匹配的关键词列表
        rule_id: 匹配的规则ID
    """
    # 预处理文本：去除多余空白
    title = normalize_whitespace(title)
    all_text = normalize_whitespace(all_text)
    product_text = normalize_whitespace(product_text)
    
    # 构建组映射
    group_map = build_group_map(df_p_keyword)
    
    # 准备规则
    rules = prepare_rules(df_p_rule)
    
    # 统一转换为小写
    title_lower = title.lower()
    all_text_lower = all_text.lower()
    product_text_lower = product_text.lower()
    
    # 提取核心文本并转换为小写
    core_text = extract_core_text(all_text)
    core_text_lower = core_text.lower() if core_text else ""
    core_text_lower = core_text_lower + product_text_lower
    
    # 存储所有匹配的规则及其得分
    candidate_rules = []
    
    # 遍历所有规则
    for _, rule in rules.iterrows():
        rule_id = rule['id']  # 获取规则ID
        target = rule['target']
        rule_str = rule['rule']
        category_2_str = rule['category_2']
        
        # 选择目标文本
        text = '' 
        if target == 'title':
            text = title_lower
        elif target == 'all_text':
            text = all_text_lower
        elif target == 'core_text':
            text = core_text_lower
        elif target == 'product':
            text = product_text_lower
        else:
            # 默认使用全部文本
            text = f"{title_lower} {all_text_lower} {core_text_lower} {product_text_lower}"
        
        # 解析规则
        combine_tokens, exclude_groups = parse_rule(rule_str, group_map)
        
        # 生成组合
        combinations = generate_combinations(combine_tokens, group_map)
        
        # 查找匹配的关键词
        matched_keywords = find_matched_keywords(text, combinations)
        
        # 如果没有匹配则继续下一条规则
        if not matched_keywords:
            continue
        
        # 检查排除条件
        if contains_exclusion(text, exclude_groups, group_map):
            continue
        
        # 如果是标题规则且匹配成功，立即返回（标题排除规则）
        if category_2_str == '标题排除':
            return (
                rule['category_1'] if 'category_1' in rule else rule.get('categroy_1', '其他'),
                rule.get('category_2', ''),
                rule.get('category_3', ''),
                list(matched_keywords),
                rule_id
            )
        
        # 计算规则得分（匹配关键词数量和最高位置）
        keyword_count, max_position = calculate_rule_score(text, matched_keywords)
        
        # 存储候选规则信息
        candidate_rules.append({
            'rule': rule,
            'matched_keywords': matched_keywords,
            'keyword_count': keyword_count,
            'max_position': max_position,
            'rule_id': rule_id
        })
    
    # 如果没有匹配的规则
    if not candidate_rules:
        return ("其他", "", "", [], "")
    
    # 选择得分最高的规则
    # 1. 按匹配关键词数量降序排序
    # 2. 如果数量相同，按最高位置降序排序（位置越靠后分数越高）
    candidate_rules.sort(key=lambda x: (x['keyword_count'], x['max_position']), reverse=True)
    best_rule = candidate_rules[0]
    
    return (
        best_rule['rule']['category_1'] if 'category_1' in best_rule['rule'] else best_rule['rule'].get('categroy_1', '其他'),
        best_rule['rule'].get('category_2', ''),
        best_rule['rule'].get('category_3', ''),
        list(best_rule['matched_keywords']),
        best_rule['rule_id']
    )


# # 示例1：测试组合规则
# # 读取测试数据
# test_df = pd.read_excel('/Users/jinxinglin/Desktop/工作簿5.xlsx')
# print(f"成功读取测试数据，共 {len(test_df)} 条记录")
#
# # 3. 准备结果存储
# results = []
#
#
# # 4. 处理每条记录
# for idx, row in test_df.iterrows():
#     docid = row['docid']
#     doctitle = row['doctitle'] if pd.notna(row['doctitle']) else ""
#     all_text = row['doctextcon'] if pd.notna(row['doctextcon']) else ""
#     all_text = re.sub(r'\s+', ' ', all_text)
#     product_text = row['product'] if pd.notna(row['product']) else ""
#     tenderee = row['tenderee'] if pd.notna(row['tenderee']) else ""
#     doctitle = doctitle.replace(tenderee, '')
#     all_text = all_text.replace(tenderee, '')
#     # 分类处理
#     category_1, category_2, category_3, matched_keywords, rule_id = classify_text(
#         doctitle, all_text, product_text, df_p_keyword, df_p_rule
#     )
#
#     # 将匹配的关键词列表转换为字符串（用逗号分隔）
#     matched_keywords_str = ", ".join(matched_keywords) if matched_keywords else ""
#
#     # 存储结果
#     results.append({
#         'docid': docid,
#         'doctitle': doctitle,
#         'product_text': product_text,
#         'category_1': category_1,
#         'category_2': category_2,
#         'category_3': category_3,
#         'matched_keywords': matched_keywords_str,
#         'rule_id': rule_id
#     })
#
#     # 每处理100条打印一次进度
#     if (idx + 1) % 100 == 0:
#         print(f"已处理 {idx + 1} 条记录")
#
# # 5. 创建结果DataFrame
# result_df = pd.DataFrame(results)
#
# # 6. 保存结果
# result_df.to_excel('/Users/jinxinglin/Desktop/分类结果.xlsx', index=False)
# print(f"处理记录数: {len(result_df)}")