分词工具 v0.42 发布：Bug 修复

KSM说道：

# -*- 编码：utf-8 -*-
导入操作系统
导入 win32com.client
导入 pandas as pd
from pandas import DataFrame
from kiwipiepy import Kiwi
从 konlpy.tag 导入 Komoran # 用于测试(2021-02-21)
# 从 datetime 导入 datetime
导入日期时间
导入 re
导入 argparse
导入时间
导入多进程

_version_ = '0.41'‘

# 版本历史记录
# v0.42（2023-02-24）：修复了仅从路径中的文件（而非数据库表或列注释文件）提取单词时出现的“列 ['DBSchema'] 不存在‘错误。
# v0.41（2023-02-19）：从数据库表和列注释文件中提取单词后，在“词频”工作表中添加了数据库模式出现频率 (DBSchema_Freq) 项。
# v0.40（2021-08-29）：在从 MS Word、PowerPoint 和文本文件中提取单词后，向“词频”工作表添加了“来源”项。
# v0.30（2021-04-26）：从数据库表“词频”表的注释文件列中提取词语后，添加了“来源”项。
# v0.20（2021-02-21）：启用多进程版本
# v0.10（2021-01-10）：初始版本

def get_word_list(df_text) -> DataFrame:
“””
从文本提取结果 DataFrame 中提取名词，并将最终输出作为 DataFrame 类型返回。
:param df_text: 从文件中提取的文本（DataFrame 类型）
返回：名词、复合词（一个或多个名词，前缀+名词+后缀）的提取结果（数据框类型）
“””
开始时间 = 时间.时间()
word_data_list = []

kiwi = Kiwi()
# tagger = Komoran()
row_idx = 0
对于 df_text.iterrows() 中的每一行索引：
row_idx += 1
如果 row_idx % 100 == 0: # 每 100 条记录打印当前进度状态
print('[pid:%d] 当前：%d，总计：%d，进度：%3.2f%%' %
(os.getpid(), row_idx, df_text.shape[0], round(row_idx / df_text.shape[0] * 100, 2)))
file_name = row['FileName']
file_type = row['FileType']
page = row['Page']
text = str(row['Text'])
来源 = (row['来源'])
如果 row['FileType'] 在 ('table', 'column') 中，则 is_db = True，否则 False
如果 row['FileType'] == 'table'，则 is_db_table = True，否则 False
如果 row['FileType'] == 'column'，则 is_db_column = True，否则 False
如果 is_db：
db = row['DB']
schema = row['Schema']
表格 = row['表格']
db_schema = row['DBSchema']
如果 is_db_column：
列 = 行['列']

如果文本为 None 或 text.strip() == ""：
继续
尝试：
# 名词 = mecab.nouns(text)
# [O]待办事项：提取包括连续名词前缀 (XPN) 和名词派生后缀 (XSN) 的内容。
# [O]待办事项：当名词（NNG，NNP）连续出现时，提取与每个名词相连的复合名词。
text_pos = [(token.form, token.tag) for token in kiwi.tokenize(text)]
words = [pos for pos, tag in text_pos if tag in ['NNG', 'NNP', 'SL']] # NNG：普通名词，NNP：专有名词
pos_list = [x for (x, y) in text_pos]
tag_list = [y for (x, y) in text_pos]
pos_str = ‘/’.join(pos_list) + ‘/’
tag_str = ‘/’.join(tag_list) + ‘/’
iterator = re.finditer('(NNP/|NNG/)+(XSN/)*|(XPN/)+(NNP/|NNG/)+(XSN/)*|(SL/)+', tag_str)
对于迭代器中的每个 mo：
x, y = mo.span()
如果 x == 0：
start_idx = 0
别的：
start_idx = tag_str[:x].count('/')
end_idx = tag_str[:y].count('/')
sub_pos = ”
# 如果 end_idx – start_idx > 1 且 start_idx 不等于 0 且 end_idx 等于 len(tag_list)：
如果 end_idx – start_idx > 1：
for i in range(start_idx, end_idx):
sub_pos += pos_list[i]
# print('%s[sub_pos]' % sub_pos)
words.append('%s[复合词]' % sub_pos) # 注册附加词素

