単語抽出ツール v0.42 展開: Bug fix

KSM より:

# -*- coding: utf-8 -*-
import os
import win32com.client
import pandas as pd
from pandas import DataFrame
from kiwipiepy import Kiwi
# from konlpy.tag import Komoran # For Test(2021-02-21)
# from datetime import datetime
import datetime
import re
import argparse
import time
import multiprocessing

_version_ = '0.41'‘

# Version History
# v0.42(2023-02-24): DB table, column comment ファイルではなく in_path のファイルからのみ単語抽出する場合 “Column(s) ['DBSchema'] do not exist‘ エラーパッチ
# v0.41(2023-02-19): DB table, column comment ファイルから単語抽出後「単語頻度」シートに DB-Schema 発生頻度 (DBSchema_Freq) 項目を追加
# v0.40（2021-08-29）：MS Word、PowerPoint、Textファイルから単語を抽出した後、「単語頻度」シートにソース（Source）項目を追加する
# v0.30(2021-04-26): DB table, column comment ファイルから単語抽出後「単語頻度」シートにソース(Source)項目を追加
# v0.20(2021-02-21): Multiprocessing 適用バージョン
# v0.10(2021-01-10): 初回作成バージョン

def get_word_list(df_text) -> DataFrame:
“「」」”
text 抽出結果 DataFrame から名詞を抽出して最終出力を DataFrame type に return
:param df_text: ファイルから抽出した text(DataFrame type)
：return：名詞、複合語（1つ以上の名詞、接頭辞+名詞+接尾辞）抽出結果（Dataframe type）
“「」」”
start_time = time.time()
word_data_list=[]

kiwi = Kiwi()
# tagger = Komoran()
row_idx = 0
for index, row in df_text.iterrows():
row_idx += 1
if row_idx % 100 == 0: # 100件ごとの現在の進行状況出力
print('[pid:%d] current: %d, total: %d, progress: %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'])
source = (row['Source'])
is_db = True if row['FileType'] in ('table', 'column') else False
is_db_table = True if row['FileType'] == 'table' else False
is_db_column = True if row['FileType'] == 'column' else False
if is_db：
db = row['DB']
schema = row['Schema']
table = row['Table']
db_schema = row['DBSchema']
if is_db_column:
column = row['Column']

if text is None or text.strip() == ”:
continue
try:
# nouns = mecab.nouns(text)
# [O]ToDo: 連続した体言接頭辞（XPN）、名詞派生接尾辞（XSN）まで含めて抽出
# [O]ToDo：名詞（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)
for mo in iterator:
x, y = mo.span()
if x == 0:
start_idx = 0
else:
start_idx = tag_str[:x].count('/')
end_idx = tag_str[:y].count('/')
sub_pos = ”
# if end_idx – start_idx > 1 and not (start_idx == 0 and end_idx == len(tag_list)):
if 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) # 追加形態素登録

if len(words) >= 1:
# print(nouns, text)
for word in words:
# print(noun, '\t', text)
data_row = {'FileName': file_name, 'FileType': file_type, 'Page': page, 'Text': text,
‘'Word': word, 'Source': source}
if is_db：
data_row['DB'] = db
data_row['Schema'] = schema
data_row['Table'] = table
data_row['DBSchema'] = db_schema
if is_db_column:
data_row['Column'] = column
word_data_list.append(data_row)
except Exception as ex:
print(f'[pid:{os.getpid()}] Exception has raised for text: {text}')
print(ex)

df_result = pd.DataFrame(word_data_list)
print（
f'[pid:{os.getpid()}] input text count:{df_text.shape[0]}, extracted word count: {df_result.shape[0]}')
end_time = time.time()
# elapsed_time = end_time – start_time
elapsed_time = str(datetime.timedelta(seconds=end_time – start_time))
print('[pid:%d] get_word_list finished. total: %d, elapsed time: %s' %
(os.getpid(), df_text.shape[0], elapsed_time))
return df_result

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

