from enum import Enum from rapidfuzz import process, fuzz import re import json from pypinyin import lazy_pinyin # 数字转换表(1-20,常见数字) digit_to_chinese = { "1": "一", "2": "二", "3": "三", "4": "四", "5": "五", "6": "六", "7": "七", "8": "八", "9": "九", "10": "十", "11": "十一", "12": "十二", "13": "十三", "14": "十四", "15": "十五", "16": "十六", "17": "十七", "18": "十八", "19": "十九", "20": "二十" } def arabic_to_chinese_number(text): """ 将文中阿拉伯数字转换为中文数字 :param text: 输入文本 :return: 转换后的文本 """ cn_to_arabic = {'一': '1', '二': '2', '三': '3', '四': '4', '五': '5', '六': '6', '七': '7', '八': '8', '九': '9', '零': '0'} arabic_to_cn = {v: k for k, v in cn_to_arabic.items()} # 反向映射 for num, cn in arabic_to_cn.items(): text = text.replace(num, cn) return text def text_to_pinyin(text): """将文本转换为拼音字符串""" return ''.join(lazy_pinyin(text)) def load_standard_data(path): with open(path, "r", encoding="utf-8") as f: return json.load(f) def extract_number(text): """ 提取项目部中的数字(支持阿拉伯数字和中文数字),并转换为统一格式(中文数字)。 """ match = re.search(r'(第?[一二三四五六七八九十百千零\d]+)', text) if match: num_str = match.group(1).replace("第", "") if num_str.isdigit(): return digit_to_chinese.get(num_str, num_str) # 阿拉伯数字转中文 return num_str # 中文数字直接返回 return None def standardize_company_and_project(input_company, input_project, standard_data): """ 将口语化的公司名和项目部名转换为标准化名称。 参数: input_company (str): 用户输入的公司名(可能是口语化或不完整的名称)。 input_project (str): 用户输入的项目部名(可能是口语化或不完整的名称)。 standard_data (dict): 标准化的公司名和项目部名数据,格式为 {公司名: [项目部名1, 项目部名2, ...]}。 返回: tuple: (标准化公司名, 匹配的项目部名列表)。如果无法匹配,返回 (None, None)。 """ # **1. 标准化公司名** company_match = process.extractOne(input_company, standard_data.keys(), scorer=fuzz.ratio) if not company_match or company_match[1] < 65: # 相似度低于 70 可能匹配错误 return None, None standard_company = company_match[0] # **2. 先尝试直接匹配最相似的项目名** project_match = process.extractOne(input_project, standard_data[standard_company], scorer=fuzz.ratio) print(f"项目部名称最相似:{project_match[0]},{project_match[1]}", flush=True) if project_match and project_match[1] >= 86: return standard_company, [project_match[0]] # 直接返回匹配的项目名 # **3. 提取项目部的数字部分** query_number = extract_number(input_project) # **4. 过滤所有符合数字的项目部** matched_projects = [] for project in standard_data[standard_company]: project_number = extract_number(project) if query_number and query_number == project_number: matched_projects.append(project) return standard_company, matched_projects def standardize_company_and_projectDepartment(input_company, input_project, origianl_company_list , company_project_department_map, company_pinyin_to_original_map = None): """ 将口语化的公司名和项目部名转换为标准化名称。 参数: input_company (str): 用户输入的公司名(可能是口语化或不完整的名称)。 input_project (str): 用户输入的项目部名(可能是口语化或不完整的名称)。 company_project_department_map (dict): 标准化的公司名和项目部名数据,格式为 {公司名: [项目部名1, 项目部名2, ...]}。 pinyin_to_original_map:分公司拼音和分公司原始名的映射 返回: tuple: (标准化公司名, 匹配的项目部名列表)。如果无法匹配,返回 (None, None)。 """ try: # **1. 标准化公司名** best_company_match = multiple_standardize_single_name(input_company, origianl_company_list,list(company_pinyin_to_original_map.keys()),company_pinyin_to_original_map,60,85,True) if not best_company_match: return None, None else: standard_company = best_company_match[0] # **2. 先尝试直接匹配最相似的项目名** project_match = process.extract(input_project, company_project_department_map[standard_company], scorer=fuzz.token_sort_ratio, limit=len(company_project_department_map[standard_company])) # project_match = process.extractOne(input_project, company_project_department_map[standard_company], scorer=fuzz.ratio) print(f"项目部名称最相似:{project_match[0]},{project_match[1]}", flush=True) if project_match and project_match[1] >= 86: return standard_company, [project_match[0]] # 直接返回匹配的项目名 # **3. 提取项目部的数字部分** query_number = extract_number(input_project) # **4. 过滤所有符合数字的项目部** matched_projects = [] for project in company_project_department_map[standard_company]: project_number = extract_number(project) if query_number and query_number == project_number: matched_projects.append(project) return standard_company, matched_projects except Exception as e: print(f"standardize_company_and_projectDepartment:{e}", flush=True) return None,None #弃用 def standardize_single_name(input_name, name_list, lower_score=70, high_score=85): """ 将输入的名称(可能是口语化或不完整的名称)转换为标准化名称。 参数: input_name (str): 用户输入的名称(可能是口语化或不完整的名称)。 name_list (list): 标准化的名称列表。 lower_score (int): 匹配的最低相似度阈值,默认值为 70。 high_score (int): 匹配的高置信度阈值,默认值为 85。 返回: list: 匹配的标准化名称列表。如果未找到匹配项,返回 None。 """ match_results = process.extract(input_name, name_list, scorer=fuzz.token_sort_ratio, limit=len(name_list)) # 找到所有相似度 > 80 的匹配项 high_confidence_matches = [(match[0], match[1]) for match in match_results if match[1] > lower_score] print(f"standardize_single_name, high_confidence_matches:{high_confidence_matches}", flush=True) if not high_confidence_matches: return None # 没有找到匹配项 # 返回匹配结果 best_match = max(high_confidence_matches, key=lambda x: x[1], default=None) print(f"best_match: {best_match}", flush=True) if best_match and best_match[1] >= high_score: return [best_match[0]] # 直接返回最高相似度的单个匹配项 return [match[0] for match in high_confidence_matches] def multiple_standardize_single_name(origin_input_name, origin_name_list, pinyin_name_list = None, pinyin_to_original_map = None, lower_score=70, high_score=85, isArabicNumConv = False): """ 使用拼音 + rapidfuzz 进行关键词模糊匹配,并返回原始的标准名 :param input_name: 口语化的名称(中文) :param name_list: 关键词列表(中文) :pinyin_name_list: 关键词列表(拼音) :param pinyin_to_original_map: 拼音到原始标准名的映射 :param lower_score: 低匹配分数阈值(默认70) :param high_score: 高匹配分数阈值(默认85) :return: 最匹配的原始关键词,或 None """ #First round, 原始标准名的匹配性查找,能找到直接返回 if isArabicNumConv: origin_input_name = arabic_to_chinese_number(origin_input_name) match_results = process.extract(origin_input_name, origin_name_list, scorer=fuzz.token_sort_ratio, limit=len(origin_name_list)) # 找到所有相似度 > 80 的匹配项 original_high_confidence_matches = [(match[0], match[1]) for match in match_results if match[1] >= lower_score] print(f"standardize_pinyin_single_name 原始名匹配, high_confidence_matches:{original_high_confidence_matches[:3]}", flush=True) combined_low_confidence_matches = [] if original_high_confidence_matches: origin_best_match = max(original_high_confidence_matches, key=lambda x: x[1], default=None) # 直接返回最高相似度的单个匹配项 # print(f"原始名匹配: {origin_best_match}", flush=True) if origin_best_match and origin_best_match[1] >= high_score: return [origin_best_match[0]] else: combined_low_confidence_matches = [match[0] for match in original_high_confidence_matches[:3]] else: if not pinyin_name_list or not pinyin_to_original_map: return None # #第二轮, 拼音名的匹配性查找,能找到直接返回 pinyin_input_name = text_to_pinyin(origin_input_name) match_results = process.extract(pinyin_input_name, pinyin_name_list, scorer=fuzz.ratio, limit=len(pinyin_name_list)) # 筛选出匹配分数 > lower_score 的结果 pinyin_high_confidence_matches = [(match[0], match[1]) for match in match_results if match[1] >= lower_score] print(f"standardize_pinyin_single_name 拼音匹配, input_name:{pinyin_input_name}, high_confidence_matches:{pinyin_high_confidence_matches[:3]}", flush=True) if not pinyin_high_confidence_matches: return combined_low_confidence_matches # 没有找到匹配项 # 选择最高相似度的匹配项 pinyin_best_match = max(pinyin_high_confidence_matches, key=lambda x: x[1], default=None) if pinyin_best_match and pinyin_best_match[1] >= high_score: return [pinyin_to_original_map[pinyin_best_match[0]]] # 直接返回最高相似度的原始工程名 combined_low_confidence_matches.extend( [pinyin_to_original_map[match[0]] for match in pinyin_high_confidence_matches[:3]] ) # 返回所有匹配项对应的原始名,最多返回最低匹配项的前5个 return list(dict.fromkeys(combined_low_confidence_matches)) def generate_project_prompt(matched_projects, original_name = "", type="项目部名"): """ 生成提示信息,用于让用户确认匹配的项目名或分公司名或项目名。 参数: matched_projects (list): 匹配的项目或分公司名称列表。 type (str): 提示信息的类型(例如 "项目名" 或 "分公司名"),默认值为 "项目名"。 返回: str: 生成的提示信息。如果未找到匹配项,返回提示用户提供更准确信息的字符串。 """ if not matched_projects: return f"未找到匹配的{type}:{original_name},请提供更准确的{type}信息。" prompt = f"您说的{type}可能是:" for idx, project in enumerate(matched_projects, start=1): prompt += f"第{idx}个: {project}," prompt += "请确认您要选择哪一个?" return prompt def load_standard_name(file_path: str): """ 从指定文件中加载标准化的名称列表。 参数: file_path (str): 文件路径,文件应包含标准化的名称列表,每行一个名称。 返回: list: 从文件中读取的标准化名称列表。 异常: FileNotFoundError: 如果文件不存在,抛出此异常。 Exception: 如果读取文件时发生其他错误,抛出此异常。 """ try: with open(file_path, 'r', encoding='utf-8') as file: lines = [line.strip() for line in file if line.strip()] return lines except FileNotFoundError: print(f"错误:文件 {file_path} 不存在", flush=True) raise FileNotFoundError(f"错误:文件 {file_path} 不存在") except Exception as e: print(f"读取文件时发生错误:{e}", flush=True) raise Exception(f"错误:文件 {file_path} 不存在") class CheckResult(Enum): NO_MATCH = 0 # 不符合检查条件 MATCH_FOUND = 1 # 匹配到了值 NEEDS_MORE_ROUNDS = 2 # 需要多轮 class StandardType(Enum): #工程名检查 PROJECT_CHECK = 0 #项目名检查 PROGRAM_CHECK = 1