Перейти к содержанию

Извлечение текста из документов

Шаг 1

В прошлом разделе мы построили безопасную границу перед обработкой файлов. И теперь у нас FileValidator проверяет путь, формат, размер и базовую структуру, а TemporaryFileStorage создаёт временную копию и удаляет её после завершения работы.

Но допустимый файл пока остаётся для приложения просто набором байтов. Чтобы искать сведения в договоре, составлять краткое содержание инструкции или передавать фрагменты документа модели, сначала нужно получить из файла обычный текст. Эту операцию будем называть извлечением текста. Она не требует LLM и не пытается понять смысл документа. Это обычный Python-код открывает файл подходящей библиотекой, читает доступный текст и сохраняет сведения о том, откуда взят каждый фрагмент.

И в этом разделе постепенно добавим поддержку TXT, Markdown, PDF с текстовым слоем и DOCX. Для каждого формата способ чтения отличается, но результат должен быть одинаковым. Поэтому начнём не с конкретной библиотеки, а с общего контракта.

Какой результат должен возвращать экстрактор?

Если вернуть из парсера только одну большую строку, мы быстро потеряем полезную структуру. В PDF важно знать номер страницы, а в Markdown и DOCX — заголовок раздела. Эти ссылки понадобятся, чтобы позднее показать пользователю источник ответа, а не просто выдать текст без происхождения.

Кроме самого текста нам нужны метаданные и диагностические предупреждения. Например, PDF может состоять из десяти страниц, но на двух страницах текстовый слой отсутствует. Это не всегда повод отклонять весь документ: полезнее извлечь остальные страницы и явно сообщить о пропуске.

Чтобы остальному приложению не приходилось работать с несколькими разрозненными значениями, результат извлечения объединим в один объект ExtractedDocument. Он будет содержать:

ExtractedDocument
├── text       — полный текст, собранный из всех найденных фрагментов
├── parts      — отдельные части документа со ссылками на страницы или разделы
├── metadata   — общие сведения о файле и результате извлечения
└── warnings   — обнаруженные проблемы, которые не помешали получить результат

Поле text удобно использовать, когда нужно обработать документ целиком. parts сохраняет его внутреннюю структуру и позволяет определить источник конкретного фрагмента. metadata сообщает формат, размер и количество извлечённых частей, а warnings предупреждает о частичных проблемах — например, о странице PDF без текстового слоя.

Для PDF одна часть будет соответствовать странице. Для Markdown и DOCX мы постараемся сохранить разделы. У обычного TXT надёжной внутренней структуры нет, поэтому весь файл станет одной частью с ссылкой document.

Перед практикой добавим две библиотеки, которые понадобятся нам далее:

pip install pypdf python-docx
pip freeze > requirements.txt

Откройте существующий пакет app/files и создайте в нём файл document_extraction.py. Пакет уже содержит __init__.py, поэтому новый файл сразу можно импортировать через app.files.

В начало файла добавьте:

from dataclasses import dataclass
from typing import Protocol
import unicodedata

from app.files.validation import FileKind, ValidatedFile

Тут у нас все стандартно, dataclass понадобится для объектов результата, а Protocol опишет единый интерфейс всех экстракторов, а функция из unicodedata поможет осторожно удалить непечатаемые служебные символы.

Теперь добавьте общую иерархию ошибок:

class DocumentExtractionError(RuntimeError):
    pass

class UnsupportedDocumentFormatError(DocumentExtractionError):
    pass

class DocumentReadError(DocumentExtractionError):
    pass

class EmptyDocumentError(DocumentExtractionError):
    pass

Тут DocumentExtractionError станет общей ошибкой нового контура. Через неё демонстрационный запуск, а позднее и сервис приложения, смогут перехватить ожидаемую проблему одним блоком except.

Ошибка UnsupportedDocumentFormatError означает, что безопасный файл относится к формату, для которого пока нет текстового экстрактора. Это возможно, потому что валидатор уже разрешает CSV, Excel и изображения, а разбирать их мы будем в следующих разделах.

А DocumentReadError сообщает, что подходящий парсер не смог прочитать содержимое. EmptyDocumentError используется, когда файл технически открылся, но после извлечения в нём не осталось пригодного текста.

Далее опишите одну часть документа:

@dataclass(frozen=True)
class DocumentPart:
    reference: str
    text: str
    page_number: int | None = None
    section_title: str | None = None
    heading_level: int | None = None

Давайте подробнее разберём поля DocumentPart, чтобы было понятно, какие сведения хранит каждый извлечённый фрагмент.

reference — внутренняя ссылка на место внутри документа. Например значение page:3 означает третью страницу PDF, а section:2 — второй найденный раздел Markdown- или DOCX-документа. Это понятный идентификатор источника и позднее приложение сможет сохранить его вместе с фрагментом и объяснить пользователю, откуда была взята информация.

Поле text содержит непосредственно извлечённый текст страницы или раздела.

page_number хранит номер страницы и заполняется только для форматов с надёжной постраничной структурой, например PDF. Для TXT, Markdown и DOCX это поле будет равно None.

section_title содержит название раздела, например «Условия возврата». Оно заполняется, когда экстрактор смог определить заголовок документа. Для обычного TXT или страницы PDF такого заголовка может не быть.

heading_level показывает уровень найденного заголовка. Например, 1 соответствует основному разделу, а 2 — вложенному подразделу. Это позволяет не только сохранить название, но и частично восстановить иерархию документа.

Поля page_number, section_title и heading_level допускают None, потому что разные форматы предоставляют разную структуру. Мы не будем придумывать недостающие сведения: экстрактор сохранит только то, что действительно удалось определить.

Следом добавьте метаданные и итоговый результат:

@dataclass(frozen=True)
class DocumentMetadata:
    original_name: str
    kind: FileKind
    media_type: str
    size_bytes: int
    part_count: int
    page_count: int | None
    character_count: int

@dataclass(frozen=True)
class ExtractedDocument:
    text: str
    parts: tuple[DocumentPart, ...]
    metadata: DocumentMetadata
    warnings: tuple[str, ...]

DocumentMetadata содержит только сведения, которые одинаково понятны приложению независимо от библиотеки. Мы не возвращаем наружу внутренний объект PdfReader или Document из python-docx. Благодаря этому остальной код не будет зависеть от устройства конкретного парсера.

Кортежи в parts и warnings подчёркивают, что готовый результат не должен неожиданно изменяться после возврата из экстрактора. Сам ExtractedDocument тоже объявлен как frozen=True.

Теперь зафиксируем единый интерфейс конкретного обработчика формата:

class DocumentFormatExtractor(Protocol):
    @property
    def supported_kinds(self) -> frozenset[FileKind]:
        ...

    def extract(self, file: ValidatedFile) -> ExtractedDocument:
        ...

Напомню, что это Protocol, а не базовый класс с готовой реализацией. Он говорит что любой объект считается экстрактором формата, если сообщает поддерживаемые FileKind и имеет метод extract() с нужной сигнатурой. Точки внутри методов здесь правильные. Мы не создаём экземпляр DocumentFormatExtractor и не выполняем эти методы. Протокол нужен для проверки типов и для единого маршрутизатора, который появится в последнем шаге этого раздела.

Разные парсеры могут вернуть разные переводы строк, неразрывные пробелы, маркер кодировки в начале файла и невидимые управляющие символы. Если ничего с ними не делать, два визуально одинаковых текста могут отличаться внутри Python, а в строках появятся лишние разрывы.