如果 len(words) >= 1：
# print(名词，文本)
for word in words:
# print(名词, '\t', 文本)
data_row = {'FileName': file_name, 'FileType': file_type, 'Page': page, 'Text': text,
‘'Word': word, 'Source': source}
如果 is_db：
data_row['DB'] = db
data_row['Schema'] = schema
data_row['Table'] = 表
data_row['DBSchema'] = db_schema
如果 is_db_column：
data_row['Column'] = 列
word_data_list.append(data_row)
例外情况如下：
print(f'[pid:{os.getpid()}] 文本 {text} 引发了异常')
print(ex)

df_result = pd.DataFrame(word_data_list)
打印（
f'[pid:{os.getpid()}] 输入文本计数:{df_text.shape[0]}，提取的单词计数:{df_result.shape[0]}')
结束时间 = 时间.时间()
# 经过时间 = 结束时间 – 开始时间
elapsed_time = str(datetime.timedelta(seconds=end_time – start_time))
print('[pid:%d] get_word_list 已完成。总计：%d，耗时：%s' %
(os.getpid(), df_text.shape[0], elapsed_time))
返回 df_result

def get_current_datetime() -> str:
返回 datetime.datetime.now().strftime(“%Y-%m-%d %H:%M:%S.%f”)

def get_ppt_text(file_name) -> DataFrame:
“””
从 PPT 文件中提取文本并将其作为 DataFrame 类型返回。
:param file_name: 输入文件名（字符串类型）
返回：从输入文件中提取的文本
“””
#：返回：使用词法分析器从输入文件中的文本中提取名词的 DataFrame
开始时间 = 时间.时间()
print('\r\nget_ppt_text: %s' % file_name)
ppt_app = win32com.client.Dispatch('PowerPoint.Application')
ppt_file = ppt_app.Presentations.Open(file_name, True)
# 结果 = []
df_data = []
页数 = 0
对于 ppt_file.Slides 中的每个幻灯片：
幻灯片编号 = 幻灯片.幻灯片编号
page_count += 1
for shape in slide.Shapes:
shape_text = []
文本 =”
如果 shape.HasTable：
col_cnt = shape.Table.Columns.Count
row_cnt = shape.Table.Rows.Count
for row_idx in range(1, row_cnt + 1):
for col_idx in range(1, col_cnt + 1):
text = shape.Table.Cell(row_idx, col_idx).Shape.TextFrame.TextRange.Text
如果文本不等于”:
text = text.replace(‘\r’, ‘ ‘)
shape_text.append(text)
elif shape.HasTextFrame:
对于 shape.TextFrame.TextRange.Paragraphs() 中的每个段落：
文本 = 段落.文本
如果文本不等于”:
shape_text.append(text)
对于 shape_text 中的每个文本：
如果 text.strip() != ”:
df_data.append({'FileName': file_name, 'FileType': 'ppt', 'Page': slide_number, 'Text': text, 'Source': f'{file_name}:{slide_number}:{text}'})
df_text = pd.DataFrame(df_data)
# 打印(结果)
ppt_file.Close()
# print(df_result)
print('文本数：%s' % str(df_text.shape[0]))
print('页数：%d' % 页数)
# print(df_text.head(10))
# print(df_result.Paragraph)
# 返回 df_result
结束时间 = 时间.时间()
# 经过时间 = 结束时间 – 开始时间
elapsed_time = str(datetime.timedelta(seconds=end_time – start_time))
print('[pid:%d] get_ppt_text 经过时间: %s' % (os.getpid(), elapsed_time))
# 返回 get_word_list(df_text)
返回 df_text

