Covenantus detectus или ещё одна DS-задача

Привет, Хабр!

Сегодня с вами участник профессионального сообщества NTA Серебренников Дмитрий.

И по дружбе, и по IT‑службе регулярно сталкиваюсь с задачами Data Science. Решением одной из них планирую сегодня поделиться. Поработаю с кредитной документацией, выжму из неё необходимое для аудиторской проверки. Из инструментов применю ловкость рук, python, pathlib, regex, pandas и Abbyy Finereader.

Пост предназначен прежде всего для столкнувшихся с такой задачкой и тех, кто недавно взял курс в науку о данных. Кстати, о данных — все совпадения случайны, исследуемые материалы вымышлены.

Итак, задача состояла в получении необходимых сущностный (ковенантов) из разных по формату и содержанию документов.

Аналитика входных данных

На входе будущей программы у меня выгруженная ранее документация. Документация отсортирована в директории по сегменту бизнеса/типу документа (например, Кредиты крупнейшего бизнеса и Гарантии малого бизнеса).

Документация представляет из себя pdf‑/docx‑файлы, вложенные в rar‑/zip‑архивы и/или директории. Структура: документация вложена внутрь директорий (Номер ИНН → КОД → Кредитный договор и т. д.).

Так как данные разнородны (имеют разный формат файлов), сразу натыкаемся на мысль, по-хорошему, привести всё к единому формату. По содержимому документация стремится иметь общую структуру. Но есть нюансы:

1. в директориях вместе с основным кредитным договором также размещены дополнительные документы — протоколы, электронные подписи и т. д. Допы в рамках задачи меня не интересуют;

2. сама документация (договоры) представлена как в виде текста, так и сканами.

С нюансами буду бороться народными средствами — ненужные файлы будут игнорироваться структурой и регулярными выражениями, с текстовым форматом работать несложно, а вот распознавать текст на сканах — дело весьма нетривиальное: придётся использовать либо библиотеки CV (Computer Vision), либо решения OCR (Optical character recognition).

Проанализировал, теперь настрою рабочее пространство.

Настройка виртуального окружения И ядра Jupyter

Рекомендую всегда взаимодействовать с Python‑проектами в виртуальном окружении — изолированной рабочей среде. Во‑первых, удобно иметь необходимый и достаточный набор инструментов для создания и запуска конкретной программы. Во‑вторых, это позволяет изолировать используемые компоненты проекта от иных проектов и системы в целом, т. е. избежать конфликта зависимостей (их версий), увеличить стабильность работы приложений.

Внутри виртуального окружения находятся интерпретатор python3, пакетный менеджер pip, директория с установленными библиотеками. Версии компонентов задаются при установке.

Ниже кратко изложены действия по созданию и запуску виртуального окружения в командной строке. Если в системе ещё не установлены Python и Jupyter — вот инструкции, которые помогут это исправить: Python, Jupyter.

Пусть имеется директория, где хотим разместить проект. Например:

C:\DAS\PythonProjects\NewPosts\covis_detect

Настраиваю виртуальное окружение:

#шаблон установки; в моём случае -- python -m venv covis_detect_env
> python -m venv <virtual-environment-name>
#запускаем окружение
> covis_detect_env\Scripts\activate
#в командной строке, слева от директории, должна появится приписка –
название окружения
> (covis_detect_env) C:\PythonProjects\...
#просмотреть список установленных библиотек можно так:
> pip list
#шаблон основной команды -- установка библиотеки через pip; например, 
pip install pandas
> pip install <lib_name>
#чтобы выйти из окружения
> diactivate

Создание Jupyter Kernel

В качестве IDE (среды разработки) буду использовать Jupyter Notebook (/Lab). В нём удобно работать с блоками кода и текстом, анализировать и отображать данные в интерактивном формате — то, что нужно для DS. Две основные сущности ноутбука — cell (ячейка) и kernel (ядро). В ячейках вводится текст/код. Ядро — вычислительный движок, своего рода тоже среда — выполняет ячейки и формирует результат, например, расчёты, таблицы, графики, другой текст. Итак, находимся в настроенном окружении, создаём ядро для задачи:

