#!/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()