Обновить parse_pol177.py
This commit is contained in:
+62
-89
@@ -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}]"])
|
||||
def parse_document(pdf_path: str) -> dict:
|
||||
"""Отправляет PDF в API (hi_res) и возвращает рассортированные данные."""
|
||||
print(f" Отправка файла {Path(pdf_path).name} в Unstructured API (hi_res)...")
|
||||
|
||||
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)
|
||||
|
||||
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 "<table" not in html.lower():
|
||||
continue
|
||||
|
||||
# Извлекаем row/col из HTML
|
||||
row_count = None
|
||||
col_count = None
|
||||
if html:
|
||||
import re
|
||||
trs = re.findall(r'<tr[^>]*>', html, re.IGNORECASE)
|
||||
first_tr = re.findall(r'<td[^>]*>|<th[^>]*>', 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 "<table" in html.lower():
|
||||
row_count = len(re.findall(r'<tr[^>]*>', html, re.IGNORECASE)) if html else None
|
||||
col_count = len(re.findall(r'<td[^>]*>|<th[^>]*>', 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)
|
||||
|
||||
# 2. Таблицы
|
||||
print("\n[2/3] Извлечение таблиц (element_type=Table)...")
|
||||
tables = extract_tables(PDF_PATH)
|
||||
print(f" ✅ Найдено {len(tables)} таблиц")
|
||||
print(f"\n[Итоги]")
|
||||
print(f" ✅ Получено текстовых элементов: {len(parsed_data['text'])}")
|
||||
print(f" ✅ Найдено валидных таблиц: {len(parsed_data['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",
|
||||
},
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user