#устанавливаем пакет с ядром
> pip install ipykernel
#добавляем ядро в Jupyter, в моём случае -- python -m ipykernel install –
user --name=covis_detect_ker
> python -m ipykernel install --name=<kernel-name>
#запускаем Jupyter (notebook или lab) -- использовал 1-й
> jupyter notebook

Основная часть. Лёгкая пробежка по блокам кода

Импорт модулей

Начну с подключения используемых модулей для моей задачи:

import pathlib
import docx2pdf
from lxml import etree
from datetime import datetime as dt
from datetime import timedelta
import subprocess
import pytz
import time
import shutil
import re
import pandas as pd
import fitz
import patoolib

Помимо подключения, модули необходимо также установить. Делаею это уже привычным способом в командной строке или прямо в тетрадке Jupyter:

#флаг -r позволяет установить несколько модулей из файла с зависимостями (requirements)
!pip install -r <reqirements.txt>

Разархивирование файлов и удаление самих архивов

Взаимодействую с ОС‑путями‑файлами с помощью pathlib. Если тезисно, модуль построен на ООП — системный путь рассматривается не как простая строка (os), а как полноценный объект с кучкой полезных методов + обеспечивает одинаковую работу путей к файлам в разных операционных системах. Это приводит к большей элегантности и читаемости кода. Подробнее с преимуществами модуля можно ознакомиться и тут и здесь. Patoolib позволяет обрабатывать (создавать, извлекать, сравнивать, переупаковывать) архивы большинства форматов.

#так можно задать путь рабочей директории
PATH = pathlib.Path(r'C:\DAS\PythonProjects\NewPosts\covis_detect\Кредиты КРП')

def unzip_and_drop(path):
        #интересующие шаблоны
        types = ('*.zip', '*.rar')
        for tp in types:
                #рекурсивно проходимся по директории и поддиректориям
                for file in path.rglob(tp):
                        patoolib.extract_archive(file, outdir=file.parent)
                        pathlib.Path.unlink(file)
        print('''
            ====================== Digital Audit ======================

                          Разархивация успешно выполнена!

            ====================== Digital Audit ======================
                            ''')

Далее необходимо прочитать pdf‑файлы и определить среди них файлы‑сканы. Docx‑файлы пока не трогаю, т.к. в них нет сканов.

Определение pdf-файлов, контент которых не удалось распарсить (сканы)

Блок используется в нескольких местах, поэтому реализован в виде функции.

def convert_pdf2py(file_path):
    file_content = ''
    pdf_file = fitz.open(filepath)
    for current_page in range(len(pdf_file)):
        page = pdf_file.load_page(current_page)
        page_text = page.get_text("text")
        file_content += page_text

    return file_path, file_content

Далее создаю список filepaths_wo_content, куда добавляем пути всех файлов‑сканов (проходим по директориям/поддиректориям через rglob(*.pdf) и определяем пути, для которых функция возвращает пустую строку; *pdf, т.к. сканы были представлены только в pdf).

Распознавание текста на сканах с помощью Abbyy FineReader

Для определения «букв, символов и цифр» на изображениях мной рассматривались 3 инструмента: Pytesseract, Abbyy Finereader и EasyOCR. Ниже приведена сравнительная таблица с ними по требуемым критериям.

Таблица — Сравнение инструментов для определения текста на сканах (меньше — лучше)

Критерием с наибольшим весом (в моём случае) стало внутреннее использование, т.к. данные расположены на рабочем месте с ограничениями на установку дополнительного ПО.

Под сложными по структуре документами подразумеваются документы, содержащие в себе изображения (например, логотип организации), таблицы, нестандартно размещённый текст. Здесь лидером можно выделить Abbyy Finereader. С помощью Pytesseract и EasyOCR возможно добиться более интересного результата, но потребуется гораздо больше времени на обучение моделей под конкретный тип документа, Finereader же — готовый универсальный инструмент из коробки.

