feat: complete academic symbol lexicon heal, 2-column Cooper extraction, and updated assets
This commit is contained in:
Executable
+278
@@ -0,0 +1,278 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user