При этом очистка должна быть осторожной. Нельзя превращать все пробелы в один: отступы могут быть частью Markdown-кода, а переводы строк разделяют абзацы. Поэтому будем убирать только явно служебные символы, пробелы в конце строк и цепочки пустых строк.

Контракт результата уже описан. Теперь в том же файле подготовим общую функцию очистки, которую смогут вызывать все будущие экстракторы. После протокола добавьте:

def normalize_extracted_text(value: str) -> str:
    normalized = value.replace("\ufeff", "")
    normalized = normalized.replace("\r\n", "\n")
    normalized = normalized.replace("\r", "\n")
    normalized = normalized.replace("\u00a0", " ")

    visible_characters = (
        character
        for character in normalized
        if character in {"\n", "\t"}
        or not unicodedata.category(character).startswith("C")
    )
    normalized = "".join(visible_characters)

    result_lines: list[str] = []
    previous_line_was_empty = False

    for line in normalized.split("\n"):
        cleaned_line = line.rstrip()
        line_is_empty = not cleaned_line.strip()

        if line_is_empty and previous_line_was_empty:
            continue

        result_lines.append("" if line_is_empty else cleaned_line)
        previous_line_was_empty = line_is_empty

    return "\n".join(result_lines).strip()

Первые четыре строки приводят разные варианты перевода строк к \n, удаляют BOM \ufeff и заменяют неразрывный пробел обычным. Затем генератор оставляет печатаемые символы, а также табуляцию и перевод строки. Остальные управляющие символы не попадут в результат.

Здесь важно, что функция принимает и возвращает обычную строку. Она ничего не знает о PDF, DOCX или Markdown и не открывает файлы. Конкретный экстрактор сначала получает текст своим способом, а затем передаёт его в normalize_extracted_text(). Благодаря этому одни и те же правила очистки не будут копироваться по нескольким модулям.

Цикл обрабатывает текст построчно. rstrip() удаляет только пробелы справа и не ломает отступ слева. Переменная previous_line_was_empty не позволяет создавать длинные последовательности пустых строк, но сохраняет один разрыв между абзацами.

На каждой итерации line_is_empty отвечает на простой вопрос: остались ли в строке видимые символы. Если текущая и предыдущая строки пусты, выполняется continue, и второй пустой разрыв не добавляется. Для непустой строки сохраняется исходный левый отступ, а для первой пустой строки добавляется пустая строка. В самом конце strip() убирает разрывы только по краям всего документа.

В этом же файле добавьте общую сборку результата:

def build_extracted_document(
    file: ValidatedFile,
    parts: tuple[DocumentPart, ...],
    warnings: tuple[str, ...] = (),
    page_count: int | None = None,
) -> ExtractedDocument:
    useful_parts = tuple(
        part
        for part in parts
        if part.text or part.section_title
    )

    rendered_parts: list[str] = []

    for part in useful_parts:
        values: list[str] = []
        if part.section_title is not None:
            values.append(part.section_title)
        if part.text:
            values.append(part.text)
        rendered_parts.append("\n\n".join(values))

    full_text = "\n\n".join(rendered_parts).strip()

    if not full_text:
        raise EmptyDocumentError(
            "В документе не найден пригодный текст."
        )

    metadata = DocumentMetadata(
        original_name=file.original_name,
        kind=file.kind,
        media_type=file.media_type,
        size_bytes=file.size_bytes,
        part_count=len(useful_parts),
        page_count=page_count,
        character_count=len(full_text),
    )

    return ExtractedDocument(
        text=full_text,
        parts=useful_parts,
        metadata=metadata,
        warnings=tuple(dict.fromkeys(warnings)),
    )

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

Параметр parts уже содержит найденную структуру документа. PDF позднее передаст сюда страницы, Markdown — разделы, а TXT — одну часть. Функция не пытается заново определить эту структуру. Она только проверяет готовые части, объединяет их в полный текст и дополняет результат общими сведениями о файле.

Сначала отбрасываются полностью пустые части. Затем для каждой части объединяются заголовок и содержимое. Если итоговый текст пуст, функция останавливает обработку через EmptyDocumentError. В конце dict.fromkeys() удаляет повторяющиеся предупреждения, сохраняя их первоначальный порядок.

Обратите внимание, что full_text не заменяет parts. Это два представления одних данных для разных задач. Полная строка удобна для последовательной обработки всего документа, а части позволяют позднее показать, на какой странице или в каком разделе находился использованный фрагмент.

Общая часть готова, поэтому можно написать первый настоящий обработчик. Создайте рядом файл app/files/text_extraction.py. В нём будут два самых простых формата, которые Python умеет читать без стороннего парсера.

Добавьте импорты:

import re

from app.files.document_extraction import (
    DocumentPart,
    DocumentReadError,
    ExtractedDocument,
    UnsupportedDocumentFormatError,
    build_extracted_document,
    normalize_extracted_text,
)
from app.files.validation import FileKind, ValidatedFile

Модуль re понадобится только для заголовков Markdown. Остальные объекты мы только что создали в общем контракте.

Ниже добавьте класс:

class TextDocumentExtractor:
    supported_kinds = frozenset(
        {FileKind.TEXT, FileKind.MARKDOWN}
    )

    _heading_pattern = re.compile(
        r"^(#{1,6})[ \t]+(.+?)\s*$"
    )

    def extract(self, file: ValidatedFile) -> ExtractedDocument:
        if file.kind not in self.supported_kinds:
            raise UnsupportedDocumentFormatError(
                f"TextDocumentExtractor не поддерживает {file.kind}."
            )

        try:
            raw_text = file.path.read_text(encoding="utf-8")
        except (OSError, UnicodeError) as error:
            raise DocumentReadError(
                "Не удалось прочитать текстовый документ."
            ) from error

        text = normalize_extracted_text(raw_text)

        if file.kind == FileKind.TEXT:
            parts = (
                DocumentPart(
                    reference="document",
                    text=text,
                ),
            )
        else:
            parts = self._split_markdown(text)

        return build_extracted_document(file, parts)

Один экземпляр поддерживает сразу TXT и Markdown, потому что оба формата читаются как UTF-8-текст. Разница начинается после чтения: TXT становится одной частью, а Markdown дополнительно делится по заголовкам.

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

Проверка file.kind защищает класс от неправильного использования. Обычно формат выберет будущий маршрутизатор, но сам обработчик всё равно явно фиксирует свою границу.

read_text(encoding="utf-8") читает содержимое. Формат и кодировку мы уже проверили в FileValidator, однако файл мог исчезнуть или измениться. Поэтому ошибки файловой системы и декодирования преобразуются в понятный DocumentReadError.

Для TXT создаётся кортеж всего из одного DocumentPart. Ссылка document означает, что точнее указать расположение текста невозможно. У Markdown вызывается отдельный метод, потому что перед созданием результата нужно найти границы разделов.

Регулярное выражение считает заголовком строку с одним–шестью символами #, пробелом и названием. Например, ## Возврат товара станет заголовком второго уровня.

Чтобы закончить класс, добавьте внутренний метод:

    def _split_markdown(
        self,
        text: str,
    ) -> tuple[DocumentPart, ...]:
        parts: list[DocumentPart] = []
        body_lines: list[str] = []
        current_reference = "document"
        current_title: str | None = None
        current_level: int | None = None
        section_number = 0

        def flush_part() -> None:
            body = normalize_extracted_text(
                "\n".join(body_lines)
            )
            if not body and current_title is None:
                return

            parts.append(
                DocumentPart(
                    reference=current_reference,
                    text=body,
                    section_title=current_title,
                    heading_level=current_level,
                )
            )
            body_lines.clear()

        for line in text.split("\n"):
            match = self._heading_pattern.match(line)
            if match is None:
                body_lines.append(line)
                continue

            flush_part()
            section_number += 1
            current_reference = f"section:{section_number}"
            current_level = len(match.group(1))
            current_title = match.group(2).strip()

        flush_part()
        return tuple(parts)