На основе публикаций [1], [2], [3], работы коллег и собственного опыта на небольшом тесте был окончательно выбран именно Abbyy FineReader. Стоит отметить, что само по себе решение распространяется платно, но, на моё счастье, в организации есть лицензия на его использование.

Функция ниже производит распознавание текста файлов, хранящихся в формате .pdf. Все файлы должны быть расположены в одной папке, распознанные файлы также сохраняются в одну папку. Пути к обеим папкам передаются в качестве аргументов функции. Сигналом о завершении обработки является появление текстового лога в папке с распознанными документами.

def recognize_text(
        path_HF=r'', #путь для сохранения файла задач
        path_HF_Temp=r'', #путь к временной папке задач
        path_output=r'', #путь, куда будут сохраняться распознанные файлы
        path_input=r'', #путь, где хранятся файлы для распознавания
        fmt="docx"): #формат выходных файлов

    path_user = pathlib.Path.home() #получаем путь к папке пользователя
    # обрабатываем пути
    if (not ":\\" in str(path_input)) and (not "\\\\" in str(path_input)):
                path_input = pathlib.Path.cwd().joinpath(path_input)
    if (not ":\\" in str(path_output)) and (not "\\\\" in str(path_output)):
                path_output = pathlib.Path.cwd() / path_output #джоинить возможно и так

    with open(r'OCR_HF_Task.hft', encoding='utf-16') as task: #открываем шаблон для генерации нового задания
        text = task.read()
        tree = etree.fromstring(text)

    tree.attrib['name'] = 'Digital-аудит'  # задаём название задачи
    tree.attrib['status'] = 'scheduled'

    # Время запуска задания
    tree.attrib['startTime'] = (dt.now(pytz.UTC) + timedelta(minutes=1)).strftime('%H:%M:%S.000 %d.%m.%Y UTC')

    # Папка для сохранения файла задач и импорта
    tree.getchildren()[0].getchildren()[0].attrib["batchToCreateFolder"] = path_user.joinpath(path_HF_Temp)

    # Папка для сохранения результатов распознавания
    tree.getchildren()[0].getchildren()[3].getchildren()[0].attrib["savePath"] = path_output

    # выходной формат
    tree.getchildren()[0].getchildren()[3].getchildren()[0].attrib["exportFormat"] = fmt

    # Папка с файлами для распознавания
    tree.getchildren()[0].getchildren()[1].attrib["folderPath"] = path_input

    # Создаём пути при их отсутствии
    if not pathlib.Path.exists(path_user.joinpath(path_HF)):
                pathlib.Path.mkdir(path_user.joinpath(path_HF))
        
    if not pathlib.Path.exists(path_user.joinpath(path_HF_Temp)):
                pathlib.Path.mkdir(path_user.joinpath(path_HF_Temp))

    if not pathlib.Path.exists(path_output):
        pathlib.Path.mkdir(path_output)

        if not pathlib.Path.exists(path_input):
        pathlib.Path.mkdir(path_input)

    # Сохраняем файл задач
    with open(path_user.joinpath(path_HF).joinpath('OCR_HF_task.hft'), 'w', encoding='utf-16') as task:
        for_save = etree.tostring(tree, encoding='utf-16', pretty_print=True)
        task.write(for_save.decode('utf-16'))

    # Закрываем HotFolder чтобы подгрузить задачу при следующем открытии
    subrocess.call('TASKKILL /F /IM \"HotFolder.exe\"'

    # Запускаем HotFolder
    subprocess.Popen(r'C:\Program Files (x86)\ABBYY FineReader 14\HotFolder.exe')

    print('''
        ====================== ABBYY Hot Folder ======================
        Ожидайте завершение ABBYY Hot Folder и сообщения о завершении.
              Статус обработки отображается в ABBYY Hot Folder.
        ====================== ABBYY Hot Folder ======================
                        ''')
        
    while 'Hot Folder Log.txt' not in pathlib.Path.iterdir(path_output)
        time.sleep(1)

    print("Распознавание текста завершено.")

    if 'Hot Folder Log.txt' in pathlib.Path.iterdir(pathlib.Path.cwd()):
        pathlib.Path.unlink('Hot Folder Log.txt')
        shutil.move(path_output.joinpath("Hot Folder Log.txt"), pathlib.Path.cwd().joinpath("Hot Folder Log.txt"))

    subprocess.call(TASKKILL /F /IM \"HotFolder.exe\"")

В качестве путей с входными/выходными файлами указываем рабочую директорию PATH. Так выглядит интерфейс преобразования pdf‑scan → docx‑text с помощью Abbyy Finereader. Обработка 159 файлов составила чуть меньше 2-х часов.

Итого, на данный момент имеею docx-файлы (первоначальные), docx*-файлы (обработанные сканы), pdf-файлы (текстовые и сканы).

Почему сразу в функции не был задан формат pdf? Было необходимо: 1) сравнить результаты со сканами, 2) сохранить pdf-сканы.