def get_ppt_text(file_name) -> DataFrame:
“「」」”
ppt ファイルから text を抽出して DataFrame type に return
:param file_name: 入力ファイル名 (str type)
:return: 入力ファイルから抽出した text
“「」」”
# :return: 入力ファイルから抽出した text に形態素アナライザで名詞抽出した DataFrame
start_time = time.time()
print('\r\nget_ppt_text: %s' % file_name)
ppt_app = win32com.client.Dispatch('PowerPoint.Application')
ppt_file = ppt_app.Presentations.Open(file_name, True)
# result = []
df_data=[]
page_count = 0
for slide in ppt_file.Slides:
slide_number = slide.SlideNumber
page_count += 1
for shape in slide.Shapes:
shape_text=[]
text = ”
if 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
if text != ”:
text = text.replace(‘\r’, ‘ ‘)
shape_text.append(text)
elif shape.HasTextFrame:
for paragraph in shape.TextFrame.TextRange.Paragraphs():
text = paragraph.Text
if text != ”:
shape_text.append(text)
for text in shape_text:
if 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)
# print(result)
ppt_file.Close()
# print(df_result)
print('text count: %s' % str(df_text.shape[0]))
print('page count: %d' % page_count)
# print(df_text.head(10))
# print(df_result.Paragraph)
# return df_result
end_time = time.time()
# elapsed_time = end_time – start_time
elapsed_time = str(datetime.timedelta(seconds=end_time – start_time))
print('[pid:%d] get_ppt_text elapsed time: %s' % (os.getpid(), elapsed_time))
# return get_word_list(df_text)
return df_text

def get_doc_text(file_name) -> DataFrame:
“「」」”
docファイルからtextを抽出してDataFrame型にreturn
:param file_name: 入力ファイル名 (str type)
:return: 入力ファイルから抽出した text
“「」」”
# :return: 入力ファイルから抽出した text に形態素アナライザで名詞抽出した DataFrame
start_time = time.time()
print('\r\nget_doc_text: %s' % file_name)
word_app = win32com.client.Dispatch(“Word.Application”)
word_file = word_app.Documents.Open(file_name, True)
# result = []
df_data=[]
page=0
for paragraph in word_file.Paragraphs:
text = paragraph.Range.Text
page = paragraph.Range.Information(3) # 3: wdActiveEndPageNumber(Text のページ番号を確認)
if 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('text count: %s' % str(df_text.shape[0]))
print('page count: %d' % page)
end_time = time.time()
# elapsed_time = end_time – start_time
elapsed_time = str(datetime.timedelta(seconds=end_time – start_time))
print('[pid:%d] get_doc_text elapsed time: %s' % (os.getpid(), elapsed_time))
# return get_word_list(df_text)
return df_text

def get_txt_text(file_name) -> DataFrame:
“「」」”
txt ファイルから text を抽出して DataFrame 型に return
:param file_name: 入力ファイル名 (str type)
:return: 入力ファイルから抽出した text
“「」」”
# :return: 入力ファイルから抽出した text に形態素アナライザで名詞抽出した DataFrame
start_time = time.time()
print('\r\nget_txt_text: ' + file_name)
df_data=[]
line_number = 0
with open(file_name, 'rt', encoding='UTF8′) as file:
for text in file:
line_number += 1
if 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('text count: %d' % df_text.shape[0])
print('line count: %d' % line_number)
end_time = time.time()
# elapsed_time = end_time – start_time
elapsed_time = str(datetime.timedelta(seconds=end_time – start_time))
print('[pid:%d] get_txt_text elapsed time: %s' % (os.getpid(), elapsed_time))
# return get_word_list(df_text)
return df_text

def make_word_cloud(df_group, now_dt, out_path):
“「」」”
名詞の頻度を求めたDataFrameでword cloudを描く
:param df_group: 名詞の頻度 DataFrame
:param now_dt: 現在の日付時刻
:param out_path: 出力パス
:return: None
“「」」”
start_time = time.time()
print('\r\nstart make_word_cloud…')
from wordcloud import WordCloud
import matplotlib.pyplot as plt
import koreanize_matplotlib
# malgun.ttf # NanumSquare.ttf # NanumSquareR.ttf NanumMyeongjo.ttf # NanumBarunpenR.ttf # NanumBarunGothic.ttf
wc = WordCloud(font_path='.\\font\\NanumBarunGothic.ttf',
background_color='white',
max_words=500,
width=1800,
height=1000
)

# print(df_group.head(10))
words = df_group.to_dict()['Freq']
# print(words)
# 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')
end_time = time.time()
# elapsed_time = end_time – start_time
elapsed_time = str(datetime.timedelta(seconds=end_time – start_time))
print('make_word_cloud elapsed time: %s' % elapsed_time)
# plt.imshow(wc)
# plt.show()

# Todo：下のハングルファイル（hwp）からテキストを抽出する
def get_hwp_text(file_name) -> DataFrame:
pass

# Todo：PDFファイルからテキストを抽出する
def get_pdf_text(file_name) -> DataFrame:
pass

# [O]ToDo: Table, column comment から text を抽出
def get_db_comment_text(file_name) -> DataFrame:
“「」」”
db_comment ファイルから text を抽出して DataFrame type に return
:param file_name: 入力ファイル名 (str type)
:return: 入力ファイルから抽出した text
“「」」”
# :return: 入力ファイルから抽出した text に形態素アナライザで名詞抽出した DataFrame
start_time = time.time()
print('\r\nget_db_comment_text: %s' % file_name)
full_path_file_name = os.path.abspath(file_name)

# region Table comment
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'] = full_path_file_name
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) + ')'‘
# endregion

