基于python的公司名分词比对

import os
from typing import List
import pandas as pd
import companynameparser as cp
import jieba
from pathlib import Path

# 定义常量
class Paths:
    CONCAT_PATH = Path('/tmp/医废联单')
    STD_CLINIC_FILE = Path('/tmp/导出机构信息.xlsx')
    CUSTOM_DICT_PATH = Path('/tmp/custom_dict.txt')
    RESULT_FILE = Path('/tmp/未交医废企业处置诊所疑点表.xlsx')

class ColumnNames:
    UNIT_NAME_A = "机构名称"  # 机构信息表中，单位名称列名
    UNIT_NAME_B = "产废单位"  # 医废联单中，单位名称列名
    FILTER_WORD = "诊所"     # 筛选机构关键词
    UNION_NAME = "简化名"

def init_jieba():
    """初始化结巴分词用户词典"""
    if not Paths.CUSTOM_DICT_PATH.exists():
        raise FileNotFoundError(f"用户词典文件不存在: {Paths.CUSTOM_DICT_PATH}")
    jieba.load_userdict(str(Paths.CUSTOM_DICT_PATH))

def get_filelist(path: Path) -> List[Path]:
    """
    获取传入路径内含子文件夹所有文件的绝对路径
    
    Args:
        path: 指定路径
    Returns:
        包含所有文件绝对路径的列表
    """
    if not path.exists():
        raise FileNotFoundError(f"指定路径不存在: {path}")
    
    return [Path(os.path.join(root, file)) 
            for root, _, files in os.walk(path) 
            for file in files]

def get_brand(column_name: str, row: pd.Series) -> str:
    """
    获取DataFrame某一列根据companynameparser分词的结果
    
    Args:
        column_name: 列名
        row: DataFrame的行数据
    Returns:
        分词后的品牌名
    """
    try:
        listbrand = cp.parse(row[column_name])['brand'].split(",")
        return listbrand[0] if listbrand else ''
    except Exception as e:
        print(f"处理行数据时出错: {row[column_name]}, 错误: {e}")
        return ''

def get_dup_table(unit_name: str, df: pd.DataFrame) -> pd.DataFrame:
    """
    获得单位名称与提取到的brand去重后的DataFrame
    
    Args:
        unit_name: 单位名称
        df: 要去重的DataFrame
    Returns:
        去重后的DataFrame
    """
    return df[[unit_name, ColumnNames.UNION_NAME]].drop_duplicates()

def main():
    """主函数入口"""
    try:
        init_jieba()
        
        # 读取并合并医废联单数据
        dfs = []
        for file in get_filelist(Paths.CONCAT_PATH):
            if file.suffix == '.xls':
                try:
                    df = pd.read_excel(file, engine='xlrd', skiprows=1)
                    dfs.append(df)
                except Exception as e:
                    print(f"读取文件失败 {file}: {e}")
        
        if not dfs:
            raise ValueError("没有找到可用的数据文件")
            
        df_all = pd.concat(dfs)
        
        # 处理医废联单数据
        df_filtered = df_all[df_all[ColumnNames.UNIT_NAME_B].str.contains(
            ColumnNames.FILTER_WORD, na=False)].copy()
        df_filtered[ColumnNames.UNION_NAME] = df_filtered.apply(
            lambda row: get_brand(ColumnNames.UNIT_NAME_B, row), axis=1)
        result_df_b = get_dup_table(ColumnNames.UNIT_NAME_B, df_filtered)
        
        # 处理标准机构表
        df_std = pd.read_excel(Paths.STD_CLINIC_FILE)
        df_std_filtered = df_std[df_std[ColumnNames.UNIT_NAME_A].str.contains(
            ColumnNames.FILTER_WORD, na=False)].copy()
        df_std_filtered[ColumnNames.UNION_NAME] = df_std_filtered.apply(
            lambda row: get_brand(ColumnNames.UNIT_NAME_A, row), axis=1)
        result_df_a = get_dup_table(ColumnNames.UNIT_NAME_A, df_std_filtered)
        
        # 合并结果并输出
        df_merged = pd.merge(result_df_a, result_df_b, 
                           on=ColumnNames.UNION_NAME, how='left')
        unmatched = df_merged[df_merged[ColumnNames.UNIT_NAME_B].isna()]
        unmatched.to_excel(Paths.RESULT_FILE, index=False)
        
    except Exception as e:
        print(f"程序执行出错: {e}")

if __name__ == "__main__":
    main()