Перевод docx → pdf

Теперь стоит преобразовать все файлы к одному типу (pdf). Используем модуль docx2pdf, который коротким движением строки docx конвертирует в pdf.

docx2pdf.convert(filepath)

Если у docx первоначально был pdf‑аналог (не скан, такое могло быть), docx игнорировался при конвертации.

Структура документации

По структуре документ состоит из логотипа организации, наименования договора, города и времени заключения д., его сторон, статей, печатей, подписей и примечаний. Нас интересуют ковенанты договора — Статья 8. По объёму документ может составлять 20–80 страниц. Основная часть входных данных находилась в отрезке 35–60 страниц.

Перехожу к получению необходимых сущностей из документации. Завожу для них списки.

file_paths = [] #пути до документации
file_contents = [] #содержимое документации, которое затем будет парситься
inns = [] #инн клиента/организации
contract_nums = [] #номер договора

#заготовка для пунктов 4.9 и 4.10 
covis_contents = [] #полное содержимое ковенантов
covis_items = [] #пункты к.
covis_item_values = [] #содержание ковенанта

Определение ИНН организации (имя директории, в которой расположен файл)

def get_INN(file_path):
        if pathlib.Path(file_path).parts[-4] == "КОД":
                inn = pathlib.Path(file_path)[-5]
        else:
                inn = pathlib.Path(file_path)[-4]
        return inn  

Определение номера договора в документе

В одинаковом типе документации встречается разный паттерн номера, поэтому используем match‑case. Для регистронезависимого поиска к параметрам функции добавляется флаг re.IGNORECASE.

def get_contract_num(f_contents):
    contract_num = re.search(r'[(договор)|(кред)|(новации)]\s(ха|№)\s*(\w*/*\w*)\s*\n*', str(f_contents), re.IGNORECASE)
        contract_num = contract_num.group(2) if contract_num else "Не определён"    
        
        match contract_num:
                case re.findall(r'Кред_ООО', contract_num, re.IGNORECASE):
                        contract_num = re.search(r'парк_(\w*)_', f_contents, re.IGNORECASE)
                        contract_num = contract_num.group(1)
                case re.findall(r'(Кред_Лизинг)', contract_num, re.IGNORECASE):                         
                        contract_num = re.search(r'(Кред_Лизинг)_(\w*)[_|\s]*?', contract_num, re.IGNORECASE)
                contract_num = contract_num.group(2)
                case re.findall(r'(Кред_)', contract_num, re.IGNORECASE):
            contract_num = re.search(r'\w*(Кред_)_(\w*)[_|\s]*', contract_num, re.IGNORECASE)
                    contract_num = contract_num.group(2)
                ...
        return contract_num

