From 0a318aa0061d5e24ee5e09915e17ce704b066d68 Mon Sep 17 00:00:00 2001 From: Repinoid Date: Sat, 31 Oct 2026 09:35:44 +0300 Subject: [PATCH] feat: add v3 extraction storage and structured parsing --- site/db/connection.py | 114 ++++++++++++++++++++++++++ site/db/dossiers.py | 113 +++++++++++++++++++++++++ site/db/exceptions.py | 71 ++++++++++++++++ site/db/extractions.py | 111 +++++++++++++++++++++++++ site/db/timeline.py | 101 +++++++++++++++++++++++ site/services/parse.py | 182 +++++++++++++++++++++++++++++++++++------ 6 files changed, 669 insertions(+), 23 deletions(-) create mode 100644 site/db/dossiers.py create mode 100644 site/db/exceptions.py create mode 100644 site/db/extractions.py create mode 100644 site/db/timeline.py diff --git a/site/db/connection.py b/site/db/connection.py index 0407fbd..ce54343 100644 --- a/site/db/connection.py +++ b/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); """) diff --git a/site/db/dossiers.py b/site/db/dossiers.py new file mode 100644 index 0000000..4d767c9 --- /dev/null +++ b/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,)) \ No newline at end of file diff --git a/site/db/exceptions.py b/site/db/exceptions.py new file mode 100644 index 0000000..9b08998 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/site/db/extractions.py b/site/db/extractions.py new file mode 100644 index 0000000..e819d30 --- /dev/null +++ b/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] \ No newline at end of file diff --git a/site/db/timeline.py b/site/db/timeline.py new file mode 100644 index 0000000..28c5e2d --- /dev/null +++ b/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,)) \ No newline at end of file diff --git a/site/services/parse.py b/site/services/parse.py index 20cc37e..e0c0e40 100644 --- a/site/services/parse.py +++ b/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)