Метод проходит строки сверху вниз и временно складывает содержимое текущего раздела в body_lines. Когда встречается новый заголовок, вложенная функция flush_part() сохраняет предыдущий раздел и очищает буфер.

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

Текст до первого заголовка получает ссылку document. Отдельный счётчик section_number увеличивается только при встрече заголовка, поэтому разделы получают section:1, section:2 и так далее, даже если перед ними был вводный текст. Уровень решёток хранится отдельно, поэтому позднее можно отличить основной раздел от подраздела.

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

Давайте подведем итоги этого шага, в нем мы создали общий результат извлечения и первый обработчик формата. В следующих двух шагах добавим PDF- и DOCX-экстракторы, которые будут возвращать те же ExtractedDocument, DocumentPart и DocumentMetadata. После этого соберём все реализации в единый DocumentExtractor и проверим их одним демонстрационным запуском.

Шаг 2

Для TXT и Markdown источник текста очевиден, ведь там символы находятся внутри файла в почти готовом виде. Но PDF файлы устроены сложнее, ведь там описываются расположение элементов на странице, а не логические абзацы документа.

Один PDF может содержать обычный текстовый слой, другой только отсканированные изображения, а третий смесь обоих вариантов. В этом шаге мы научимся работать только с PDF с текстовым слоем. Распознавание текста на изображениях относится к OCR. В разделе 4.6 мы кратко сравним OCR с прямым анализом изображения vision-моделью, но отдельный OCR-маршрут создавать не будем.

При этом важно не путать внешний вид и доступность текста. Пользователь может видеть на странице слова, но для библиотеки это будет одно изображение. И наоборот, у визуально сложного PDF может существовать нормальный текстовый слой, который извлекается без модели и без OCR.

Какие ограничения нужны PDF-парсеру?

В прошлом разделе мы уже ограничили размер файла на диске. Для PDF этого недостаточно, ведь небольшой сжатый поток внутри документа при распаковке способен занять значительно больше памяти. Кроме того, документ может содержать слишком много страниц для одного запроса.

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

Откройте app/files/document_extraction.py. После EmptyDocumentError добавьте ещё две ошибки:

class EncryptedDocumentError(DocumentExtractionError):
    pass

class DocumentLimitError(DocumentExtractionError):
    pass

EncryptedDocumentError будет означать, что PDF защищён паролем. Мы не будем запрашивать пароль и хранить его внутри приложения, поэтому такой файл отклоняется явно. DocumentLimitError сообщает, что сам документ допустим по размеру, но его внутренняя структура превышает лимит парсера.

Теперь в пакете app/files создайте файл pdf_extraction.py и добавьте импорты:

from dataclasses import dataclass
from typing import Any

from pypdf import PdfReader
from pypdf.errors import PdfReadError

from app.files.document_extraction import (
    DocumentLimitError,
    DocumentPart,
    DocumentReadError,
    EmptyDocumentError,
    EncryptedDocumentError,
    ExtractedDocument,
    UnsupportedDocumentFormatError,
    build_extracted_document,
    normalize_extracted_text,
)
from app.files.validation import FileKind, ValidatedFile

PdfReader открывает структуру PDF и предоставляет коллекцию страниц. PdfReadError — исключение самой библиотеки для повреждённых или неподдерживаемых данных. Наружу из нашего модуля оно не выйдет: преобразуем его в общую ошибку DocumentReadError.

Сначала опишем настройки извлечения:

@dataclass(frozen=True)
class PdfExtractionPolicy:
    max_pages: int = 200
    max_page_content_bytes: int = 5 * 1_024 * 1_024

    def __post_init__(self) -> None:
        if self.max_pages <= 0:
            raise ValueError("max_pages должен быть больше нуля.")
        if self.max_page_content_bytes <= 0:
            raise ValueError(
                "max_page_content_bytes должен быть больше нуля."
            )

В этом классе max_pages ограничивает количество страниц, а max_page_content_bytes объём распакованного потока одной страницы. Обратите внимание, в реальном проекте именно эти значения нужно подобрать по доступной памяти, времени обработки и типичным документам пользователей, но так как мы пишем учебный проект, нам этих значений будет достаточно.

Теперь добавьте класс, который применит эти ограничения и извлечёт текст каждой страницы:

class PdfDocumentExtractor:
    supported_kinds = frozenset({FileKind.PDF})

    def __init__(
        self,
        policy: PdfExtractionPolicy | None = None,
    ) -> None:
        self._policy = policy or PdfExtractionPolicy()

    def extract(self, file: ValidatedFile) -> ExtractedDocument:
        if file.kind not in self.supported_kinds:
            raise UnsupportedDocumentFormatError(
                f"PdfDocumentExtractor не поддерживает {file.kind}."
            )

        reader = self._open_reader(file)

        try:
            if reader.is_encrypted:
                raise EncryptedDocumentError(
                    "Зашифрованные PDF пока не поддерживаются."
                )

            page_count = len(reader.pages)
            if page_count > self._policy.max_pages:
                raise DocumentLimitError(
                    "PDF содержит слишком много страниц."
                )

            parts: list[DocumentPart] = []
            warnings: list[str] = []

            for index, page in enumerate(reader.pages):
                page_number = index + 1
                text = self._extract_page(page, page_number)

                if not text:
                    warnings.append(
                        f"Страница {page_number}: "
                        "текстовый слой не найден."
                    )
                    continue

                parts.append(
                    DocumentPart(
                        reference=f"page:{page_number}",
                        text=text,
                        page_number=page_number,
                    )
                )

            if not parts:
                raise EmptyDocumentError(
                    "В PDF не найден текстовый слой. "
                    "Вероятно, документ требует OCR."
                )

            return build_extracted_document(
                file=file,
                parts=tuple(parts),
                warnings=tuple(warnings),
                page_count=page_count,
            )
        finally:
            reader.close()

Конструктор принимает политику извлечения PDF файлов, но при создании PdfDocumentExtractor сетевых запросов и чтения файлов ещё не происходит. Объект только запоминает правила, которые будет использовать при каждом вызове extract(), а обработка начинается лишь тогда, когда в метод передан готовый ValidatedFile.

Метод extract() сначала проверяет вид файла, а затем открывает PdfReader. Зашифрованный документ останавливается до чтения страниц. После этого проверяется общее количество страниц.

Далее создаются два пустых списка. В parts будут попадать только страницы с найденным текстом, а в warnings — сообщения о страницах, которые удалось открыть, но из которых нечего извлекать. Такое разделение позволяет не смешивать полезные данные с диагностикой.

enumerate(reader.pages) возвращает индексы с нуля, но человеку привычнее страницы с единицы. Поэтому page_number = index + 1. То же число попадает в поле page_number и в ссылку page:1, page:2 и так далее.

На каждой итерации основной метод передаёт отдельную страницу в _extract_page(). Внутренний метод вернёт уже очищенную строку. Если строка пустая, цикл добавит предупреждение и перейдёт к следующей странице. Если текст есть, создаётся DocumentPart, где сам текст хранится вместе с номером и ссылкой на исходное место.