def get_doc_text(file_name) -> DataFrame:
“””
从 doc 文件中提取文本并将其作为 DataFrame 类型返回。
:param file_name: 输入文件名（字符串类型）
返回：从输入文件中提取的文本
“””
#：返回：使用词法分析器从输入文件中的文本中提取名词的 DataFrame
开始时间 = 时间.时间()
print('\r\nget_doc_text: %s' % file_name)
word_app = win32com.client.Dispatch(“Word.Application”)
word_file = word_app.Documents.Open(file_name, True)
# 结果 = []
df_data = []
页 = 0
对于 word_file.Paragraphs 中的每个段落：
文本 = 段落.范围.文本
page = paragraph.Range.Information(3) # 3: wdActiveEndPageNumber(检查文本的页码)
如果 text.strip() != ”:
df_data.append({'FileName': file_name, 'FileType': 'doc', 'Page': page, 'Text': text, 'Source': f'{file_name}:{page}:{text}'})

df_text = pd.DataFrame(df_data)

word_file.Close()
print('文本数：%s' % str(df_text.shape[0]))
print('页数：%d' % 页)
结束时间 = 时间.时间()
# 经过时间 = 结束时间 – 开始时间
elapsed_time = str(datetime.timedelta(seconds=end_time – start_time))
print('[pid:%d] get_doc_text 耗时: %s' % (os.getpid(), elapsed_time))
# 返回 get_word_list(df_text)
返回 df_text

def get_txt_text(file_name) -> DataFrame:
“””
从txt文件中提取文本并将其作为DataFrame类型返回。
:param file_name: 输入文件名（字符串类型）
返回：从输入文件中提取的文本
“””
#：返回：使用词法分析器从输入文件中的文本中提取名词的 DataFrame
开始时间 = 时间.时间()
print('\r\nget_txt_text: ' + file_name)
df_data = []
行号 = 0
with open(file_name, 'rt', encoding='UTF8') as file:
对于文件中的文本：
行号 += 1
如果 text.strip() != ”:
df_data.append({'FileName': file_name, 'FileType': 'txt', 'Page': line_number, 'Text': text, 'Source': f'{file_name}:{line_number}:{text}'})
df_text = pd.DataFrame(df_data)
print('文本计数：%d' % df_text.shape[0])
print('行数：%d' % 行号)
结束时间 = 时间.时间()
# 经过时间 = 结束时间 – 开始时间
elapsed_time = str(datetime.timedelta(seconds=end_time – start_time))
print('[pid:%d] get_txt_text 耗时: %s' % (os.getpid(), elapsed_time))
# 返回 get_word_list(df_text)
返回 df_text

def make_word_cloud(df_group, now_dt, out_path):
“””
使用包含名词频率的 DataFrame 绘制词云
:param df_group: 名词频度数据框
:param now_dt: 当前日期和时间
:param out_path: 输出路径
返回：无
“””
开始时间 = 时间.时间()
print('\r\n开始创建词云…')
从 wordcloud 导入 WordCloud
导入 matplotlib.pyplot as plt
导入 koreanize_matplotlib
# malgun.ttf # NanumSquare.ttf # NanumSquareR.ttf NanumMyeongjo.ttf # NanumBarunpenR.ttf # NanumBarunGothic.ttf
wc = WordCloud(font_path='.\\font\\NanumBarunGothic.ttf',
background_color='白色',
max_words=500，,
宽度=1800，,
高度=1000
)

# print(df_group.head(10))
words = df_group.to_dict()['Freq']
# 打印（字数）
# words = df_group.T.to_dict('list')
wc.generate_from_frequencies(words)
wc.to_file('%s\\wordcloud_%s.png' % (out_path, now_dt))
# plt.axis('off')
结束时间 = 时间.时间()
# 经过时间 = 结束时间 – 开始时间
elapsed_time = str(datetime.timedelta(seconds=end_time – start_time))
print('make_word_cloud 运行时间：%s' % 运行时间)
# plt.imshow(wc)
# plt.show()

# 待办事项：从 Area Hangul 文件 (hwp) 中提取文本
def get_hwp_text(file_name) -> DataFrame:
经过

