279 lines
10 KiB
Python
Executable File
279 lines
10 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
parse_local_dictionaries.py
|
||
===========================
|
||
Парсер локальных словарей из единой папки dictionaries/.
|
||
Любой файл (PDF, TXT, FB2, EPUB, HTML, JSON), положенный в dictionaries/,
|
||
автоматически сканируется, разбирается по словарным статьям и наполняет
|
||
ТЕСТОВУЮ базу (assets/data/oracle_passports_staging.json).
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import json
|
||
import re
|
||
import time
|
||
from pathlib import Path
|
||
from typing import Dict, Any, List, Optional, Tuple
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
||
sys.path.insert(0, str(PROJECT_ROOT))
|
||
|
||
from oracle_enricher.enricher.passport_schema import (
|
||
PROMPTS_78,
|
||
count_filled_fields,
|
||
render_passport_markdown,
|
||
)
|
||
|
||
DICT_DIR = PROJECT_ROOT / "dictionaries"
|
||
STAGING_DB_PATH = PROJECT_ROOT / "assets" / "data" / "oracle_passports_staging.json"
|
||
|
||
try:
|
||
import fitz # PyMuPDF
|
||
HAS_FITZ = True
|
||
except ImportError:
|
||
HAS_FITZ = False
|
||
|
||
try:
|
||
from bs4 import BeautifulSoup
|
||
HAS_BS4 = True
|
||
except ImportError:
|
||
HAS_BS4 = False
|
||
|
||
|
||
def scan_dictionary_files() -> List[Path]:
|
||
"""Сканирует единую папку dictionaries/ на наличие файлов книг."""
|
||
if not DICT_DIR.exists():
|
||
DICT_DIR.mkdir(parents=True, exist_ok=True)
|
||
return []
|
||
|
||
valid_exts = {".pdf", ".txt", ".fb2", ".epub", ".html", ".htm", ".json", ".xml", ".doc", ".docx"}
|
||
files = [
|
||
p for p in DICT_DIR.iterdir()
|
||
if p.is_file() and p.suffix.lower() in valid_exts and not p.name.startswith(".")
|
||
]
|
||
return sorted(files, key=lambda x: x.name)
|
||
|
||
|
||
def detect_book_meta(path: Path) -> Dict[str, str]:
|
||
"""Определяет автора и название книги по имени файла."""
|
||
name = path.name.lower()
|
||
|
||
if any(k in name for k in ["керлот", "cirlot"]):
|
||
return {"author": "Х.Э. Керлот", "title": "«Словарь символов»", "tag": "cirlot"}
|
||
elif any(k in name for k in ["бидерман", "biedermann"]):
|
||
return {"author": "Г. Бидерманн", "title": "«Энциклопедия символов»", "tag": "biedermann"}
|
||
elif any(k in name for k in ["шевалье", "гербран", "chevalier"]):
|
||
return {"author": "Ж. Шевалье, А. Гербрант", "title": "«Словарь символов»", "tag": "chevalier"}
|
||
elif any(k in name for k in ["славянск", "толст", "толстого"]):
|
||
return {"author": "ИСл РАН (ред. Н.И. Толстой)", "title": "«Славянские древности»", "tag": "tolstoy"}
|
||
elif any(k in name for k in ["фасмер", "vasmer"]):
|
||
return {"author": "М. Фасмер", "title": "«Этимологический словарь»", "tag": "fasmer"}
|
||
else:
|
||
clean_title = path.stem.replace("_", " ").title()
|
||
return {"author": clean_title, "title": f"«{clean_title}»", "tag": "generic"}
|
||
|
||
|
||
def extract_text_from_file(path: Path) -> str:
|
||
"""Извлекает текст из PDF, TXT, FB2, HTML."""
|
||
ext = path.suffix.lower()
|
||
|
||
if ext == ".txt":
|
||
for enc in ["utf-8", "windows-1251", "cp1251", "latin-1"]:
|
||
try:
|
||
with open(path, "r", encoding=enc) as f:
|
||
return f.read()
|
||
except UnicodeDecodeError:
|
||
continue
|
||
return ""
|
||
|
||
elif ext == ".pdf":
|
||
if not HAS_FITZ:
|
||
print(f" [WARN: нужен PyMuPDF для чтения {path.name}]", end="")
|
||
return ""
|
||
doc = fitz.open(str(path))
|
||
pages = []
|
||
for p in doc:
|
||
pages.append(p.get_text())
|
||
return "\n".join(pages)
|
||
|
||
elif ext in [".fb2", ".html", ".htm", ".xml"]:
|
||
for enc in ["utf-8", "windows-1251"]:
|
||
try:
|
||
with open(path, "r", encoding=enc) as f:
|
||
raw = f.read()
|
||
if HAS_BS4:
|
||
return BeautifulSoup(raw, "html.parser").get_text(separator="\n")
|
||
else:
|
||
return re.sub(r"<[^>]+>", " ", raw)
|
||
except UnicodeDecodeError:
|
||
continue
|
||
|
||
elif ext == ".json":
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
if isinstance(data, dict):
|
||
# Если словарь уже разобран на статьи
|
||
return json.dumps(data, ensure_ascii=False)
|
||
except Exception:
|
||
pass
|
||
|
||
return ""
|
||
|
||
|
||
def split_into_articles(text: str) -> Dict[str, str]:
|
||
"""Находит словарные статьи по алфавитным заголовкам."""
|
||
articles = {}
|
||
lines = text.splitlines()
|
||
|
||
current_head = ""
|
||
current_lines = []
|
||
|
||
# Заголовок словарной статьи: ОГОНЬ, ВОДА, BREAD, ВОЛК, ЗЕРКАЛО...
|
||
head_pattern = re.compile(r"^([А-ЯЁA-Z\s\-]{3,35})(?:\s*[\.,–—:]|\s*$)")
|
||
|
||
for line in lines:
|
||
s = line.strip()
|
||
if not s:
|
||
continue
|
||
|
||
m = head_pattern.match(s)
|
||
if m and len(s) <= 40 and not s.startswith("ГЛАВА") and not s.startswith("ЧАСТЬ"):
|
||
head_word = m.group(1).strip().lower()
|
||
if len(head_word) >= 3:
|
||
if current_head and current_lines:
|
||
body = "\n".join(current_lines).strip()
|
||
if len(body) > 30:
|
||
articles[current_head] = body
|
||
|
||
current_head = head_word
|
||
current_lines = [s]
|
||
continue
|
||
|
||
if current_head:
|
||
current_lines.append(s)
|
||
|
||
if current_head and current_lines:
|
||
articles[current_head] = "\n".join(current_lines).strip()
|
||
|
||
return articles
|
||
|
||
|
||
def main():
|
||
import argparse
|
||
parser = argparse.ArgumentParser(description="Парсер единой папки dictionaries/ для ТЕСТОВОЙ базы")
|
||
parser.add_argument("--scan", action="store_true", help="Показать найденные книги в папке dictionaries/")
|
||
args = parser.parse_args()
|
||
|
||
print("=" * 70)
|
||
print("📚 ПАРСЕР ЕДИНОЙ ПАПКИ СЛОВАРЕЙ «ОРАКУЛА 999»")
|
||
print(f"📁 Папка словарей: {DICT_DIR}")
|
||
print(f"🎯 Тестовая база: {STAGING_DB_PATH.name}")
|
||
print("=" * 70)
|
||
|
||
files = scan_dictionary_files()
|
||
|
||
if not files:
|
||
print(f"\n📂 В папке dictionaries/ пока нет файлов.")
|
||
print(" Просто положите туда любые книги (PDF, TXT, FB2, EPUB) и запустите скрипт:")
|
||
print(" python3 scripts/parse_local_dictionaries.py\n")
|
||
return
|
||
|
||
print(f"\nНайдено словарей: {len(files)}")
|
||
for f in files:
|
||
meta = detect_book_meta(f)
|
||
size_mb = f.stat().st_size / (1024 * 1024)
|
||
print(f" 📖 {f.name:<45} [{meta['author']} · {meta['title']}] ({size_mb:.1f} MB)")
|
||
|
||
if args.scan:
|
||
return
|
||
|
||
if not STAGING_DB_PATH.exists():
|
||
print(f"[ERROR] Тестовая база не найдена: {STAGING_DB_PATH}")
|
||
return
|
||
|
||
with open(STAGING_DB_PATH, "r", encoding="utf-8") as f:
|
||
staging_db = json.load(f)
|
||
|
||
total_articles = 0
|
||
total_injected = 0
|
||
t0 = time.time()
|
||
|
||
for fpath in files:
|
||
meta = detect_book_meta(fpath)
|
||
tag = meta["tag"]
|
||
citation = f"[{meta['author']} · {meta['title']}]"
|
||
|
||
print(f"\nЧтение: {fpath.name}...")
|
||
text = extract_text_from_file(fpath)
|
||
print(f" Извлечено символов: {len(text):,}")
|
||
|
||
if not text:
|
||
continue
|
||
|
||
# Если это уже готовый json {word: text}
|
||
if fpath.suffix.lower() == ".json":
|
||
try:
|
||
articles = json.loads(text)
|
||
except Exception:
|
||
articles = split_into_articles(text)
|
||
else:
|
||
articles = split_into_articles(text)
|
||
|
||
print(f" Распознано словарных статей: {len(articles):,}")
|
||
total_articles += len(articles)
|
||
|
||
# Сопоставляем с 1000 знаками
|
||
injected_this_book = 0
|
||
for entry in staging_db.values():
|
||
word = entry.get("word", "").lower().strip()
|
||
fields = entry.setdefault("fields", {})
|
||
|
||
if word in articles:
|
||
raw_text = articles[word].replace("\n", " ").strip()
|
||
excerpt = (raw_text[:350] + "...") if len(raw_text) > 350 else raw_text
|
||
quoted = f"{citation}: «{excerpt}»"
|
||
|
||
if tag == "cirlot" and not fields.get("meta_synthesis"):
|
||
fields["meta_synthesis"] = quoted
|
||
injected_this_book += 1
|
||
elif tag == "biedermann":
|
||
if not fields.get("culture_antiquity_light"):
|
||
fields["culture_antiquity_light"] = quoted
|
||
injected_this_book += 1
|
||
elif tag == "tolstoy" and not fields.get("culture_slavic_light"):
|
||
fields["culture_slavic_light"] = quoted
|
||
injected_this_book += 1
|
||
elif tag == "chevalier" and not fields.get("semantic_core"):
|
||
fields["semantic_core"] = quoted
|
||
injected_this_book += 1
|
||
elif tag == "generic":
|
||
if not fields.get("semantic_core"):
|
||
fields["semantic_core"] = quoted
|
||
injected_this_book += 1
|
||
|
||
filled, _ = count_filled_fields(entry)
|
||
entry["filled_count"] = filled
|
||
entry["is_complete"] = (filled >= 78)
|
||
|
||
print(f" ✓ Привязано статей к знакам: +{injected_this_book}")
|
||
total_injected += injected_this_book
|
||
|
||
# Сохраняем тестовую базу
|
||
with open(STAGING_DB_PATH, "w", encoding="utf-8") as f:
|
||
json.dump(staging_db, f, ensure_ascii=False, indent=2)
|
||
|
||
dur = time.time() - t0
|
||
print("\n" + "=" * 70)
|
||
print(f"🎉 ПАРСИНГ ЗАВЕРШЕН ЗА {dur:.1f}с!")
|
||
print(f"📚 Всего словарных статей в книгах: {total_articles:,}")
|
||
print(f"📈 Наполнено ячеек в ТЕСТОВОЙ базе: +{total_injected}")
|
||
print(f"💾 База сохранена: {STAGING_DB_PATH.name}")
|
||
print("=" * 70)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|