Пустая страница не завершает весь документ ошибкой. Мы добавляем предупреждение и продолжаем цикл. Если текст найден хотя бы на одной странице, вызывающий код получит полезный результат вместе с информацией о пропусках. Если пустыми оказались все страницы, возвращать нечего, поэтому создаётся EmptyDocumentError с подсказкой про OCR.

Только после завершения всего цикла вызывается build_extracted_document(). Именно здесь список страниц превращается в неизменяемый кортеж, собирается полный текст и рассчитываются общие метаданные. Таким образом, частично прочитанный документ не возвращается раньше времени.

Основной метод вызывает два внутренних помощника, которых в классе пока нет. Сначала после extract() добавьте открытие PDF:

    @staticmethod
    def _open_reader(file: ValidatedFile) -> PdfReader:
        try:
            return PdfReader(file.path, strict=False)
        except (OSError, PdfReadError) as error:
            raise DocumentReadError(
                "Не удалось открыть PDF."
            ) from error

В этом коде PdfReader(file.path, strict=False) создаёт объект для чтения PDF-файла, где file.path это путь к файлу, а strict=False означает, что библиотека будет нестрого проверять структуру PDF и постарается прочитать файл даже при небольших нарушениях или ошибках.

Этот метод специально получился коротким. Успешный путь возвращает PdfReader, а ожидаемая ошибка библиотеки сразу переводится на язык нашего приложения. Поэтому в большом extract() остаётся логика документа, а технические подробности открытия файла находятся отдельно.

Ниже добавьте чтение одной страницы:

    def _extract_page(
        self,
        page: Any,
        page_number: int,
    ) -> str:
        try:
            contents = page.get_contents()
            if contents is not None:
                content_size = len(contents.get_data())
                if (
                    content_size
                    > self._policy.max_page_content_bytes
                ):
                    raise DocumentLimitError(
                        f"Страница {page_number} превышает "
                        "лимит распакованного содержимого."
                    )

            raw_text = page.extract_text() or ""
        except DocumentLimitError:
            raise
        except (
            OSError,
            PdfReadError,
            KeyError,
            TypeError,
            ValueError,
        ) as error:
            raise DocumentReadError(
                f"Не удалось извлечь страницу {page_number}."
            ) from error

        return normalize_extracted_text(raw_text)

В данном коде параметр page специально имеет тип Any, чтобы не зависеть от внутренних типов pypdf, которые могут измениться в новых версиях библиотеки.

page.get_contents() получает поток команд страницы, а get_data() распаковывает его. Мы измеряем результат и отклоняем страницу, если она превысила лимит, ведь можем упереться в лимит памяти.

После проверки объёма вызывается page.extract_text(). Если библиотека вернула None, выражение or "" заменит его пустой строкой. Благодаря этому остальной код работает только со строками и не проверяет None отдельно.

page.extract_text() возвращает строку или None. Выражение or "" приводит отсутствие текста к пустой строке, после чего общий normalize_extracted_text() удаляет служебные символы и лишние пустые строки.

В итоге вызов проходит по понятной цепочке: extract() управляет всем документом, _open_reader() отвечает только за открытие, а _extract_page() только за одну страницу. И теперь наш PDF-экстрактор готов и возвращает тот же ExtractedDocument, что и обработчик TXT/Markdown. В следующем шаге мы разберём DOCX и добавим его в наш проект.

Шаг 3

В PDF мы опирались на страницы, потому что они являются частью самого формата. Но вот DOCX хранит документ иначе, а именно внутри архива находятся XML-файлы с абзацами, таблицами, стилями и другими элементами, а окончательная разбивка на страницы рассчитывается редактором при отображении.

Один и тот же DOCX может занять разное количество страниц при замене шрифта, полей, версии Word или настроек страницы. Поэтому мы не должны ориентироваться на страницы, а вместо них нам нужно сохранить более надёжную логическую структуру:

Для чтения используем установленную в первой лекции библиотеку python-docx, и давайте обсудим а как определить заголовок документа. На самом деле все довольно просто, встроенные стили Word называются Heading 1, Heading 2 и так далее.

Именно стиль сообщает о роли абзаца, а не только о его внешнем виде:

Heading 1 → основной раздел
Heading 2 → подраздел
Normal    → обычный абзац

Даже если Word показывает локализованное название стиля, python-docx обычно обращается к встроенному стилю по английскому имени. Пользовательские стили могут называться иначе, поэтому первая версия экстрактора распознает только стандартные Heading 1–Heading 9. Обычный текст при этом не потеряется, он просто останется в текущем разделе.

Давайте перейдем к практике и в пакете app/files создайте файл docx_extraction.py и добавьте импорты:

import re
from zipfile import BadZipFile

from docx import Document
from docx.document import Document as DocumentObject
from docx.opc.exceptions import PackageNotFoundError
from docx.table import Table
from docx.text.paragraph import Paragraph

from app.files.document_extraction import (
    DocumentPart,
    DocumentReadError,
    EmptyDocumentError,
    ExtractedDocument,
    UnsupportedDocumentFormatError,
    build_extracted_document,
    normalize_extracted_text,
)
from app.files.validation import FileKind, ValidatedFile

Давайте кратко разберем эти импорты. Document открывает DOCX, а классы Paragraph и Table помогут отличить два вида элементов во время обхода. PackageNotFoundError и BadZipFile относятся к повреждённому или неподходящему Office-архиву.

Ниже добавьте класс и его основной метод:

class DocxDocumentExtractor:
    supported_kinds = frozenset({FileKind.DOCX})

    _heading_pattern = re.compile(r"^Heading ([1-9])$")

    def extract(self, file: ValidatedFile) -> ExtractedDocument:
        if file.kind not in self.supported_kinds:
            raise UnsupportedDocumentFormatError(
                f"DocxDocumentExtractor не поддерживает {file.kind}."
            )

        try:
            document = Document(file.path)
            parts, table_found = self._extract_parts(document)
        except EmptyDocumentError:
            raise
        except (
            OSError,
            BadZipFile,
            PackageNotFoundError,
            KeyError,
            ValueError,
        ) as error:
            raise DocumentReadError(
                "Не удалось прочитать DOCX."
            ) from error

        warnings = [
            "Для DOCX номера страниц не определяются: "
            "сохранены ссылки на разделы."
        ]
        if table_found:
            warnings.append(
                "Таблицы DOCX преобразованы в обычный текст."
            )

        return build_extracted_document(
            file=file,
            parts=parts,
            warnings=tuple(warnings),
        )

Как и остальные обработчики, класс явно сообщает поддерживаемый FileKind и отклоняет неправильный формат. Затем Document(file.path) открывает проверенную временную копию.

Основной метод здесь намеренно не разбирает каждый абзац самостоятельно. Он управляет общим сценарием, проверяет формат, открывает документ, получает части из внутреннего метода, формирует предупреждения и собирает единый результат. Благодаря этому по extract() сразу видно, что происходит с документом целиком.

Метод _extract_parts(), который добавим ниже, вернёт готовые части и флаг table_found. Флаг нужен не для бизнес-логики, а для диагностики, ведь если в исходном документе встретилась таблица, пользователь должен знать, что её ячейки были превращены в строки и исходное оформление не сохранилось.

Переменная parts содержит не сырой объект библиотеки, а наши DocumentPart. Поэтому после завершения внутреннего метода остальному приложению уже не нужно знать, как python-docx представляет абзацы и таблицы.