Достаю ковенанты

Функция определяет пункты 8-й Статьи целиком (кусок от Статьи 8 до Статьи 9) — обязанности и права Заёмщика.

def find_coven_cond(text):
        #предварительная чистка от переносов строк, лишних спецсимволов
    cleaned_text = re.sub(r'\n|\t|■|\*|<|>|\`|\'|\‘|\’|^|\\|•|/|%|@|«|»|\?|\[|=|\+|&|°|„|“|\]|£|♦|\{|\}|™||•|▪|!|', '',str(text))
        #основная
    reg_text = re.findall(r'Стать[я|и]\s*8\s*[\.]*\s*Обязанности\s*и\s*права\s*принципала.*[Статья]?\s*9\.\s*[Ответственность\s*Сторон|Обеспечение]', cleaned_text, re.IGNORECASE)
    return reg_text

Дробление содержимого Статьи 8 на отдельные элементы - Пункт ковенанта | Текст ковенанта

Делю с помощью sub и findall — интересует Статья 8 ([8..() — 8..()]) Статья 9.

def split_coven_items(covens_data):
    split_items = re.findall(r'8[\.\,]\d[\.\,]\d{1,2}', str(covens_data))

    return split_items

def split_coven_values(covens_data):
    split_item_values = re.findall(r'(?:8[\.\,]\d[\.\,]\d{1,2})([\s\S]*?)(?=8[\.\,]\d[\.\,]\d{1,2}|Статья)', str(covens_data))

    for i in range(len(split_item_values)):
                #чистка от лишних символов в начале строки каждого элемента списка
        split_item_values[i] = re.sub(r'^[\s*\.\s*|\s]\s*\.*', '', split_item_values[i])
                #чистка от лишних символов в конце 
        split_item_values[i] = re.sub(r'\s[;|\.|,|-|\w]?\s$', '', split_item_values[i])

    return split_item_values

Создание output-файла

Наконец, формирую результирующий файл:

def get_output_file(data):
    if not pathlib.Path.is_dir('outputs'):
        pathlib.Path.mkdir('outputs')
    headers = ['Путь до файла', 'Номер договора', 'ИНН', 'Пункт', 'Текст ковенанта']
    df = pd.DataFrame(data, columns=headers)
    df.to_excel(f'outputs/{dt.now():%Y-%m-%d-%H%M%S_covidetect_results.xlsx}')

Output следующий: excel-файл со столбцами | Путь до файла | Номер договора | ИНН | Пункт ковенанта | Содержание ковенанта |. Если ковенанты условия не найдены в файле — в пункте и содержании ставится прочерк. Согласно вводным вся информация по изучаемой директории должна находиться в одном файле, поэтому путь, номер и ИНН могли повторяться.

Результаты

Ниже представлена блок‑схема моего пути по определению ковенантов в кредитной документации.

По поводу махинаций с «docx→pdf→txt» и почему docx‑файлы не были сразу переведены в текст: нумерованные списки реализованы в Word»е своеобразно и парсинг такой структуры — нетривиальная задача. При использовании docx2txt, pdfminer, pywin32, Aspose многие пункты не сохранялись/сохранялись некорректно (особенно не «1», «2», «3», а, например, «7.2.47»). Поскольку сохранение нумерации — одно из обязательных условий при извлечении ковенантов, привёл всё к pdf и далее собирал номера пунктов с помощью регулярок.

Модули pathlib, re, pandas хорошо показали себя в работе с «датой». Хорошо отработал и Abbyy Finereader — не идеально, конечно (пришлось добавлять дополнительные регулярки‑условия, чистить маленькие аномалии), но с учётом того, что структура страницы сложная, случаями качество изображения плохое и часть букв смазана, результат вполне себе хороший!

Результатом заказчик (DA) остался доволен — сохранено значительное время из‑за отсутствия ручного поиска ковенантов в большом объёме данных. Ну довольный заказчик — довольный разработчик.