# 待办事项：从 PDF 文件中提取文本
def get_pdf_text(file_name) -> 数据框：
经过

# [O]待办事项：从表格中提取文本，列注释
def get_db_comment_text(file_name) -> DataFrame:
“””
从 db_comment 文件中提取文本并将其作为 DataFrame 类型返回。
:param file_name: 输入文件名（字符串类型）
返回：从输入文件中提取的文本
“””
#：返回：使用词法分析器从输入文件中的文本中提取名词的 DataFrame
开始时间 = 时间.时间()
print('\r\nget_db_comment_text: %s' % file_name)
full_path_file_name = os.path.abspath(file_name)

# 区域表注释
df_table_raw = pd.read_excel(full_path_file_name, sheet_name=0, engine='openpyxl')
df_table = df_table_raw.iloc[:, :4].copy()
df_table.columns = ['DB', 'Schema', 'Table', 'Text']
df_table['FileName'] = 完整文件路径
df_table['FileType'] = 'table'‘
df_table['Page'] = 0
df_table = df_table[df_table.Text.notnull()] # 删除没有 Text 值的行
df_table['Source'] = df_table['DB'].astype(str) + '.' + df_table['Schema'].astype(str) + '.' + df_table['Table'].astype(str) \
+ '(' + df_table['Text'].astype(str) + ')'‘
# 末端区域

# 区域列注释
df_column_raw = pd.read_excel(full_path_file_name, sheet_name=1, engine='openpyxl')
df_column = df_column_raw.iloc[:, :5].copy()
df_column.columns = ['DB', 'Schema', 'Table', 'Column', 'Text']
df_column['FileName'] = full_path_file_name
df_column['FileType'] = 'column'‘
df_column['Page'] = 0
df_column = df_column[df_column.Text.notnull()] # 删除没有 Text 值的行
df_column['Source'] = df_column['DB'].astype(str) + '.' + df_column['Schema'].astype(str) + '.' + df_column['Table'].astype(str) \
+ '.' + df_column['Column'].astype(str) + '(' + df_column['Text'].astype(str) + ')'‘
# 末端区域
df_text = pd.concat([df_column, df_table], ignore_index=True)
df_text['DBSchema'] = df_text['DB'] + '.' + df_text['Schema'] # DB.Schema 值已生成（2023-02-19）
# print(df_text)
结束时间 = 时间.时间()
# 经过时间 = 结束时间 – 开始时间
elapsed_time = str(datetime.timedelta(seconds=end_time – start_time))
print('[pid:%d] get_db_comment_text 耗时: %s' % (os.getpid(), elapsed_time))
print('文本数：%s' % str(df_text.shape[0]))
# 返回 get_word_list(df_text)
返回 df_text

def get_file_text(file_name) -> DataFrame:
“””
一个从 MS Word、PowerPoint、文本和 DB Comment（Excel）文件中提取文本的函数
:param file_name: 文件名
返回：从文件中提取的文本（DataFrame 类型）
“””
df_text = DataFrame()
如果文件名以‘.doc’或‘.docx’结尾：
df_text = get_doc_text(file_name)
elif file_name.endswith(('.ppt', '.pptx')):
df_text = get_ppt_text(file_name)
elif file_name.endswith('.txt'):
df_text = get_txt_text(文件名)
elif file_name.endswith(('.xls', '.xlsx', '.xlsb')):
df_text = get_db_comment_text(文件名)
返回 df_text

# 待办事项：并行处理（分别处理 db_comment_file 和 in_path，如果 db_comment_file 很大则拆分处理）
def main():
“””
从指定路径的子文件夹中的文件中提取文本，从每个文本中提取名词，并将它们保存为 Excel 文件。
返回：无
“””

# 区域参数解析和使用设置 ————————————————————-
# 解析器 = argparse.ArgumentParser(usage='用法测试', description='描述测试')
usage_description = r”””— 描述 —
* db_comment_file 或 in_path 中必须包含一个。