Теперь реализуем недостающий разбор содержимого. Напишем метод для DocxDocumentExtractor и после extract() добавьте:

    def _extract_parts(
        self,
        document: DocumentObject,
    ) -> tuple[tuple[DocumentPart, ...], bool]:
        parts: list[DocumentPart] = []
        body_blocks: list[str] = []
        current_reference = "document"
        current_title: str | None = None
        current_level: int | None = None
        section_number = 0
        table_found = False

        def flush_part() -> None:
            body = normalize_extracted_text(
                "\n\n".join(body_blocks)
            )
            if not body and current_title is None:
                return

            parts.append(
                DocumentPart(
                    reference=current_reference,
                    text=body,
                    section_title=current_title,
                    heading_level=current_level,
                )
            )
            body_blocks.clear()

        for block in document.iter_inner_content():
            if isinstance(block, Paragraph):
                text = normalize_extracted_text(block.text)
                if not text:
                    continue

                heading_level = self._get_heading_level(block)
                if heading_level is None:
                    body_blocks.append(text)
                    continue

                flush_part()
                section_number += 1
                current_reference = f"section:{section_number}"
                current_title = text
                current_level = heading_level
                continue

            if isinstance(block, Table):
                table_text = self._table_to_text(block)
                if table_text:
                    body_blocks.append(table_text)
                    table_found = True

        flush_part()

        if not parts:
            raise EmptyDocumentError(
                "В DOCX не найден пригодный текст."
            )

        return tuple(parts), table_found

Метод document.iter_inner_content() идёт по основным абзацам и таблицам в том порядке, в котором они расположены в документе.

Перед началом обхода создаются переменные состояния. parts хранит уже завершённые разделы, body_blocks уже текст раздела, который собирается сейчас. Остальные переменные запоминают ссылку, заголовок и его уровень. То есть в каждый момент метод знает две вещи, а именно что уже готово и что пока только накапливается.

Переменные current_reference, current_title и current_level описывают раздел, который собирается сейчас. До первого заголовка используется ссылка document. После первого заголовка появится section:1, затем section:2 и так далее.

body_blocks это временный буфер содержимого текущего раздела. В него попадают обычные абзацы и текстовое представление таблиц. Вложенная функция flush_part() объединяет накопленные блоки, создаёт DocumentPart и очищает буфер.

Внутри flush_part() сначала выполняется общая очистка текста. Если нет ни содержимого, ни заголовка, сохранять нечего, и функция завершает работу через return. В остальных случаях она использует текущее состояние для создания одной части. После body_blocks.clear() тот же список готов накапливать следующий раздел.

Когда встречается заголовок, сначала сохраняется предыдущий раздел. Только после этого меняются ссылка, название и уровень. Поэтому заголовок всегда относится к тексту, который идёт после него, а не к уже прочитанным абзацам.

Обычный абзац проходит более короткий путь. Его текст очищается, определяется стиль, и при отсутствии заголовочного уровня строка добавляется в body_blocks. Команда continue сразу переводит цикл к следующему элементу и не позволяет обработать тот же абзац повторно.

Для таблицы вызывается _table_to_text(). Если после преобразования остался текст, он добавляется в тот же буфер на том месте, где таблица находилась в документе. Одновременно table_found становится True, чтобы основной метод позднее добавил предупреждение.

Если в документе нет ни одного стандартного заголовка, все абзацы и таблицы окажутся в одной части document. Такой файл по-прежнему можно обработать.

А после окончания цикла снова вызывается flush_part(). Без этого последняя накопленная часть потерялась бы, потому что после неё уже нет следующего заголовка, который мог бы запустить сохранение. Затем метод убеждается, что хотя бы одна часть создана, и возвращает неизменяемый кортеж вместе с флагом таблицы.

Этот основной обход уже показывает последовательность документа, но использует два ещё не написанных помощника. Они изолируют небольшие операции и не перегружают цикл дополнительными условиями. Сначала добавьте метод определения уровня заголовка:

    def _get_heading_level(
        self,
        paragraph: Paragraph,
    ) -> int | None:
        style = paragraph.style
        if style is None:
            return None

        match = self._heading_pattern.fullmatch(style.name)
        if match is None:
            return None

        return int(match.group(1))

У абзаца может не быть доступного стиля, поэтому сначала проверяется None. Затем регулярное выражение ищет точное имя Heading N. Если совпадения нет, метод возвращает None, и абзац рассматривается как обычный текст.

Возвращаемое значение одновременно служит ответом и для проверки, и для сохранения структуры. None означает обычный абзац, а число от 1 до 9 это найденный уровень заголовка. Основному циклу не нужно повторно читать имя стиля.

Если стиль найден, первая группа выражения содержит цифру и int() превратит её в уровень заголовка, который попадёт в DocumentPart.heading_level.

После этого добавьте преобразование таблицы:

    @staticmethod
    def _table_to_text(table: Table) -> str:
        result_rows: list[str] = []

        for row in table.rows:
            cells = []
            for cell in row.cells:
                cell_text = normalize_extracted_text(cell.text)
                cells.append(cell_text.replace("\n", " / "))

            if any(cells):
                result_rows.append(" | ".join(cells))

        return "\n".join(result_rows)

Мы обходим строки сверху вниз и читаем текст каждой ячейки. Внутренние переводы строк заменяются на / , чтобы одна строка таблицы не распалась на несколько неотличимых абзацев, а ячейки разделяются символом |.

Список cells создаётся заново для каждой строки. После очистки ячеек условие any(cells) отбрасывает полностью пустые строки. Непустые значения соединяются через | и добавляются в result_rows. В конце строки таблицы соединяются переводами строк и возвращаются в основной обход как один текстовый блок.

Например, таблица:

Товар       Срок возврата
Кофемашина  14 дней

превратится в:

Товар | Срок возврата
Кофемашина | 14 дней

Конечно это не полноценная табличная модель, ведь объединённые ячейки и сложное оформление могут выглядеть неоднозначно. Но в следующем разделе для CSV и Excel мы создадим отдельное представление строк и колонок. А здесь таблица является лишь частью текстового документа, поэтому сохраняем её читаемое текстовое содержимое и выдаём предупреждение.

И если подводить итоги трех шагов этого раздела, то мы можем сказать что у нас есть три реализации с разным внутренним устройством:

Все они возвращают одинаковый объект, и в последнем шаге этого раздела добавим единый маршрутизатор, который выберет нужную реализацию по ValidatedFile.kind, а затем запустим полную цепочку для различных документов.

Шаг 4

Мы написали три обработчика, но ведь вызывающему коду не следует самостоятельно выбирать между ними через длинную цепочку if/elif. То есть консоль или будущий сервис(когда у нас появится бэкенд) должны передать один ValidatedFile и получить один ExtractedDocument, не разбираясь в особенностях каждой библиотеки.

Для этого создадим единый Document Extractor. Он не извлекает PDF или DOCX самостоятельно. Его задача проще, ему нужно посмотреть на уже подтверждённый FileKind, найти зарегистрированный обработчик и передать ему файл. Давайте посмотрим на схему:

Такая логика позволит позднее добавить новый формат без изменения кода, который пользуется экстрактором, ведь нам будет достаточно написать ещё один обработчик с тем же протоколом и зарегистрировать его при сборке.

Перейдём к сборке этой схемы в коде. Откройте app/files/document_extraction.py и в самом конце файла, после build_extracted_document(), добавьте класс:

class DocumentExtractor:
    def __init__(
        self,
        extractors: tuple[DocumentFormatExtractor, ...],
    ) -> None:
        if not extractors:
            raise ValueError(
                "Нужен хотя бы один экстрактор формата."
            )

        registry: dict[FileKind, DocumentFormatExtractor] = {}

        for extractor in extractors:
            for kind in extractor.supported_kinds:
                if kind in registry:
                    raise ValueError(
                        f"Для формата {kind} уже есть экстрактор."
                    )
                registry[kind] = extractor

        self._registry = registry

    def extract(self, file: ValidatedFile) -> ExtractedDocument:
        extractor = self._registry.get(file.kind)
        if extractor is None:
            raise UnsupportedDocumentFormatError(
                f"Извлечение из формата {file.kind} "
                "пока не поддерживается."
            )

        return extractor.extract(file)

Конструктор получает готовые обработчики, а не создаёт их внутри себя. И при создании объекта конструктор один раз собирает реестр, например стандартный набор превратится в словарь, где FileKind.PDF указывает на PdfDocumentExtractor, а FileKind.TEXT и FileKind.MARKDOWN на один TextDocumentExtractor. Во время обработки документа этот словарь уже не перестраивается.

Временный словарь registry связывает каждый FileKind с объектом, который умеет его читать. TextDocumentExtractor зарегистрируется два раза — для TXT и Markdown, — но это всё тот же экземпляр.

Внешний цикл берёт очередной объект экстрактора, а внутренний проходит все заявленные им форматы. Поэтому маршрутизатор не содержит знаний о том, сколько форматов поддерживает конкретный класс. Он работает только с контрактом DocumentFormatExtractor.

Проверка дубликата не позволяет двум обработчикам незаметно объявить один формат. Иначе результат зависел бы от порядка объектов в кортеже, а ошибка конфигурации осталась бы скрытой.

Сам метод extract() состоит из двух действий, а именно ищет обработчик и делегирует ему работу. Если формат не зарегистрирован, создаётся понятный UnsupportedDocumentFormatError.

Самое важное то, что маршрутизатор не проверяет расширение повторно и не открывает файл. Эту работу уже выполнил FileValidator, а чтением занимается выбранный обработчик. DocumentExtractor отвечает только за правильное направление вызова.

Чтобы сборка стандартного набора не повторялась в разных точках входа, создайте файл app/files/default_document_extractor.py:

from app.files.document_extraction import DocumentExtractor
from app.files.docx_extraction import DocxDocumentExtractor
from app.files.pdf_extraction import (
    PdfDocumentExtractor,
    PdfExtractionPolicy,
)
from app.files.text_extraction import TextDocumentExtractor

def build_default_document_extractor(
    pdf_policy: PdfExtractionPolicy | None = None,
) -> DocumentExtractor:
    return DocumentExtractor(
        extractors=(
            TextDocumentExtractor(),
            PdfDocumentExtractor(policy=pdf_policy),
            DocxDocumentExtractor(),
        )
    )

Функция отвечает только за стандартную композицию объектов. Сам DocumentExtractor ничего не знает о конкретных классах, а конкретные классы ничего не знают друг о друге.

Когда вызывается build_default_document_extractor(), Python последовательно создаёт три конкретных обработчика, помещает их в кортеж и передаёт маршрутизатору. Необязательная pdf_policy сразу попадает только в PDF-экстрактор. То есть уникальные настройки одного формата не затрагивают остальные.

Подготавливаем учебные документы

Для наглядной проверки продолжим использовать каталог sample_files из прошлого раздела. __init__.py в нём по-прежнему не нужен, потому что это данные, а не Python-код.

Оставьте существующий valid.txt и добавьте три документа:

sample_files/
├── valid.txt
├── return_policy.md
├── return_policy.docx
└── return_policy.pdf

В return_policy.md сохраните:

# Возврат товара

Покупатель может обратиться в поддержку в течение 14 дней.

## Повреждённый товар

Укажите номер заказа и приложите фотографию повреждения.

## Исправный товар

Товар должен сохранить комплектность и внешний вид.

DOCX удобнее создать в Word или Pages. Добавьте заголовок «Возврат товара» со стилем Заголовок 1, два подзаголовка со стилем Заголовок 2, несколько обычных абзацев и небольшую таблицу со столбцами «Ситуация» и «Срок».

Важно назначить именно стили заголовков, а не только сделать текст крупным и жирным. Так мы проверим, что экстрактор читает логическую структуру документа.

Из этого же DOCX экспортируйте файл return_policy.pdf с помощью команды Сохранить как или Экспортировать. Обычный экспорт из текстового редактора, как правило, сохраняет текстовый слой. Не используйте фотографию страницы: для неё понадобится OCR, которого в текущем экстракторе ещё нет.

Я также подготовил архив с готовыми файлами для проверки. Вы можете скачать его по этой ссылке и использовать документы, если не хотите создавать их самостоятельно.

Теперь в корне проекта создайте временный файл document_extraction_demo.py и добавьте в него код:

from dataclasses import replace
from pathlib import Path

from app.files.default_document_extractor import (
    build_default_document_extractor,
)
from app.files.document_extraction import (
    DocumentExtractionError,
    ExtractedDocument,
)
from app.files.temporary_storage import (
    TemporaryFileStorage,
    TemporaryFileStorageError,
)
from app.files.validation import (
    FileValidationError,
    FileValidationPolicy,
    FileValidator,
)

PROJECT_ROOT = Path(__file__).resolve().parent
SAMPLE_FILES_ROOT = PROJECT_ROOT / "sample_files"
SAMPLE_PATHS = (
    SAMPLE_FILES_ROOT / "valid.txt",
    SAMPLE_FILES_ROOT / "return_policy.md",
    SAMPLE_FILES_ROOT / "return_policy.pdf",
    SAMPLE_FILES_ROOT / "return_policy.docx",
)

def print_document(document: ExtractedDocument) -> None:
    metadata = document.metadata
    print(f"Формат: {metadata.kind}")
    print(f"Частей: {metadata.part_count}")
    print(f"Страниц: {metadata.page_count}")
    print(f"Символов: {metadata.character_count}")

    for warning in document.warnings:
        print(f"Предупреждение: {warning}")

    for part in document.parts:
        preview = part.text.replace("\n", " ")[:160]
        print(f"  {part.reference}: {preview}")

def main() -> None:
    policy = FileValidationPolicy(
        allowed_root=SAMPLE_FILES_ROOT,
        max_size_bytes=10 * 1_024 * 1_024,
    )
    validator = FileValidator(policy)
    storage = TemporaryFileStorage(
        max_size_bytes=policy.max_size_bytes
    )
    extractor = build_default_document_extractor()

    for source_path in SAMPLE_PATHS:
        print(f"\nДокумент: {source_path.name}")

        try:
            validated = validator.validate(source_path)

            with storage.stage(validated) as temporary_path:
                staged_policy = replace(
                    policy,
                    allowed_root=temporary_path.parent,
                )
                staged_validator = FileValidator(staged_policy)
                staged_file = staged_validator.validate(
                    temporary_path,
                    declared_media_type=validated.media_type,
                )
                staged_file = replace(
                    staged_file,
                    original_name=validated.original_name,
                )
                document = extractor.extract(staged_file)

            print_document(document)
        except (
            FileValidationError,
            TemporaryFileStorageError,
            DocumentExtractionError,
        ) as error:
            print(f"Обработка не выполнена: {error}")

if __name__ == "__main__":
    main()

В начале создаются уже знакомые валидатор и временное хранилище. Новый объект extractor собирает обработчики четырёх текстовых форматов. Затем один цикл выполняет одинаковую последовательность для каждого образца.

