# -*- coding: utf-8 -*- """ Created by Shengbo.Zhang on 2021/10/08 """ import os import re import logging import pdfplumber from docx import Document from Pdf2Txt.config import * from pdf2docx import Converter from collections import Counter from Pdf2Txt.config import _check_ann_title_processable # 临时关闭pdf2docx模块中Converter的日志输出 logging.disable(logging.INFO) logging.disable(logging.WARNING) def get_string_and_abscissa_list_from_pdf(pdf_path): ''' 从一个PDF文件中直接逐行读取文本内容（除表格以外的正文）以及最左侧字符的距左边距，结果存放在一个列表中 :param pdf_path: 一个字符串，PDF文件的路径地址 :return: 一个列表：string_abscissa_list，列表元素为[i, j]。其中，i为PDF的整行文本块内容，j为该文本块的横坐标（距左边距） ''' string_abscissa_list = [] temp_list = [] temp_string_list = [] temp_abscissa_list = [] with pdfplumber.open(pdf_path) as pdf: for id, page in enumerate(pdf.pages): bboxes = [table.bbox for table in page.find_tables()] def _not_within_bboxes(obj): def _obj_in_bbox(_bbox): v_mid = (obj["top"] + obj["bottom"]) / 2 h_mid = (obj["x0"] + obj["x1"]) / 2 x0, top, x1, bottom = _bbox return (h_mid >= x0) and (h_mid < x1) and (v_mid >= top) and (v_mid < bottom) return not any(_obj_in_bbox(__bbox) for __bbox in bboxes) new_page = page.filter(_not_within_bboxes) words_list = new_page.extract_words() for item in words_list: text = item['text'].replace('\n', '').replace('\t', '').replace(' ', '').replace(' ', '').replace(',', '') x0 = int(str(item['x0']).split('.')[0]) y0 = int(str(item['top']).split('.')[0]) if text != '': temp_list.append([text, x0, y0]) for id, _ in enumerate(temp_list): if id < len(temp_list)-1 and temp_list[id+1][2] != temp_list[id][2] and abs(temp_list[id+1][2] - temp_list[id][2]) <= 3: temp_list[id+1][2] = temp_list[id][2] i = 0 j = 1 while True: if i < len(temp_list): temp_str = temp_list[i][0] while j < len(temp_list): if temp_list[i][2] == temp_list[j][2]: temp_str += temp_list[j][0] else: break j += 1 if i < len(temp_list)-1 and j == len(temp_list): temp_string_list.append(temp_str) temp_abscissa_list.append(temp_list[i][1]) break temp_string_list.append(temp_str) temp_abscissa_list.append(temp_list[i][1]) i = j j += 1 if i == len(temp_list)-1 and j == len(temp_list): temp_string_list.append(temp_list[i][0]) temp_abscissa_list.append(temp_list[i][1]) break else: break for i, j in zip(temp_string_list, temp_abscissa_list): string_abscissa_list.append([i, j]) return string_abscissa_list def get_ann_info_from_pdf(pdf_path): ''' 获取PDF公告文件的头部信息（此处截取了前5行文本，可能包括非头部数据，将在refine_txt_list()中进一步处理） :param pdf_path: 一个字符串，PDF文件的路径地址 :return: 一个列表，存放PDF公告文件的头部信息（例如：证券代码、证券简称、公告编号等） ''' try: with pdfplumber.open(pdf_path) as pdf: string = pdf.pages[0].extract_text() string_split = string.split('\n') ann_info_list = string_split[:10] except: ann_info_list = [] return ann_info_list def get_document_from_pdf_converted_docx(pdf_path, docx_path): ''' 将PDF文件转换为Docx格式，逐行读取Docx文件中的正文内容（除表格以外） :param pdf_path: 一个字符串，PDF文件的路径地址 :return: 一个列表，string_list，存放PDF的逐行文本内容；一个Document实例对象，存放临时的Docx文件 ''' document = None if docx_path == '': output_docx_file_path = f"{os.path.dirname(pdf_path)}//{os.path.basename(pdf_path)[:-4]}_{TEMP_DOCX_SUFFIX}.docx" else: output_docx_file_path = docx_path is_success = get_docx_from_pdf(pdf_path=pdf_path, out_path=output_docx_file_path) if is_success: document = Document(output_docx_file_path) if os.path.exists(output_docx_file_path): os.remove(output_docx_file_path) return document def get_min_abscissa_value(abscissa_list, string_list_length): ''' 计算PDF文本块横坐标的最小值（正文块），这里假设该值至少应大于或等于某一阈值（此处设为文本总行数的1/4） :param abscissa_dict: 一个字典，存放PDF文件中某一文本块的起始横坐标值 :param string_list_length: 整型，PDF的文本字符串列表 :return: 整型，PDF正文块横坐标的最小值 ''' abscissa_x_list = abscissa_list abscissa_x_list_counter = list(dict(Counter(abscissa_x_list)).items()) abscissa_x_list_counter.sort() x_threshold = string_list_length // 4 min_abscissa_value = min(abscissa_x_list) for item in abscissa_x_list_counter: if item[1] >= x_threshold: min_abscissa_value = item[0] break return min_abscissa_value def refine_txt_list(txt, ann_info, string_abscissa_dict): ''' 此时PDF文件的文本字符串列表（正文）已经过首轮处理，此处将对它进行最后的格式上的优化 :param txt: PDF的文本列表，包含PDF的正文文本内容 :param ann_info: PDF的公告的头部信息 :return: 一个新的PDF文本列表 ''' # 格式化PDF的【公告头部信息】 if ann_info != []: new_ann_info_list = [] for i, val in enumerate(ann_info): if val.strip() == '': continue if val.strip()[-4:] == '有限公司': break else: new_ann_info_list.append(' '.join(val.split()) + SEGMENT_SYMBOL) if new_ann_info_list != []: new_ann_info_list[-1] = new_ann_info_list[-1].replace(SEGMENT_SYMBOL, '') if txt[0].strip()[-4:] == '有限公司': for i in range(len(new_ann_info_list)): txt.insert(0, '') for i, val in enumerate(new_ann_info_list): txt[i] = val # 格式化PDF的【公告标题】【董事会承诺说明】 for i, val in enumerate(txt): if i > 10: break else: val = val.strip() if _check_ann_title_processable(val): if SEGMENT_SYMBOL not in val: txt[i] = (SEGMENT_SYMBOL + val) if val[-4:] == '有限公司': if SEGMENT_SYMBOL not in txt[i]: txt[i] = (SEGMENT_SYMBOL + val) if _check_ann_title_processable(txt[i+1]): txt[i+1] = txt[i+1].replace(SEGMENT_SYMBOL, '') if txt[i+2].replace(SEGMENT_SYMBOL, '')[:3] == '本公司': if SEGMENT_SYMBOL not in txt[i+2]: txt[i+2] = (SEGMENT_SYMBOL + txt[i+2]) txt[i+3] = txt[i+3].replace(SEGMENT_SYMBOL, '') break if _check_ann_title_processable(txt[i+2]): txt[i+1] = txt[i+1].replace(SEGMENT_SYMBOL, '') txt[i+2] = txt[i+2].replace(SEGMENT_SYMBOL, '') if txt[i+3].replace(SEGMENT_SYMBOL, '')[:3] == '本公司': if SEGMENT_SYMBOL not in txt[i+3]: txt[i+3] = (SEGMENT_SYMBOL + txt[i+3]) txt[i+4] = txt[i+4].replace(SEGMENT_SYMBOL, '') break # 次轮遍历PDF的文本字符串列表 for i, _ in enumerate(txt): # 格式化PDF的【大小节编号】 if (SEGMENT_SYMBOL not in txt[i]): match_check = [1, 1, 1, 1, 1] # 形如: '一、'的匹配模式 match_1 = re.match('[\u96f6\u4e00\u4e8c\u4e09\u56db\u4e94\u516d\u4e03\u516b\u4e5d\u5341]{1,2}、', txt[i]) # 形如: '1、'的匹配模式 match_2 = re.match('[0-9]{1,2}、', txt[i]) # 形如: '1.'的匹配模式 match_3 = re.match('[0-9]{1,2}\.', txt[i]) # 形如: '（一）'或'(一)'的匹配模式 match_4 = re.match('[\（\(]+[\u96f6\u4e00\u4e8c\u4e09\u56db\u4e94\u516d\u4e03\u516b\u4e5d\u5341]{1,2}[\）\)]+', txt[i]) # 形如: '（1）'或'(1)'的匹配模式 match_5 = re.match('[\（\(]+[0-9]{1,2}[\）\)]+', txt[i]) if match_1: match_check[0] = match_1.start() if match_2: match_check[1] = match_2.start() if match_3: match_check[2] = match_3.start() if match_4: match_check[3] = match_4.start() if match_5: match_check[4] = match_5.start() if 0 in match_check: txt[i] = SEGMENT_SYMBOL + txt[i] # 修正某些情况下【重要内容提示】字段未自成一行的错误 if ('重要内容提示' in txt[i]) and (SEGMENT_SYMBOL not in txt[i]) and (txt[i].index('重要内容提示') == 0): txt[i] = SEGMENT_SYMBOL + txt[i] if txt[i][-1] != '\n': txt[i] += SEGMENT_SYMBOL # 修正某些情况下【单位：元】【单位：人民币元】字段未被删除的错误 if (txt[i] == '单位：元') or (txt[i] == SEGMENT_SYMBOL + '单位：元'): txt[i] = '' if (txt[i] == '单位：人民币元') or (txt[i] == SEGMENT_SYMBOL + '单位：人民币元'): txt[i] = '' # 修正某些情况下【特别提示】字段未自成一行的错误 if ('特别提示' in txt[i]) and (SEGMENT_SYMBOL not in txt[i]) and (txt[i].index('特别提示') == 0): txt[i] = SEGMENT_SYMBOL + txt[i] if txt[i][-1] != '\n': txt[i] += SEGMENT_SYMBOL # 修正某些情况下【特此公告】字段未自成一行的错误 if ('特此公告' in txt[i]) and (SEGMENT_SYMBOL not in txt[i]) and (txt[i].index('特此公告') == 0): txt[i] = SEGMENT_SYMBOL + txt[i] if txt[i][-1] != '\n': txt[i] += SEGMENT_SYMBOL # 修正某些情况下【附件：】【附件1：】字段前未断行的错误 match_6 = re.match('附件[0-9]{0,2}：', txt[i]) if match_6: if match_6.start() == 0: txt[i] = SEGMENT_SYMBOL + txt[i] # 修正某些情况下该行文本与下一行文本内容重复的错误（仅保留一行） if (i+1) < len(txt) and (txt[i] == txt[i+1]): txt[i] = '' # 修正某些情况下【】【●】符号分段的内容未自成一行的错误 if ('' in txt[i]) or ('●' in txt[i]): txt[i] = txt[i].replace('', '').replace('●', '') for idx in range(i+1, len(txt)-1): if ('' in txt[idx]) or ('●' in txt[idx]): break txt[idx] = txt[idx].replace(SEGMENT_SYMBOL, '') if string_abscissa_dict[txt[idx+1].replace(SEGMENT_SYMBOL, '')] < string_abscissa_dict[txt[idx].replace(SEGMENT_SYMBOL, '')]: break # 修正某些情况下非首页页眉重复出现的错误 if i != 0 and txt[i].replace(SEGMENT_SYMBOL, '').replace(' ', '') in txt[0].replace(SEGMENT_SYMBOL, '').replace(' ', ''): txt[i] = '' # 修正某些特殊形式的页码标识未被正确移除的错误 if (re.match('^[0-9]{1,2}/[0-9]{1,2}', txt[i].strip().replace('', ''))) or \ (re.match('^第[0-9]{1,2}页', txt[i].strip().replace('', ''))) or \ (re.match(r'^-[0-9]{1,2}-', txt[i].strip().replace('', ''))): txt[i] = '' return txt def get_docx_from_pdf(pdf_path, out_path): ''' 读入一个PDF文件，将其转换为Docx格式并临时存放于本地 :param pdf_path: 输入的PDF公告文件的完整路径 :param out_path: 输出的中间Docx结果文件的完整路径 :return: 布尔值，是否转换成功 ''' cv = Converter(pdf_path) try: cv.convert(out_path) except Exception: cv.close() return False for p in cv.pages: if not p.finalized: cv.close() return False cv.close() return True def _get_table_row_feat(str): ''' 给定一个空格分割的表格行字符串，计算它的特征（01组成的字符串） :param str: 字符串 :return: 字符串 ''' s = str.split() r = '' for c in s: try: _ = float(c) r += '1' except Exception: r += '0' return r def append_table_from_docx(doc, txt): ''' 读取Docx文件中每个表格的内容，格式化处理后追加至PDF的文本列表中 :param doc: 一个Document对象实例 :param txt: 一个字符串列表，包含PDF的正文文本内容 :return: 一个新的PDF文本列表 ''' data = [] table_txt = [] table_tag = '-' + TABLE_SYMBOL + '-' for table in doc.tables[:]: table_txt.append(f'{table_tag}\n') for i, row in enumerate(table.rows[:]): row_content = [] for cell in row.cells[:]: c = cell.text new_c = c.replace('\n', '').replace(' ','').replace('\t','').replace(',','') row_content.append(new_c) if row_content == []: continue if '本公司' in row_content[0]: local_flag = True for val in txt[:10]: if '本公司' in val: local_flag = False break if local_flag: tmp = SEGMENT_SYMBOL for line in row_content: tmp += line.strip() if '特别提示' in tmp: tmp = tmp[:tmp.index('特别提示')+4]+SEGMENT_SYMBOL+tmp[tmp.index('特别提示')+4:] for id, val in enumerate(txt): if id > 10: break else: if _check_ann_title_processable(val): txt.insert(id+1, tmp) break continue if '证券代码' in row_content[0]: continue data.append(row_content) new_row = '^' + TABLE_CELL_SYMBOL.join(row_content) + '$\n' if new_row.replace(TABLE_CELL_SYMBOL, '') != '^$\n': table_txt.append(new_row) data.append(f'{table_tag}\n') table_txt.append(f'{table_tag}\n') flag = False for i, val in enumerate(table_txt): if val == f'{table_tag}\n': if not flag: flag = True else: table_txt[i] = '^$\n' else: flag = False table_txt = list(filter(lambda x: x != '^$\n', table_txt)) for i, val in enumerate(table_txt): if val == f'{table_tag}\n' and (i > 0) and (i < len(table_txt)-1): feat1 = _get_table_row_feat(table_txt[i-1].replace('\n', '')) feat2 = _get_table_row_feat(table_txt[i+1].replace('\n', '')) if feat1 == feat2: table_txt[i] = '^$\n' if len(table_txt) == 1 and table_txt[0] == f'{table_tag}\n': table_txt[0] = '^$\n' for i, val in enumerate(table_txt): if val == table_tag: continue if val == '^$\n': table_txt[i] = '' continue table_txt[i] = val[1:][:-2] + '\n' txt.extend(table_txt) return txt def output_txt_string(txt_path, txt_string): ''' 将PDF公告的格式化文本字符串写出至一个.txt的纯文本文件 :param txt_path: 纯文本文件的路径 :param txt_string: PDF公告的纯文本字符串 :return: 布尔值，是否写出成功 ''' try: with open(txt_path, "w", encoding='utf-8') as f: f.write(txt_string) # txt_string_split = txt_string.split('\n') # with open(txt_path, "w", encoding='utf-8') as f: # for string in txt_string_split: # if string != '': # f.write('^' + string + '$\n') except: return False return True def refine_table_txt(txt): ''' 对传入的txt_list再进行针对表头和跨页的优化 :param txt: PDF的文本列表，包含PDF的正文文本内容和追加的表格文本内容 :return: 一个新的文本列表 ''' new_txt_list = [] j = -1 for i, _ in enumerate(txt): if txt[i] == f'{TABLE_SYMBOL}\n': j = i break else: new_txt_list.append(txt[i]) table_txt = txt[j:] table_txt = list(filter(None, table_txt)) for i, _ in enumerate(table_txt): if table_txt[i] == f'{TABLE_SYMBOL}\n' and i + 2 < len(table_txt): pre_cut = table_txt[i + 1].split(TABLE_CELL_SYMBOL) if (len(pre_cut) == 1) or (len(pre_cut) == 2 and pre_cut[0] == ''): table_txt[i + 1] = '' if '公司及董事会' in table_txt[i + 1]: table_txt[i + 1] = '' if table_txt[i + 2] == f'{TABLE_SYMBOL}\n': table_txt[i] = '' table_txt[i + 2] = table_txt[i + 1] table_txt[i + 1] = f'{TABLE_SYMBOL}\n' table_txt = list(filter(None, table_txt)) for i, _ in enumerate(table_txt): if table_txt[i] == f'{TABLE_SYMBOL}\n' and i + 2 < len(table_txt): if '同意' in table_txt[i + 1] and table_txt[i + 1].count('同意') == 2: cut = table_txt[i + 1].split(TABLE_CELL_SYMBOL) for k, val in enumerate(cut): if val == '同意': cut[k] += '票数' cut[k+1] += '比例' if val == '反对': cut[k] += '票数' cut[k+1] += '比例' if val == '弃权': cut[k] += '票数' cut[k+1] += '比例' table_txt[i + 1] = TABLE_CELL_SYMBOL.join(cut).replace(SEGMENT_SYMBOL, '')+SEGMENT_SYMBOL table_txt[i + 2] = '' continue cut1 = table_txt[i + 1].split(TABLE_CELL_SYMBOL) set_cut1 = list(set(cut1)) set_cut1.sort(key=cut1.index) set_cut1 = list(filter(None, set_cut1)) cut2 = table_txt[i + 2].split(TABLE_CELL_SYMBOL) set_cut2 = list(set(cut2)) set_cut2.sort(key=cut2.index) set_cut2 = list(filter(None, set_cut2)) head_cut = [] counter = 0 for val1, val2 in zip(set_cut1, set_cut2): if counter: if len(set_cut1) > len(set_cut2): head_cut = set_cut1 else: head_cut = set_cut2 break if val1 == val2: counter += 1 if counter and head_cut: table_txt[i + 1] = TABLE_CELL_SYMBOL.join(head_cut) table_txt[i + 2] = '' if counter: if i+4 < len(table_txt): cut3 = table_txt[i + 3].split(TABLE_CELL_SYMBOL) set_cut3 = list(set(cut3)) set_cut3.sort(key=cut3.index) set_cut3 = list(filter(None, set_cut3)) flag = False for val3 in set_cut3: if re.match(r'^[0-9]+(|.)[0-9]+(|%)$', val3): flag = True break if not flag: cut4 = table_txt[i + 4].split(TABLE_CELL_SYMBOL) set_cut4 = list(set(cut4)) set_cut4.sort(key=cut4.index) set_cut4 = list(filter(None, set_cut4)) counter_2 = 0 for val3, val4 in zip(set_cut3, set_cut4): if counter_2: if len(set_cut4) > len(set_cut3): head_cut = set_cut4 else: head_cut = set_cut3 break if val3 == val4: counter_2 += 1 if counter_2 and head_cut: table_txt[i + 1] = TABLE_CELL_SYMBOL.join(head_cut) table_txt[i + 2] = '' table_txt[i + 3] = '' table_txt[i + 4] = '' for val in table_txt: new_txt_list.append(val) return new_txt_list def get_txt_from_pdf(pdf_path, docx_path=''): ''' 给定一个PDF格式的公告文件，将其转化为格式化的TXT文本字符串 :param pdf_path: 一个字符串，PDF文件的路径地址 :return: 一个字符串，PDF经转换后的纯文本（已格式化，前部正文，后部表格） ''' txt_string = '' ann_info_list = get_ann_info_from_pdf(pdf_path) string_abscissa_list = get_string_and_abscissa_list_from_pdf(pdf_path) document = get_document_from_pdf_converted_docx(pdf_path, docx_path) string_abscissa_dict = {} if ann_info_list != [] and string_abscissa_list != [] and document is not None: abscissa_list = [x[1] for x in string_abscissa_list] min_abscissa_value = get_min_abscissa_value(abscissa_list, len(abscissa_list)) for id, item in enumerate(string_abscissa_list): if id > 10: break if item[0].replace('\n', '')[-4:] == '有限公司': break else: string_abscissa_list[id][1] = min_abscissa_value txt_list = [] for id, item in enumerate(string_abscissa_list): if (not (len(item[0]) <= 3 and item[0].isdigit())): string_abscissa_dict[item[0]] = item[1] if item[1] > min_abscissa_value: if abs(item[1]-min_abscissa_value) <= 8: txt_list.append(item[0]) else: txt_list.append(SEGMENT_SYMBOL + item[0]) else: txt_list.append(item[0]) txt_list = refine_txt_list(txt_list, ann_info_list, string_abscissa_dict) if document is not None: txt_list.append(SEGMENT_SYMBOL) txt_list = append_table_from_docx(doc=document, txt=txt_list) txt_list = refine_table_txt(txt_list) for val in txt_list: txt_string += val return txt_string