执行示例
1. 从文件中提取文本和单词：指定输入路径和输出路径
python word_extractor.py –multi_process_count 4 –in_path .\\test_files –out_path .\out

2. 从数据库注释中提取文本和单词：指定 db_comment_file 和 out_path
python word_extractor.py –db_comment_file “table,column comments.xlsx” –out_path .\out

3. 从文件和数据库注释中提取文本和单词：指定 db_comment_file、in_path 和 out_path
python word_extractor.py –db_comment_file “table,column comments.xlsx” –in_path .\\test_files –out_path .\out

* 数据库表、列注释文件格式
– 第一张工作表（表注释）：数据库名称、模式名称、表名称、表注释
– 第二张工作表（列注释）：数据库名称、模式名称、表名称、列名称、列注释”””

#待办事项：添加选项：是否提取复合词、是否提取英文字符、是否排除长度为1位数字的英文字符……
parser = argparse.ArgumentParser(description=usage_description, formatter_class=argparse.RawTextHelpFormatter)
添加 # 名称参数
parser.add_argument('–multi_process_count', required=False, type=int,
help='同时运行文本提取和单词提取的多进程数（如果未指定，则设置为（逻辑）CPU 计数）')
parser.add_argument('–db_comment_file', required=False,
help='数据库表，列注释信息文件名（例如，comment.xlsx）')
parser.add_argument('–in_path', required=False, help=r'输入文件（ppt、doc、txt）路径（例如，.\in）')
parser.add_argument('–out_path', required=True, help=r'输出文件（xlsx、png）路径（例如，.\out）')

args = parser.parse_args()

如果 args.multi_process_count：
multi_process_count = int(args.multi_process_count)
别的：
multi_process_count = multiprocessing.cpu_count()

db_comment_file = args.db_comment_file
如果 db_comment_file 不为 None 且 os.path.isfile(db_comment_file):
print('未找到db_comment_file: %s' % db_comment_file)
退出(-1)

in_path = args.in_path
out_path = args.out_path
打印（' - - - - - - - - - - '）
print('Word Extractor v%s 开始 — %s' % (_version_, get_current_datetime()))
print('##### 参数 #####')
print('multi_process_count: %d' % multi_process_count)
print('db_comment_file: %s' % db_comment_file)
print('in_path: %s' % in_path)
print('out_path: %s' % out_path)
打印（' - - - - - - - - - - '）
# 结束区域参数解析和用法设置 ————————————————————-

开始时间 = 时间.时间()

df_text = DataFrame() 从 # 文件读取的文本
df_result = DataFrame() 从 # df_text 中提取的单词
文件列表 = []
如果 in_path 不为 None 且 in_path.strip() != ”:
print('[%s] 开始获取文件列表…' % get_current_datetime())
in_abspath = os.path.abspath(in_path) # os.path.abspath('.') + '\\test_files'‘
file_types = ('.ppt', '.pptx', '.doc', '.docx', '.txt')
for root, dir, files in os.walk(in_abspath):
for file in sorted(files):
要排除的 # 文件
如果文件以‘~’开头：
继续
包含 # 文件
如果文件以文件类型结尾：
file_list.append(root + '\\' + file)

print('[%s] 获取文件列表完成。' % get_current_datetime())
print('— 文件列表 —')
print(‘\n’.join(file_list))

如果 db_comment_file 不为 None：
file_list.append(db_comment_file)

print('[%s] 开始获取文件文本…' % get_current_datetime())
使用 multiprocessing.Pool(processes=multi_process_count) 作为 pool：
mp_text_result = pool.map(get_file_text, file_list)
df_text = pd.concat(mp_text_result,ignore_index=True)
print('[%s] 获取文件文本完成。' % get_current_datetime())
文本提取已完成至 #。以下开始提取单词。