Константа SAMPLE_PATHS содержит только исходные пути. Мы не создаём ValidatedFile вручную, потому что этот объект должен появляться исключительно после успешной проверки. Каждый элемент кортежа проходит весь путь независимо от остальных.

Сначала проверяется исходный путь. Только допустимый ValidatedFile копируется во временный каталог. Временная копия повторно проверяется политикой с новым allowed_root, после чего передаётся единому экстрактору.

Блок try охватывает все ожидаемые этапы одного документа. Если проверка, копирование или извлечение завершается прикладной ошибкой, общий except печатает понятное сообщение, а цикл продолжает работу со следующим образцом. Ошибка одного файла не останавливает всю демонстрацию.

Повторный валидатор видит случайное внутреннее имя файла. Поэтому после проверки мы используем replace() и возвращаем в staged_file.original_name исходное пользовательское имя. Без этого метаданные содержали бы UUID временной копии, который бесполезен пользователю.

Само извлечение выполняется внутри with, пока временный файл существует. Объект ExtractedDocument содержит обычные строки и метаданные, поэтому после выхода из блока его можно безопасно печатать, хотя временная копия уже удалена.

Обратите внимание на точное место строки document = extractor.extract(staged_file). Если вынести её после блока with, временный каталог к этому моменту уже будет удалён и парсер не сможет открыть файл. Печать результата, наоборот, можно выполнять после with, потому что извлечение уже завершилось.

print_document() показывает общие метаданные, все предупреждения и короткое начало каждой части. Ограничение [:160] касается только вывода демонстрации: полный текст остаётся в document.text и part.text.

Сначала функция получает короткое имя metadata для вложенного объекта и выводит общие числа. Затем отдельный цикл печатает все предупреждения, а следующий — части документа. Перед сокращением символы перевода строки заменяются пробелами, поэтому одна часть занимает в консоли одну читаемую строку.

Запускаем полную цепочку

Запустим данный файл и давайте посмотрим на результат. Точные числа зависят от ваших документов, но общая форма результата будет такой:

/Users/permin0ff/PycharmProjects/AI_Course_Developer/ai-support-desk/.venv/bin/python /Users/permin0ff/PycharmProjects/AI_Course_Developer/ai-support-desk/document_extraction_demo.py

Документ: valid.txt
Формат: text
Частей: 1
Страниц: None
Символов: 27
  document: Короткое обращение клиента.

Документ: return_policy.md
Формат: markdown
Частей: 3
Страниц: None
Символов: 222
  section:1: Покупатель может обратиться в поддержку в течение 14 дней.
  section:2: Укажите номер заказа и приложите фотографию повреждения.
  section:3: Товар должен сохранить комплектность и внешний вид.

Документ: return_policy.pdf
Ignoring wrong pointing object 6 0 (offset 0)
Ignoring wrong pointing object 8 0 (offset 0)
Ignoring wrong pointing object 10 0 (offset 0)
Ignoring wrong pointing object 12 0 (offset 0)
Ignoring wrong pointing object 14 0 (offset 0)
Формат: pdf
Частей: 1
Страниц: 1
Символов: 1128
  page:1: Возврат товара Покупатель может вернуть товар, если он не подошёл, имеет недостатки или не соответствует описанию. Порядок и сроки возврата зависят от состояния

Документ: return_policy.docx
Формат: docx
Частей: 3
Страниц: None
Символов: 1149
Предупреждение: Для DOCX номера страниц не определяются: сохранены ссылки на разделы.
Предупреждение: Таблицы DOCX преобразованы в обычный текст.
  document: Возврат товара  Покупатель может вернуть товар, если он не подошёл, имеет недостатки или не соответствует описанию.  Порядок и сроки возврата зависят от состоян
  section:1: Если товар был повреждён во время доставки, покупателю необходимо обратиться в поддержку и указать номер заказа.  К обращению рекомендуется приложить фотографии
  section:2: Исправный товар можно вернуть в течение 14 дней после получения, если он сохранил товарный вид, комплектность и потребительские свойства.  Перед возвратом необх

Мы видим что у Markdown и DOCX ссылки указывают на логические разделы, а у PDF — на физические страницы. И это правильная логика, ведь DocumentPart позволяет каждому формату сохранить ту привязку, которой действительно можно доверять. Так же обратите внимание, что у на обрезается только текст, который показывается в консоли. Сам извлечённый текст в part.text сохраняется полностью. Число 1128 как раз показывает полное количество извлечённых символов.

А на сообщения Ignoring wrong pointing object... можно не обращать внимания. Это предупреждения о некорректных служебных ссылках внутри PDF. Библиотека pypdf пропускает их и продолжает обработку. Если текст успешно извлечён и программа завершилась без ошибки, документ обработан правильно.

После успешного запуска прежний file_validation_demo.py можно удалить. Ведь наш новый сценарий уже выполняет его полезную цепочку, а именно первую проверку, временное копирование и повторную проверку, а затем идёт дальше и извлекает содержимое. Но сам document_extraction_demo.py пока оставьте. Он пригодится для проверки новых форматов независимо от основной консоли. Когда в конце этого модуля появится полноценный Document AI Service и работа с документами будет подключена к приложению, этот временный файл тоже выполнил свою задачу и будет удалён.

Подведём итоги раздела. Мы отделили четыре разных действия:

Но извлечение ещё не является пониманием документа, ведь экстрактор не знает, разрешён ли возврат конкретному клиенту, не составляет резюме и не отвечает на вопросы. Он лишь готовит проверенный текст с происхождением фрагментов. Такая граница важна и детерминированную работу с файлами выполняет обычный код, а модель позже получает только уже подготовленное содержимое.

В следующем разделе займёмся CSV и Excel. И тут есть момент, ведь таблицы нельзя бездумно превращать в одну длинную строку, а нам нужно сохранить листы, колонки, типы значений и пропуски, выбрать полезный диапазон и выполнить простые вычисления до обращения к LLM.

Шаг 5

Что является основной обязанностью экстрактора документов?

Правильный ответ
  • ✅ Получить текст и сохранить ссылки на исходные страницы или разделы

Шаг 6

PDF визуально содержит слова, но extract_text() не нашёл текст ни на одной странице. Как поступить?

Правильный ответ
  • ✅ Сообщить, что текстовый слой не найден и документ, вероятно, требует OCR

Шаг 7

Почему для DOCX используются ссылки section:N, а не page:N?

Правильный ответ
  • ✅ Разбивка DOCX на страницы зависит от редактора, шрифтов и настроек отображения

Шаг 8

Какие свойства нужны единому DocumentExtractor?

Правильный ответ
  • ✅ Один зарегистрированный обработчик для каждого FileKind
  • ✅ Передача ValidatedFile выбранному обработчику без повторного определения типа
  • ✅ Ошибка при повторной регистрации одного формата
  • ✅ Специализированная ошибка для неподдерживаемого формата

Шаг 9

Экстрактор обходит страницы PDF так:

for page_number, page in enumerate(reader.pages, start=1):
    text = page.extract_text() or ""
    if not text.strip():
        continue
    parts.append({"text": text, "reference": f"page:{page_number}"})

На второй странице нет текста, а на третьей он найден. Какое значение reference получит часть с текстом третьей страницы?

Правильный ответ
  • ✅ page:3

Шаг 10

Приведение извлечённого текста к единому виду

Парсеры документов могут возвращать разные переводы строк, BOM, неразрывные пробелы и невидимые управляющие символы. Такой текст неудобно сравнивать, делить на части и передавать модели. Функция нормализации должна выполнять преобразования в определённом порядке и возвращать предсказуемую строку.

