diff --git a/parse_pol177.py b/parse_pol177.py index f499671..618e2be 100644 --- a/parse_pol177.py +++ b/parse_pol177.py @@ -1,13 +1,11 @@ #!/usr/bin/env python3 """ -Парсинг ПОЛ-177 через Unstructured API: -- Текст: Unstructured API (192.168.1.103:8005) -- Таблицы: Unstructured API (element_type=Table, strategy=hi_res) - -Сохраняет результат в /opt/okf-regulations/concepts/pol177_parsed.json +Парсинг регламента через Unstructured API. +Делает один проход (strategy=hi_res) и разделяет элементы на текст и таблицы. """ import json -import subprocess +import re +import requests from pathlib import Path # === НАСТРОЙКИ === @@ -15,106 +13,81 @@ UNSTRUCTURED_API = "http://192.168.1.103:8005/general/v0/general" PDF_PATH = "/opt/documents_xk/АОХКСибцем/07 Служба Вице-президента по экономике и финансам/ДИТ/02 Регламенты деятельности/08 Положение Управление ИТ-инцидентами от 20.02.2024 № ПОЛ-123-2.pdf" OUTPUT_PATH = "/opt/okf-regulations/concepts/pol123_parsed.json" -def fetch_unstructured(pdf_path: str, strategy: str = "hi_res", element_type: str = None) -> list[dict]: - """Получаем элементы из Unstructured API. - - Args: - pdf_path: путь к PDF - strategy: hi_res для таблиц, fast для текста - element_type: фильтр по типу (None = все, 'Table' = только таблицы) - """ - cmd = [ - "curl", "-X", "POST", UNSTRUCTURED_API, - "-F", f"files=@{pdf_path}", - "-F", f"strategy={strategy}", - "-F", "output_format=application/json", - "-F", "coordinates=true", - ] - if element_type: - cmd.extend(["-F", f"element_types=[{element_type}]"]) - - result = subprocess.run(cmd, capture_output=True, text=True) - if result.returncode != 0: - print(f" ❌ Unstructured API error: {result.stderr}") - return [] - data = json.loads(result.stdout) - if isinstance(data, list): - return data - return data.get("elements", []) -def extract_tables(pdf_path: str) -> list[dict]: - """Извлекаем таблицы через Unstructured API и фильтруем по типу + HTML.""" - print(" Запрос всех элементов через Unstructured API (strategy=hi_res)...") - elements = fetch_unstructured(pdf_path, strategy="hi_res", element_type=None) +def parse_document(pdf_path: str) -> dict: + """Отправляет PDF в API (hi_res) и возвращает рассортированные данные.""" + print(f" Отправка файла {Path(pdf_path).name} в Unstructured API (hi_res)...") - real_tables = [] - for el in elements: - # Фильтр 1: тип должен быть Table - if el.get("type") != "Table": - continue - - meta = el.get("metadata", {}) - text = el.get("text", "") - html = meta.get("text_as_html", "") - - # Фильтр 2: должен быть HTML-представление (настоящая таблица) - if not html or "]*>', html, re.IGNORECASE) - first_tr = re.findall(r']*>|]*>', html[:500], re.IGNORECASE) - if trs: - row_count = len(trs) - if first_tr: - col_count = len(first_tr) - - table_data = { - "page": meta.get("page_number"), - "text": text, - "html": html, - "row_count": row_count, - "col_count": col_count, + with open(pdf_path, "rb") as f: + files = {"files": (Path(pdf_path).name, f, "application/pdf")} + data = { + "strategy": "hi_res", + "coordinates": "true" } - real_tables.append(table_data) - - return real_tables + + try: + response = requests.post(UNSTRUCTURED_API, files=files, data=data, timeout=300) + response.raise_for_status() + except requests.exceptions.RequestException as e: + print(f" ❌ Ошибка Unstructured API: {e}") + return {"text": [], "tables": []} + + elements = response.json() + if isinstance(elements, dict): + elements = elements.get("elements", []) + + text_elements = [] + real_tables = [] + + print(" Сортировка и фильтрация элементов...") + for el in elements: + el_type = el.get("type") + + if el_type == "Table": + meta = el.get("metadata", {}) + html = meta.get("text_as_html", "") + + # Строгий фильтр на наличие HTML-таблицы + if html and "]*>', html, re.IGNORECASE)) if html else None + col_count = len(re.findall(r']*>|]*>', html[:500], re.IGNORECASE)) if html else None + + real_tables.append({ + "page": meta.get("page_number"), + "text": el.get("text", ""), + "html": html, + "row_count": row_count, + "col_count": col_count, + }) + else: + # Если таблица оказалась фейковой, отправляем её текст в общий котел + text_elements.append(el) + else: + # Всё остальное (Title, NarrativeText, ListItem и т.д.) + text_elements.append(el) + + return {"text": text_elements, "tables": real_tables} -def extract_text(pdf_path: str) -> list[dict]: - """Извлекаем текст через Unstructured API (strategy=fast).""" - print(" Запрос текста через Unstructured API (strategy=fast)...") - elements = fetch_unstructured(pdf_path, strategy="fast") - return elements def main(): print("=" * 60) - print("📄 Парсинг ПОЛ-177") + print(f"📄 Парсинг документа: {Path(PDF_PATH).name}") print("=" * 60) - # 1. Текст - print("\n[1/3] Извлечение текста (strategy=fast)...") - text_elements = extract_text(PDF_PATH) - print(f" ✅ Получено {len(text_elements)} элементов") + parsed_data = parse_document(PDF_PATH) + + print(f"\n[Итоги]") + print(f" ✅ Получено текстовых элементов: {len(parsed_data['text'])}") + print(f" ✅ Найдено валидных таблиц: {len(parsed_data['tables'])}") - # 2. Таблицы - print("\n[2/3] Извлечение таблиц (element_type=Table)...") - tables = extract_tables(PDF_PATH) - print(f" ✅ Найдено {len(tables)} таблиц") - - # 3. Сохраняем результат result = { "document": "ПОЛ-123", "title": "Положение Управление ИТ-инцидентами от 20.02.2024 № ПОЛ-123-2", - "text": text_elements, - "tables": tables, + "text": parsed_data["text"], + "tables": parsed_data["tables"], "metadata": { "unstructured_api": UNSTRUCTURED_API, - "strategy_text": "fast", - "strategy_tables": "hi_res", + "strategy": "hi_res", "timestamp": "2026-07-06", }, } @@ -124,4 +97,4 @@ def main(): print("=" * 60) if __name__ == "__main__": - main() + main() \ No newline at end of file