# ———- 并行执行 ———-
print('[%s] 开始从文件文本中获取 Word…' % get_current_datetime())
导入数学
chunk_size = math.ceil(len(df_text) / multi_process_count) if multi_process_count > 0 else 1
如果 chunk_size == 0：
df_text_split = [df_text]
别的：
df_text_split = [df_text[i:i + chunk_size] for i in range(0, len(df_text), chunk_size)]
# mp_result = []
使用 multiprocessing.Pool(processes=multi_process_count) 作为 pool：
mp_result = pool.map(get_word_list, df_text_split)

df_result = pd.concat(mp_result, ignore_index=True)
# 如果 'DB' 不在 df_result.columns 中：
# df_result['DB'] = ‘
# df_result['Schema'] = ‘
# df_result['Table'] = ‘
# df_result['Column'] = ‘
# df_result['DBSchema'] = ‘

print('[%s] 从文件文本中获取 Word 完成。' % get_current_datetime())
# ———————————

print('[%s] 开始获取词频…' % get_current_datetime())
如果 'DB' 在 df_result.columns 中：
df_group = df_result.groupby('Word').agg({
‘'单词': '计数'，,
‘Source’: lambda x: ‘\n’.join(list(x)[:10]),
‘'DBSchema': '唯一'’
}).rename(columns={
‘'词': '频率',
‘'来源': '来源',
‘'DBSchema': 'DBSchema_Freq'’
})
别的：
df_result['DB'] = ‘
df_result['Schema'] = ‘
df_result['Table'] = ‘
df_result['Column'] = ‘
df_result['DBSchema'] = ‘

df_group = df_result.groupby('Word').agg({
‘'单词': '计数'，,
‘Source’: lambda x: ‘\n’.join(list(x)[:10])
}).rename(columns={
‘'词': '频率',
‘'来源': '来源'’
})
df_group = df_group.sort_values(by='Freq', ascending=False)
print('[%s] 获取词频完成。' % get_current_datetime())
# df_group['Len'] = df_group['Word'].str.len()
# df_group[‘Len’] = df_group[‘Word’].apply(lambda x: len(x))
print('[%s] 开始生成词云…' % get_current_datetime())
now_dt = datetime.datetime.now().strftime(“%Y%m%d%H%M%S”)
make_word_cloud(df_group, now_dt, out_path)
print('[%s] 完成词云制作。' % get_current_datetime())

print('[%s] 开始将提取结果保存到 Excel 文件…' % get_current_datetime())
df_result.index += 1
excel_style = {
‘'font-size': '10pt'’
}
df_result = df_result.style.set_properties(**excel_style)
df_group = df_group.style.set_properties(**excel_style)
out_file_name = '%s\\extract_result_%s.xlsx' % (out_path, now_dt) # 'out\\extract_result_%s.xlsx' % now_dt

print('开始写入Excel文件…')
使用 pd.ExcelWriter(path=out_file_name, engine='xlsxwriter') 作为写入器：
df_result.to_excel(writer,
header=True，,
sheet_name='词语提取结果',
index=True，,
index_label='否',
freeze_panes=(1, 0),
columns=['Word', 'FileName', 'FileType', 'Page', 'Text', 'DB', 'Schema', 'Table', 'Column'])
df_group.to_excel(writer,
header=True，,
sheet_name='词频',
index=True，,
index_label='单词',
freeze_panes=(1, 0))
工作簿 = 作家.书
工作表 = writer.sheets['词频']
wrap_format = workbook.add_format({'text_wrap': True})
worksheet.set_column(“C:C”, None, wrap_format)

# print('Excel 文件写入完成')
print('[%s] 完成将提取结果保存到 Excel 文件…' % get_current_datetime())

结束时间 = 时间.时间()
# 经过时间 = 结束时间 – 开始时间
elapsed_time = str(datetime.timedelta(seconds=end_time – start_time))
打印（' - - - - - - - - - - '）
print('[%s] 完成。' % get_current_datetime())
print('总耗时：%s' % 耗时)
打印（' - - - - - - - - - - '）

如果 __name__ == '__main__':
主要的（）
# print_usage()
# get_db_comment_text('table,column comments.xlsx')