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)}")