feat: add v3 extraction storage and structured parsing

site/db/connection.py

@@ -130,11 +130,125 @@ def _create_schema(conn):
             created_at TEXT DEFAULT (datetime('now'))
         );
 
+        -- Извлечённый паспорт хранится отдельно от исходного документа, чтобы
+        -- повторный запуск пайплайна не перезаписывал исходные байты и парсинг.
+        CREATE TABLE IF NOT EXISTS document_extractions (
+            id TEXT PRIMARY KEY,
+            document_id TEXT NOT NULL UNIQUE REFERENCES documents(id) ON DELETE CASCADE,
+            doc_class TEXT NOT NULL DEFAULT 'unknown',
+            intent_mode TEXT NOT NULL DEFAULT 'no_spec',
+            own_number TEXT,
+            own_date TEXT,
+            counterparty_name TEXT,
+            counterparty_inn TEXT,
+            referenced_numbers_json TEXT NOT NULL DEFAULT '[]',
+            sections_json TEXT NOT NULL DEFAULT '[]',
+            math_validation_json TEXT NOT NULL DEFAULT '{}',
+            status TEXT NOT NULL DEFAULT 'pending',
+            prompt_version TEXT NOT NULL DEFAULT '',
+            raw_llm_response TEXT NOT NULL DEFAULT '{}',
+            created_at TEXT DEFAULT (datetime('now')),
+            updated_at TEXT DEFAULT (datetime('now'))
+        );
+
+        -- Досье создаётся в рамках batch; отсутствие рамочного договора не
+        -- запрещает реконструкцию по спецификациям и дополнительным соглашениям.
+        CREATE TABLE IF NOT EXISTS contract_dossiers (
+            id TEXT PRIMARY KEY,
+            batch_id TEXT,
+            canonical_number TEXT NOT NULL,
+            alias_numbers_json TEXT NOT NULL DEFAULT '[]',
+            counterparty_name TEXT NOT NULL DEFAULT '',
+            counterparty_inn TEXT NOT NULL DEFAULT '',
+            has_frame_contract INTEGER NOT NULL DEFAULT 0,
+            status TEXT NOT NULL DEFAULT 'ACTIVE',
+            created_at TEXT DEFAULT (datetime('now'))
+        );
+
+        -- Связь задаёт принятую принадлежность документа и сохраняет её основание.
+        -- UNIQUE предотвращает дубли при повторной сборке групп одного batch.
+        CREATE TABLE IF NOT EXISTS dossier_documents (
+            id TEXT PRIMARY KEY,
+            dossier_id TEXT NOT NULL REFERENCES contract_dossiers(id) ON DELETE CASCADE,
+            document_id TEXT NOT NULL REFERENCES documents(id) ON DELETE CASCADE,
+            role TEXT NOT NULL DEFAULT 'supplement',
+            link_status TEXT NOT NULL DEFAULT 'CONFIRMED',
+            link_reason TEXT NOT NULL DEFAULT '',
+            seq_order INTEGER NOT NULL DEFAULT 0,
+            created_at TEXT DEFAULT (datetime('now')),
+            UNIQUE (dossier_id, document_id)
+        );
+
+        -- Идентичность услуги отделена от параметров и цены, которые меняются
+        -- по временным интервалам.
+        CREATE TABLE IF NOT EXISTS service_items (
+            id TEXT PRIMARY KEY,
+            dossier_id TEXT NOT NULL REFERENCES contract_dossiers(id) ON DELETE CASCADE,
+            service_family TEXT NOT NULL,
+            instance_anchor TEXT NOT NULL DEFAULT '',
+            canonical_name TEXT NOT NULL,
+            catalog_sku TEXT,
+            sku_status TEXT NOT NULL DEFAULT 'UNKNOWN',
+            created_at TEXT DEFAULT (datetime('now'))
+        );
+
+        -- Источник интервала каскадно удаляется вместе с документом. Документ,
+        -- закрывший интервал, очищается через SET NULL: сам интервал остаётся.
+        CREATE TABLE IF NOT EXISTS timeline_intervals (
+            id TEXT PRIMARY KEY,
+            dossier_id TEXT NOT NULL REFERENCES contract_dossiers(id) ON DELETE CASCADE,
+            service_item_id TEXT NOT NULL REFERENCES service_items(id) ON DELETE CASCADE,
+            charge_type TEXT NOT NULL DEFAULT 'recurring',
+            valid_from TEXT,
+            valid_to TEXT,
+            date_status TEXT NOT NULL DEFAULT 'explicit',
+            date_basis TEXT NOT NULL DEFAULT '',
+            raw_name TEXT NOT NULL,
+            config_params_json TEXT NOT NULL DEFAULT '{}',
+            price TEXT,
+            qty TEXT,
+            sum TEXT,
+            vat_mode TEXT NOT NULL DEFAULT 'with_vat',
+            source_document_id TEXT REFERENCES documents(id) ON DELETE CASCADE,
+            closed_by_document_id TEXT REFERENCES documents(id) ON DELETE SET NULL,
+            table_ref TEXT NOT NULL DEFAULT '',
+            quote TEXT NOT NULL DEFAULT '',
+            status TEXT NOT NULL DEFAULT 'VERIFIED',
+            created_at TEXT DEFAULT (datetime('now'))
+        );
+
+        -- Исключения могут быть привязаны к разным этапам и разрешаться отдельно.
+        CREATE TABLE IF NOT EXISTS exceptions_queue (
+            id TEXT PRIMARY KEY,
+            batch_id TEXT,
+            dossier_id TEXT REFERENCES contract_dossiers(id) ON DELETE CASCADE,
+            document_id TEXT REFERENCES documents(id) ON DELETE CASCADE,
+            service_item_id TEXT REFERENCES service_items(id) ON DELETE SET NULL,
+            severity TEXT NOT NULL DEFAULT 'CONFLICT',
+            stage TEXT NOT NULL,
+            title TEXT NOT NULL,
+            source_quote TEXT NOT NULL DEFAULT '',
+            options_json TEXT NOT NULL DEFAULT '[]',
+            resolution_json TEXT,
+            status TEXT NOT NULL DEFAULT 'OPEN',
+            created_at TEXT DEFAULT (datetime('now')),
+            resolved_at TEXT
+        );
+
         CREATE INDEX IF NOT EXISTS idx_documents_batch ON documents(batch_id);
         CREATE INDEX IF NOT EXISTS idx_documents_hash ON documents(batch_id, content_hash);
         CREATE INDEX IF NOT EXISTS idx_supplements_contract ON supplements(contract_id);
         CREATE INDEX IF NOT EXISTS idx_spec_current_contract ON spec_current(contract_id);
         CREATE INDEX IF NOT EXISTS idx_spec_events_supplement ON spec_events(supplement_id);