Задача

Реализуйте функцию normalize_extracted_text(value). Она должна очистить служебные символы, привести переводы строк к \n и убрать лишние пустые строки.

Требования

  • Не изменяйте имя функции. Она принимает один аргумент value.
  • Если value не является строкой, выбросьте TypeError.
  • Удалите из текста символ BOM \ufeff.
  • Сначала замените последовательности \r\n на \n, затем оставшиеся \r также замените на \n.
  • Замените неразрывный пробел \u00a0 на обычный пробел.
  • Удалите Unicode-символы, категория которых начинается с C, кроме перевода строки \n и табуляции \t. Для определения категории используйте unicodedata.category().
  • Разделите текст на строки и удалите пробельные символы справа у каждой строки.
  • В итоговом тексте допускается не более одной пустой строки подряд.
  • После объединения строк удалите внешние пробельные символы всего результата с помощью strip().
  • Верните нормализованную строку.

Пример результата

source = "\ufeff Заголовок\r\n\r\n\r\nТекст\u00a0 \x00"
normalized = normalize_extracted_text(source)
print(repr(normalized))

Результат:

'Заголовок\n\nТекст'

unicodedata уже импортирован. Работать с файлами в этой задаче не нужно.

Правильный ответ 
import unicodedata

def normalize_extracted_text(value: str) -> str:
    if not isinstance(value, str):
        raise TypeError

    value = value.replace("\ufeff", "")
    value = value.replace("\r\n", "\n")
    value = value.replace("\r", "\n")
    value = value.replace("\u00a0", " ")

    value = "".join(
        char
        for char in value
        if char in ("\n", "\t") or not unicodedata.category(char).startswith("C")
    )

    lines = [line.rstrip() for line in value.split("\n")]

    result = []
    previous_empty = False

    for line in lines:
        is_empty = not line

        if is_empty and previous_empty:
            continue

        result.append(line)
        previous_empty = is_empty

    return "\n".join(result).strip()

Шаг 11

Разбиение Markdown по заголовкам

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

Задача

Реализуйте функцию split_markdown_sections(text). Она должна пройти по строкам Markdown и вернуть кортеж словарей с частями в исходном порядке.

Требования

  • Не изменяйте имя функции. Она принимает один аргумент text.
  • Если text не является строкой, выбросьте TypeError.
  • Если text.strip() пуст, выбросьте ValueError.
  • Считайте заголовком только строку, которая начинается с последовательности от одного до шести символов #, затем содержит один пробел и непустое название.
  • Количество символов # является числовым уровнем заголовка.
  • Если до первого заголовка есть непустой текст, создайте первую часть с reference="document", title=None, level=None и очищенным текстом.
  • Каждый найденный заголовок начинает новую часть. Текстом части являются строки после заголовка и до следующего заголовка.
  • Для частей с заголовками формируйте ссылки "section:1", "section:2" и далее в порядке появления.
  • В каждой такой части сохраните ключи reference, title, level и text.
  • Значение title должно быть очищенным названием без символов #. Значение text также очистите по краям.
  • Раздел с заголовком допустим даже без текста; в этом случае сохраните text="".
  • Верните части как tuple, сохранив порядок документа.

Пример результата

parts = split_markdown_sections(
    "Введение\n# Установка\nШаги\n## Детали\nТекст"
)
for part in parts:
    print(part)

Результат:

{'reference': 'document', 'title': None, 'level': None, 'text': 'Введение'}
{'reference': 'section:1', 'title': 'Установка', 'level': 1, 'text': 'Шаги'}
{'reference': 'section:2', 'title': 'Детали', 'level': 2, 'text': 'Текст'}

Регулярное выражение для заголовка нужно написать самостоятельно. Дополнительные библиотеки не нужны.

Правильный ответ 
import re

def split_markdown_sections(text: str) -> tuple[dict, ...]:
    if not isinstance(text, str):
        raise TypeError

    if not text.strip():
        raise ValueError

    heading_pattern = re.compile(r"^(#{1,6}) (.+)$")

    parts = []
    current_title = None
    current_level = None
    current_lines = []
    section_number = 0

    def save_current():
        nonlocal section_number

        content = "\n".join(current_lines).strip()

        if current_title is None:
            if content:
                parts.append({
                    "reference": "document",
                    "title": None,
                    "level": None,
                    "text": content,
                })
        else:
            section_number += 1
            parts.append({
                "reference": f"section:{section_number}",
                "title": current_title,
                "level": current_level,
                "text": content,
            })

    for line in text.splitlines():
        match = heading_pattern.match(line)

        if match:
            save_current()

            current_title = match.group(2).strip()
            current_level = len(match.group(1))
            current_lines = []
        else:
            current_lines.append(line)

    save_current()

    return tuple(parts)

Шаг 12

Сбор текста со страниц PDF

PDF-парсер может вернуть строку для страницы с текстовым слоем и None для страницы без доступного текста. Непустые страницы нужно сохранить с их настоящими номерами, а пустые — перечислить в предупреждениях. Пропуск пустой страницы не должен менять номера следующих страниц.

Задача

Реализуйте функцию collect_pdf_text(page_texts). Она должна преобразовать результаты чтения страниц в кортеж частей и кортеж предупреждений.

Требования

  • Не изменяйте имя функции. Она принимает один аргумент page_texts.
  • page_texts должен быть списком.
  • Каждый элемент списка должен быть строкой или None. При другом типе контейнера или элемента выбросьте TypeError.
  • Нумеруйте страницы с единицы по их позиции в исходном списке.
  • Для строки удалите внешние пробелы. Если после очистки текст непустой, создайте словарь с ключами reference, page_number и text.
  • reference формируйте как "page:N", где N — настоящий номер страницы. В page_number сохраните то же число.
  • Если значение равно None или очищенная строка пуста, не создавайте часть для этой страницы.
  • Для каждой пустой страницы добавьте строку "Страница N: текстовый слой не найден." в список предупреждений.
  • Если после обхода не создано ни одной части, выбросьте ValueError.
  • Верните пару (tuple(parts), tuple(warnings)).

Пример результата

parts, warnings = collect_pdf_text([
    " Первая ",
    None,
    "  ",
    "Четвёртая",
])
print(parts)
print(warnings)

Результат:

({'reference': 'page:1', 'page_number': 1, 'text': 'Первая'}, {'reference': 'page:4', 'page_number': 4, 'text': 'Четвёртая'})
('Страница 2: текстовый слой не найден.', 'Страница 3: текстовый слой не найден.')

Настоящий PDF и библиотека pypdf не используются. Проверяющая система передаёт готовые результаты extract_text().

Правильный ответ 
def collect_pdf_text(page_texts) -> tuple[tuple[dict, ...], tuple[str, ...]]:
    if not isinstance(page_texts, list):
        raise TypeError

    parts = []
    warnings = []

    for page_number, value in enumerate(page_texts, start=1):
        if value is not None and not isinstance(value, str):
            raise TypeError

        if value is None:
            warnings.append(
                f"Страница {page_number}: текстовый слой не найден."
            )
            continue

        text = value.strip()

        if not text:
            warnings.append(
                f"Страница {page_number}: текстовый слой не найден."
            )
            continue

        parts.append({
            "reference": f"page:{page_number}",
            "page_number": page_number,
            "text": text,
        })

    if not parts:
        raise ValueError

    return tuple(parts), tuple(warnings)