# region Column comment
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) + ')'‘
# endregion
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)
end_time = time.time()
# elapsed_time = end_time – start_time
elapsed_time = str(datetime.timedelta(seconds=end_time – start_time))
print('[pid:%d] get_db_comment_text elapsed time: %s' % (os.getpid(), elapsed_time))
print('text count: %s' % str(df_text.shape[0]))
# return get_word_list(df_text)
return df_text

def get_file_text(file_name) -> DataFrame:
“「」」”
MS Word、PowerPoint、Text、DB Comment（Excel）ファイルからテキストを抽出する関数
：param file_name：ファイル名
:return: file から抽出した text(DataFrame type)
“「」」”
df_text = DataFrame()
if file_name.endswith(('.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(file_name)
elif file_name.endswith(('.xls', '.xlsx', '.xlsb')):
df_text = get_db_comment_text(file_name)
return df_text

# Todo：並列処理（db_comment_fileとin_path処理の分離、db_comment_fileが大きい場合は分割処理）
def main():
“「」」”
指定したパスサブフォルダのFileからTextを抽出し、各Textの名詞を抽出してExcelファイルとして保存
：return：なし
“「」」”

# region Args Parse & Usage set-up ————————————————————-
# parser = argparse.ArgumentParser(usage='usage test', description='description test')
usage_description = r”””— Description —
* db_comment_fileとin_pathの1つは必須として入力

*実行例
1. Fileからtext、単語抽出：in_path、out_pathを指定
python word_extractor.py –multi_process_count 4 –in_path .\\test_files –out_path .\out

2. DB commentからtext、単語抽出：db_comment_file、out_pathを指定
python word_extractor.py –db_comment_file “table,column comments.xlsx” –out_path .\out

3. File, DB comment から text, 単語抽出: 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

* DB Table, Column comment ファイル形式
– 最初の sheet(Table comment): DBName, SchemaName, Tablename, TableComment
– 2 番目の sheet(Column comment): DBName, SchemaName, Tablename, ColumnName, ColumnComment”””

# ToDo：オプションを追加：複合語を抽出するかどうか、英字を抽出するかどうか、英字の長さを1桁除外するか、…
parser = argparse.ArgumentParser(description=usage_description, formatter_class=argparse.RawTextHelpFormatter)
# name argument を追加
parser.add_argument('–multi_process_count', required=False, type=int,
help='text 抽出、単語抽出を同時に実行する multi process 個数 (指定しない場合 (logical)cpu 個数に設定される)')
parser.add_argument('–db_comment_file', required=False,
help='DB Table, Column comment 情報ファイル名 (例: 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()

if args.multi_process_count:
multi_process_count = int(args.multi_process_count)
else:
multi_process_count = multiprocessing.cpu_count()

db_comment_file = args.db_comment_file
if db_comment_file is not None and not os.path.isfile(db_comment_file):
print('db_comment_file not found: %s' % db_comment_file)
exit(-1)

in_path = args.in_path
out_path = args.out_path
print('—————————————————————')
print('Word Extractor v%s start — %s' % (_version_, get_current_datetime()))
print('##### arguments #####')
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)
print('—————————————————————')
# endregion Args Parse & Usage set-up ————————————————————-

start_time = time.time()

df_text = DataFrame() # ファイルから読み取った text
df_result = DataFrame() # df_text から抽出した単語
file_list=[]
if in_path is not None and in_path.strip() != ”:
print('[%s] Start Get File List…' % 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):
#除外するファイル
if file.startswith('~'):
continue
# 含めるファイル
if file.endswith(file_types):
file_list.append(root + '\\' + file)

print('[%s] Finish Get File List.' % get_current_datetime())
print('— File List —')
print(‘\n’.join(file_list))

if db_comment_file is not None:
file_list.append(db_comment_file)

print('[%s] Start Get File Text…' % get_current_datetime())
with multiprocessing.Pool(processes=multi_process_count) as pool:
mp_text_result = pool.map(get_file_text, file_list)
df_text = pd.concat(mp_text_result, ignore_index=True)
print('[%s] Finish Get File Text.' % get_current_datetime())
#ここまでtext抽出完了。以下に単語抽出を開始

# ———- 並列実行 ———-
print('[%s] Start Get Word from File Text…' % get_current_datetime())
import math
chunk_size = math.ceil(len(df_text) / multi_process_count) if multi_process_count > 0 else 1
if chunk_size == 0:
df_text_split = [df_text]
else:
df_text_split = [df_text[i:i + chunk_size] for i in range(0, len(df_text), chunk_size)]
# mp_result = []
with multiprocessing.Pool(processes=multi_process_count) as pool:
mp_result = pool.map(get_word_list, df_text_split)

df_result = pd.concat(mp_result, ignore_index=True)
# if 'DB' not in df_result.columns:
# df_result['DB'] = ‘
# df_result['Schema'] = ‘
# df_result['Table'] = ‘
# df_result['Column'] = ‘
# df_result['DBSchema'] = ‘

print('[%s] Finish Get Word from File Text.' % get_current_datetime())
# ———————————

print('[%s] Start Get Word Frequency…' % get_current_datetime())
if 'DB' in df_result.columns:
df_group = df_result.groupby('Word').agg({
‘'Word': 'count',
‘Source’: lambda x: ‘\n’.join(list(x)[:10]),
‘'DBSchema': 'nunique'’
}).rename(columns={
‘'Word': 'Freq',
‘'Source': 'Source',
‘'DBSchema': 'DBSchema_Freq'’
})
else:
df_result['DB'] = ‘
df_result['Schema'] = ‘
df_result['Table'] = ‘
df_result['Column'] = ‘
df_result['DBSchema'] = ‘

df_group = df_result.groupby('Word').agg({
‘'Word': 'count',
‘Source’: lambda x: ‘\n’.join(list(x)[:10])
}).rename(columns={
‘'Word': 'Freq',
‘'Source': 'Source'’
})
df_group = df_group.sort_values(by='Freq', ascending=False)
print('[%s] Finish Get Word Frequency.' % get_current_datetime())
# df_group['Len'] = df_group['Word'].str.len()
# df_group[‘Len’] = df_group[‘Word’].apply(lambda x: len(x))
print('[%s] Start Make Word Cloud…' % 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] Finish Make Word Cloud.' % get_current_datetime())

print('[%s] Start Save the Extract result to Excel File…' % 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('start writing excel file…')
with pd.ExcelWriter(path=out_file_name, engine='xlsxwriter') as writer:
df_result.to_excel(writer,
header=True,
sheet_name='単語抽出結果',
index=True,
index_label='No',
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))
workbook = writer.book
worksheet = writer.sheets['単語頻度']
wrap_format = workbook.add_format({'text_wrap': True})
worksheet.set_column(“C:C”, None, wrap_format)

# print('finished writing excel file')
print('[%s] Finish Save the Extract result to Excel File…' % get_current_datetime())

end_time = time.time()
# elapsed_time = end_time – start_time
elapsed_time = str(datetime.timedelta(seconds=end_time – start_time))
print('—————————————————————')
print('[%s] Finished.' % get_current_datetime())
print('overall elapsed time: %s' % elapsed_time)
print('—————————————————————')

if __name__ == '__main__':
main()
# print_usage()
# get_db_comment_text('table,column comments.xlsx')