+        CREATE INDEX IF NOT EXISTS idx_extractions_document ON document_extractions(document_id);
+        CREATE INDEX IF NOT EXISTS idx_dossiers_batch ON contract_dossiers(batch_id);
+        CREATE INDEX IF NOT EXISTS idx_dossier_documents_dossier ON dossier_documents(dossier_id);
+        CREATE INDEX IF NOT EXISTS idx_dossier_documents_document ON dossier_documents(document_id);
+        CREATE INDEX IF NOT EXISTS idx_service_items_dossier ON service_items(dossier_id);
+        CREATE INDEX IF NOT EXISTS idx_timeline_dossier ON timeline_intervals(dossier_id);
+        CREATE INDEX IF NOT EXISTS idx_timeline_service ON timeline_intervals(service_item_id);
+        CREATE INDEX IF NOT EXISTS idx_exceptions_dossier ON exceptions_queue(dossier_id);
+        CREATE INDEX IF NOT EXISTS idx_exceptions_batch ON exceptions_queue(batch_id);
     """)
 
 

site/db/dossiers.py

@@ -0,0 +1,113 @@
+"""CRUD для досье договоров и доказанных связей с документами.
+
+Досье идентифицируется внутренним UUID, а номер договора и его алиасы
+сохраняются как изменяемые утверждения. Поэтому опечатки в документах
+не становятся ключом базы и могут быть пересмотрены без переноса данных.
+"""
+import json
+import uuid
+
+from db.connection import execute, query, transaction
+
+
+def _json_value(value, default):
+    """Кодировать поля-списки, принимая уже сериализованный JSON без повтора."""
+    if value is None:
+        value = default
+    return value if isinstance(value, str) else json.dumps(value, ensure_ascii=False)
+
+
+def _decode_dossier(row):
+    """Добавить к строке досье Python-представление списка алиасов."""
+    if not row:
+        return None
+    result = dict(row)
+    try:
+        result["alias_numbers"] = json.loads(result.get("alias_numbers_json") or "[]")
+    except (TypeError, json.JSONDecodeError):
+        result["alias_numbers"] = []
+    return result
+
+
+def create_dossier(batch_id, canonical_number, alias_numbers=None,
+                   counterparty_name="", counterparty_inn="",
+                   has_frame_contract=False, status="ACTIVE"):
+    """Создать досье и вернуть его запись со стабильным внутренним ID."""
+    dossier_id = str(uuid.uuid4())
+    execute(
+        """INSERT INTO contract_dossiers
+           (id, batch_id, canonical_number, alias_numbers_json,
+            counterparty_name, counterparty_inn, has_frame_contract, status)
+           VALUES (%s, %s, %s, %s, %s, %s, %s, %s)""",
+        (dossier_id, batch_id, canonical_number,
+         _json_value(alias_numbers, []), counterparty_name or "",
+         counterparty_inn or "", int(bool(has_frame_contract)), status),
+    )
+    return get_dossier(dossier_id)
+
+
+def link_document(dossier_id, document_id, role="supplement",
+                  link_status="CONFIRMED", link_reason="", seq_order=0):
+    """Создать или обновить связь dossier-document идемпотентно.
+
+    Повторная сборка групп не должна создавать дубликаты. При уже имеющейся
+    связи сохраняется её ID, но актуальные роль, статус, основание и порядок
+    заменяются значениями нового решения группировщика.
+    """
+    existing = query(
+        "SELECT id FROM dossier_documents WHERE dossier_id = %s AND document_id = %s",
+        (dossier_id, document_id),
+    )
+    if existing:
+        link_id = existing[0]["id"]
+        execute(
+            """UPDATE dossier_documents
+               SET role = %s, link_status = %s, link_reason = %s, seq_order = %s
+               WHERE id = %s""",
+            (role, link_status, link_reason or "", seq_order, link_id),
+        )
+    else:
+        link_id = str(uuid.uuid4())
+        execute(
+            """INSERT INTO dossier_documents
+               (id, dossier_id, document_id, role, link_status, link_reason, seq_order)
+               VALUES (%s, %s, %s, %s, %s, %s, %s)""",
+            (link_id, dossier_id, document_id, role, link_status,
+             link_reason or "", seq_order),
+        )
+    rows = query("SELECT * FROM dossier_documents WHERE id = %s", (link_id,))
+    return rows[0] if rows else None
+
+
+def list_dossiers_by_batch(batch_id):
+    """Получить досье batch вместе с упорядоченным списком документов."""
+    dossiers = query(
+        "SELECT * FROM contract_dossiers WHERE batch_id = %s ORDER BY created_at, id",
+        (batch_id,),
+    )
+    result = []
+    for raw_dossier in dossiers:
+        dossier = _decode_dossier(raw_dossier)
+        dossier["documents"] = query(
+            """SELECT dd.*, d.filename, d.doc_type, d.own_number, d.parent_number,
+                      d.doc_date, d.counterparty, d.zip_source
+               FROM dossier_documents dd
+               JOIN documents d ON d.id = dd.document_id
+               WHERE dd.dossier_id = %s
+               ORDER BY dd.seq_order, d.created_at, d.id""",
+            (dossier["id"],),
+        )
+        result.append(dossier)
+    return result
+
+
+def get_dossier(dossier_id):
+    """Получить досье по внутреннему ID, не загружая список документов."""
+    rows = query("SELECT * FROM contract_dossiers WHERE id = %s", (dossier_id,))
+    return _decode_dossier(rows[0]) if rows else None
+
+
+def clear_batch_dossiers(batch_id):
+    """Удалить все досье batch; таблицы с ON DELETE CASCADE очистятся вместе с ними."""
+    with transaction():
+        return execute("DELETE FROM contract_dossiers WHERE batch_id = %s", (batch_id,))

site/db/exceptions.py

@@ -0,0 +1,71 @@
+"""CRUD для очереди исключений и сохранения решений оператора."""
+import json
+import uuid
+
+from db.connection import execute, query
+
+
+def add_exception(batch_id, dossier_id, document_id, service_item_id,
+                  severity, stage, title, source_quote="", options=None):
+    """Добавить конкретную нерешённую задачу и вернуть её запись."""
+    exception_id = str(uuid.uuid4())
+    options_json = json.dumps(options or [], ensure_ascii=False, default=str)
+    execute(
+        """INSERT INTO exceptions_queue
+           (id, batch_id, dossier_id, document_id, service_item_id, severity,
+            stage, title, source_quote, options_json)
+           VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)""",
+        (exception_id, batch_id, dossier_id, document_id, service_item_id,
+         severity, stage, title, source_quote or "", options_json),
+    )
+    return get_exception(exception_id)
+
+
+def get_exception(exception_id):
+    """Получить исключение по ID и декодировать варианты решения."""
+    rows = query("SELECT * FROM exceptions_queue WHERE id = %s", (exception_id,))
+    if not rows:
+        return None
+    result = dict(rows[0])
+    try:
+        result["options"] = json.loads(result.get("options_json") or "[]")
+    except (TypeError, json.JSONDecodeError):
+        result["options"] = []
+    try:
+        result["resolution"] = json.loads(result.get("resolution_json") or "null")
+    except (TypeError, json.JSONDecodeError):
+        result["resolution"] = None
+    return result
+
+
+def list_exceptions(batch_id=None, dossier_id=None, status=None):
+    """Список исключений с необязательными фильтрами batch/dossier/status."""
+    clauses = []
+    params = []
+    for field, value in (("batch_id", batch_id), ("dossier_id", dossier_id),
+                         ("status", status)):
+        if value is not None:
+            clauses.append(f"{field} = %s")
+            params.append(value)
+    where = f" WHERE {' AND '.join(clauses)}" if clauses else ""
+    rows = query(
+        f"SELECT id FROM exceptions_queue{where} ORDER BY created_at, id",
+        params,
+    )
+    return [get_exception(row["id"]) for row in rows]
+
+
+def resolve_exception(exception_id, resolution):
+    """Сохранить решение без удаления исходной причины и цитаты.
+
+    Пересчёт зависимых данных выполняет сервисный слой; CRUD отвечает только
+    за аудируемую фиксацию решения человека.
+    """
+    resolution_json = json.dumps(resolution, ensure_ascii=False, default=str)
+    changed = execute(
+        """UPDATE exceptions_queue
+           SET resolution_json = %s, status = 'RESOLVED', resolved_at = datetime('now')
+           WHERE id = %s AND status = 'OPEN'""",
+        (resolution_json, exception_id),
+    )
+    return get_exception(exception_id) if changed else None

site/db/extractions.py

@@ -0,0 +1,111 @@
+"""CRUD для результатов stateless-извлечения отдельных документов.
+
+Исходный документ и результат LLM разделены намеренно: повторный запуск
+извлечения может обновить паспорт, не затрагивая байты файла и его структуру.
+Составные поля сохраняются как JSON-текст, поскольку SQLite не требует
+отдельного JSON-типа и существующий слой БД уже использует такой подход.
+"""
+import json
+import uuid
+
+from db.connection import execute, query
+
+
+_JSON_FIELDS = (
+    "referenced_numbers_json",
+    "sections_json",
+    "math_validation_json",
+)
+_FIELDS = (
+    "doc_class",
+    "intent_mode",
+    "own_number",
+    "own_date",
+    "counterparty_name",
+    "counterparty_inn",
+    *_JSON_FIELDS,
+    "status",
+    "prompt_version",
+    "raw_llm_response",
+)
+
+
+def _encode_json(value):
+    """Сериализовать значения в JSON, сохраняя Decimal как точную строку."""
+    return json.dumps(value, ensure_ascii=False, default=str)
+
+
+def _decode_row(row):
+    """Вернуть составные поля паспорта как Python-объекты для сервисов."""
+    if not row:
+        return None
+    result = dict(row)
+    for field in _JSON_FIELDS:
+        raw_value = result.get(field)
+        try:
+            result[field.removesuffix("_json")] = json.loads(raw_value or "null")
+        except (TypeError, json.JSONDecodeError):
+            result[field.removesuffix("_json")] = None
+    return result
+
+
+def upsert_extraction(document_id, data):
+    """Создать или заменить паспорт документа, не меняя его стабильный ID.
+
+    Входные JSON-поля принимают как Python-списки/словари, так и уже
+    сериализованные строки. Поля, которых нет в `data`, получают безопасные
+    значения по умолчанию при вставке и остаются без изменений при обновлении.
+    """
+    existing = query(
+        "SELECT id FROM document_extractions WHERE document_id = %s",
+        (document_id,),
+    )
+    values = {}
+    for field in _FIELDS:
+        if field not in data:
+            continue
+        value = data[field]
+        if field in _JSON_FIELDS and not isinstance(value, str):
+            value = _encode_json(value)
+        values[field] = value
+
+    if existing:
+        if values:
+            assignments = ", ".join(f"{field} = %s" for field in values)
+            params = [*values.values(), document_id]
+            execute(
+                f"UPDATE document_extractions SET {assignments}, "
+                "updated_at = datetime('now') WHERE document_id = %s",
+                params,
+            )
+        return get_by_document(document_id)
+
+    extraction_id = str(uuid.uuid4())
+    columns = ["id", "document_id", *values.keys()]
+    placeholders = ", ".join(["%s"] * len(columns))
+    execute(
+        f"INSERT INTO document_extractions ({', '.join(columns)}) "
+        f"VALUES ({placeholders})",
+        [extraction_id, document_id, *values.values()],
+    )
+    return get_by_document(document_id)
+
+
+def get_by_document(document_id):
+    """Получить паспорт документа или `None`, если он ещё не извлечён."""
+    rows = query(
+        "SELECT * FROM document_extractions WHERE document_id = %s",
+        (document_id,),
+    )
+    return _decode_row(rows[0]) if rows else None
+
+
+def list_by_batch(batch_id):
+    """Получить паспорта всех документов batch в порядке загрузки."""
+    rows = query(
+        """SELECT e.* FROM document_extractions e
+           JOIN documents d ON d.id = e.document_id
+           WHERE d.batch_id = %s ORDER BY d.created_at, d.id""",
+        (batch_id,),
+    )
+    return [_decode_row(row) for row in rows]

site/db/timeline.py

@@ -0,0 +1,101 @@
+"""CRUD для услуг и временных интервалов обязательств.
+
+Денежные поля принимаются как десятичные строки. SQLite здесь отвечает за
+хранение и фильтрацию дат, а арифметика остаётся в сервисном слое на Decimal.
+"""
+import uuid
+
+from db.connection import execute, query, transaction
+
+
+def insert_service_item(dossier_id, service_family, canonical_name,
+                        instance_anchor="", catalog_sku=None,
+                        sku_status="UNKNOWN"):
+    """Создать устойчивую идентичность услуги в пределах досье."""
+    service_id = str(uuid.uuid4())
+    execute(
+        """INSERT INTO service_items
+           (id, dossier_id, service_family, instance_anchor, canonical_name,
+            catalog_sku, sku_status)
+           VALUES (%s, %s, %s, %s, %s, %s, %s)""",
+        (service_id, dossier_id, service_family, instance_anchor or "",
+         canonical_name, catalog_sku, sku_status),
+    )
+    return get_service_item(service_id)
+
+
+def get_service_item(service_item_id):
+    """Получить услугу по ID либо вернуть `None`."""
+    rows = query("SELECT * FROM service_items WHERE id = %s", (service_item_id,))
+    return rows[0] if rows else None
+
+
+def insert_interval(data):
+    """Создать интервал, сохраняя точные цены/количества строками.
+
+    `data` должен содержать dossier_id, service_item_id и raw_name; остальные
+    поля берутся из значений по умолчанию схемы либо передаются явно.
+    """
+    interval_id = str(uuid.uuid4())
+    fields = (
+        "dossier_id", "service_item_id", "charge_type", "valid_from", "valid_to",
+        "date_status", "date_basis", "raw_name", "config_params_json", "price",
+        "qty", "sum", "vat_mode", "source_document_id", "closed_by_document_id",
+        "table_ref", "quote", "status",
+    )
+    values = [data[field] for field in fields if field in data]
+    selected_fields = [field for field in fields if field in data]
+    columns = ["id", *selected_fields]
+    placeholders = ", ".join(["%s"] * len(columns))
+    execute(
+        f"INSERT INTO timeline_intervals ({', '.join(columns)}) VALUES ({placeholders})",
+        [interval_id, *values],
+    )
+    rows = query("SELECT * FROM timeline_intervals WHERE id = %s", (interval_id,))
+    return rows[0] if rows else None
+
+
+def close_interval(interval_id, valid_to, closed_by_document_id=None):
+    """Закрыть интервал полуоткрытой границей и указать закрывший документ."""
+    return execute(
+        """UPDATE timeline_intervals
+           SET valid_to = %s, closed_by_document_id = %s
+           WHERE id = %s""",
+        (valid_to, closed_by_document_id, interval_id),
+    )
+
+
+def list_timeline(dossier_id, as_of_date=None):
+    """Получить историю либо активный на дату срез по досье.
+
+    Для среза используется единая полуоткрытая семантика: начало включено,
+    окончание исключено. NULL-дата начала не считается подтверждённо активной.
+    """
+    if as_of_date:
+        return query(
+            """SELECT * FROM timeline_intervals
+               WHERE dossier_id = %s AND valid_from <= %s
+                 AND (valid_to IS NULL OR valid_to > %s)
+               ORDER BY valid_from, service_item_id, id""",
+            (dossier_id, as_of_date, as_of_date),
+        )
+    return query(
+        """SELECT * FROM timeline_intervals WHERE dossier_id = %s
+           ORDER BY valid_from, service_item_id, id""",
+        (dossier_id,),
+    )
+
+
+def reset_dossier_timeline(dossier_id):
+    """Сбросить вычисленный таймлайн и открытые исключения сопоставления.
+
+    Решённые исключения сохраняются как аудит пользовательского решения;
+    нерешённые lineage-исключения пересоздаются вместе с новым построением.
+    """
+    with transaction():
+        execute(
+            """DELETE FROM exceptions_queue
+               WHERE dossier_id = %s AND stage = 'lineage_matching' AND status = 'OPEN'""",
+            (dossier_id,),
+        )
+        return execute("DELETE FROM service_items WHERE dossier_id = %s", (dossier_id,))

site/services/parse.py

@@ -1,4 +1,9 @@
-"""Parse service — PDF (pdfplumber), DOCX (python-docx), plain text fallback."""
+"""Структурный парсер PDF/DOCX с порядком чтения и контекстом таблиц.
+
+Формат elements_json остаётся совместимым с текущей загрузкой: верхнеуровневый
+status по-прежнему только `parsed` или `error`. Новые метаданные добавляются
+в элементы, чтобы выбор файлов и отображение результата загрузки не менялись.
+"""
 import io
 
 
@@ -19,37 +24,126 @@ def parse_file(filename, data):
 
 
 def _parse_pdf(data):
-    """Parse PDF with pdfplumber — preserves table structure (unlike PyPDF2)."""
+    """Извлечь PDF в порядке страницы, привязав предшествующий текст к таблицам.
+
+    Координаты pdfplumber позволяют отделить заголовки таблиц от текста в её
+    ячейках и расположить абзацы/таблицы в порядке чтения. Для сканов без
+    текстового слоя возвращается parsed с отдельным флагом, чтобы существующий
+    upload-контракт не трактовал нечитаемый файл как ошибку передачи.
+    """
     import pdfplumber
     elements = []
+    table_index = 0
     with pdfplumber.open(io.BytesIO(data)) as pdf:
         for page in pdf.pages:
-            # Tables first — preserves column structure critical for specs
-            tables = page.extract_tables()
-            for table in tables:
-                if table:
-                    rows = []
-                    for row in table:
-                        if row:
-                            cells = [str(cell or "").strip() for cell in row]
-                            if any(cells):
-                                rows.append(cells)
-                    if rows:
-                        elements.append({"type": "table", "rows": rows})
-            # Remaining text as paragraphs
-            text = page.extract_text()
-            if text:
-                for line in text.split("\n"):
-                    line = line.strip()
-                    if line:
-                        elements.append({"type": "paragraph", "text": line, "style": ""})
-    return {"status": "parsed", "element_count": len(elements), "elements": elements}
+            table_events = []
+            table_regions = []
+            for table in page.find_tables():
+                rows = _clean_table_rows(table.extract())
+                if not rows:
+                    continue
+                table_index += 1
+                x0, top, x1, bottom = table.bbox
+                table_regions.append((x0, top, x1, bottom))
+                table_events.append((float(top), {
+                    "type": "table",
+                    "table_index": table_index,
+                    "rows": rows,
+                }))
+
+            word_events = _pdf_paragraph_events(page, table_regions)
+            events = [(top, 0, element) for top, element in word_events]
+            events.extend((top, 1, element) for top, element in table_events)
+            events.sort(key=lambda event: (event[0], event[1]))
+
+            preceding_paragraphs = []
+            for _top, _kind, element in events:
+                if element["type"] == "paragraph":
+                    elements.append(element)
+                    preceding_paragraphs.append(element["text"])
+                    preceding_paragraphs = preceding_paragraphs[-4:]
+                else:
+                    element["preceding_paragraphs"] = list(preceding_paragraphs)
+                    elements.append(element)
+                    preceding_paragraphs = []
+
+    result = {"status": "parsed", "element_count": len(elements), "elements": elements}
+    if not elements:
+        result["is_unreadable_scan"] = True
+    return result
+
+
+def _clean_table_rows(table):
+    """Нормализовать ячейки таблицы, удаляя только полностью пустые строки."""
+    rows = []
+    for row in table or []:
+        if not row:
+            continue
+        cells = [str(cell or "").strip() for cell in row]
+        if any(cells):
+            rows.append(cells)
+    return rows
+
+
+def _pdf_paragraph_events(page, table_regions):
+    """Сгруппировать слова PDF в строки вне bbox таблиц.
+
+    Каждая строка получает верхнюю координату для общей сортировки с таблицами.
+    Если PDF-библиотека смогла получить текст страницы, но не вернула слова,
+    весь текст сохраняется одним абзацем вместо ошибочной пометки «скан».
+    """
+    words = page.extract_words() or []
+    visible_words = []
+    for word in words:
+        x0 = float(word.get("x0", 0))
+        x1 = float(word.get("x1", x0))
+        top = float(word.get("top", 0))
+        bottom = float(word.get("bottom", top))
+        center_x = (x0 + x1) / 2
+        center_y = (top + bottom) / 2
+        inside_table = any(
+            left <= center_x <= right and upper <= center_y <= lower
+            for left, upper, right, lower in table_regions
+        )
+        text = str(word.get("text", "")).strip()
+        if text and not inside_table:
+            visible_words.append((top, x0, text))
+
+    visible_words.sort(key=lambda item: (item[0], item[1]))
+    lines = []
+    for top, x0, text in visible_words:
+        if not lines or abs(top - lines[-1]["top"]) > 3:
+            lines.append({"top": top, "words": []})
+        lines[-1]["words"].append((x0, text))
+
+    events = []
+    for line in lines:
+        line_text = " ".join(text for _x0, text in sorted(line["words"]))
+        if line_text:
+            events.append((line["top"], {
+                "type": "paragraph", "text": line_text, "style": "",
+            }))
+    if events:
+        return events
+    if words and table_regions:
+        return []
+
+    text = page.extract_text()
+    if not text:
+        return []
+    return [
+        (0.0, {"type": "paragraph", "text": line.strip(), "style": ""})
+        for line in text.split("\n") if line.strip()
+    ]
 
 
 def _parse_docx(data):
+    """Извлечь DOCX, сохранив порядок блоков и абзацы перед каждой таблицей."""
     import docx
     doc = docx.Document(io.BytesIO(data))
     elements = []
+    preceding_paragraphs = []
+    table_index = 0
 
     for block in doc.element.body:
         tag = block.tag.split("}")[-1] if "}" in block.tag else block.tag
@@ -57,6 +151,8 @@ def _parse_docx(data):
             text = _extract_paragraph_text(block)
             if text:
                 elements.append({"type": "paragraph", "text": text, "style": ""})
+                preceding_paragraphs.append(text)
+                preceding_paragraphs = preceding_paragraphs[-4:]
         elif tag == "tbl":
             rows = []
             for tr in block.findall(".//{http://schemas.openxmlformats.org/wordprocessingml/2006/main}tr"):
@@ -67,7 +163,14 @@ def _parse_docx(data):
                 if cells:
                     rows.append(cells)
             if rows:
-                elements.append({"type": "table", "rows": rows})
+                table_index += 1
+                elements.append({
+                    "type": "table",
+                    "table_index": table_index,
+                    "preceding_paragraphs": list(preceding_paragraphs),
+                    "rows": rows,
+                })
+                preceding_paragraphs = []
 
     return {"status": "parsed", "element_count": len(elements), "elements": elements}
 
@@ -85,3 +188,36 @@ def _parse_text(data):
     lines = [l.strip() for l in text.split("\n") if l.strip()]
     return {"status": "parsed", "element_count": len(lines),
             "elements": [{"type": "paragraph", "text": l, "style": ""} for l in lines]}
+
+
+def format_elements_with_structure(elements):
+    """Отформатировать элементы документа для stateless-промпта.
+
+    Табличный контекст продублирован в заголовке таблицы намеренно: extractor
+    получает локальную связь периода/итога с таблицей, даже если его окно
+    внимания не удерживает длинный юридический текст перед ней.
+    """
+    lines = []
+    paragraph_index = 0
+    table_index = 0
+    for element in elements or []:
+        if not isinstance(element, dict):
+            continue
+        if element.get("type") == "paragraph":
+            text = str(element.get("text") or "").strip()
+            if text:
+                paragraph_index += 1
+                lines.append(f"[ПАРАГРАФ p{paragraph_index}] {text}")
+        elif element.get("type") == "table":
+            table_index += 1
+            table_ref = element.get("table_index") or table_index
+            context = " | ".join(
+                str(value).strip()
+                for value in element.get("preceding_paragraphs", [])
+                if str(value).strip()
+            )
+            lines.append(f"[ТАБЛИЦА #{table_ref} | Контекст над таблицей: {context!r}]")
+            for row_index, row in enumerate(element.get("rows", []), start=1):
+                cells = [str(cell or "").strip() for cell in (row or [])]
+                lines.append(f"  [Строка {table_ref}.{row_index}] " + " | ".join(cells))
+    return "\n".join(lines)