#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ ingest_newsources_corpus.py =========================== Парсер библиотеки newsources: 1. М. Баттистини — «Символы и аллегории» (2008) + извлечение сотен цветных иллюстраций 2. Ч. Уильямс — «Китайская культура. Мифы, герои, символы» (2011) + гравюры 3. Роберт Бир — «Тибетские буддийские символы» (2016) + тушевые диаграммы 4. П.Д. Успенский — «Символы Таро» (2023) 5. Г. Циммер — «Мифы и символы в индийской культуре» (2020) 6. М. Элиаде — «Образы и символы» (2000) 7. В.Г. Рохмистров — «Книга алхимии» (2008) 8. Н.Н. Велецкая — «Символы славянского язычества» (2009) ПРАВИЛА: - Приоритет 1: знаки 000–999 Оракула -> пишутся ТОЛЬКО в staging (oracle_passports_staging.json) - Все неучтенное -> пишется в базу внешних символов (oracle_external_symbols.json) - Картинки обязательны -> сохраняются в assets/illustrations/ и привязываются к объектам! """ import os import sys import json import re from pathlib import Path from typing import Dict, Any, List, Optional, Tuple from datetime import datetime PROJECT_ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(PROJECT_ROOT)) import fitz from oracle_enricher.enricher.passport_schema import count_filled_fields NEWSOURCES_DIR = Path("/home/alexives/Desktop/newsources") ILLUSTRATIONS_DIR = PROJECT_ROOT / "assets" / "illustrations" CATALOG_PATH = PROJECT_ROOT / "assets" / "data" / "oracle_illustrations_catalog.json" STAGING_DB_PATH = PROJECT_ROOT / "assets" / "data" / "oracle_passports_staging.json" EXTERNAL_DB_PATH = PROJECT_ROOT / "assets" / "data" / "oracle_external_symbols.json" ENRICHER_EXTERNAL_PATH = PROJECT_ROOT / "oracle_enricher" / "data" / "oracle_external_symbols.json" def clean_spaces(text: str) -> str: return re.sub(r'\s+', ' ', text).strip() def transliterate_to_slug(text: str) -> str: ru_to_en = { 'а': 'a', 'б': 'b', 'в': 'v', 'г': 'g', 'д': 'd', 'е': 'e', 'ё': 'yo', 'ж': 'zh', 'з': 'z', 'и': 'i', 'й': 'y', 'к': 'k', 'л': 'l', 'м': 'm', 'н': 'n', 'о': 'o', 'п': 'p', 'р': 'r', 'с': 's', 'т': 't', 'у': 'u', 'ф': 'f', 'х': 'kh', 'ц': 'ts', 'ч': 'ch', 'ш': 'sh', 'щ': 'shch', 'ъ': '', 'ы': 'y', 'ь': '', 'э': 'e', 'ю': 'yu', 'я': 'ya' } slug = [] for c in text.lower(): if c in ru_to_en: slug.append(ru_to_en[c]) elif c.isalnum(): slug.append(c) elif c in [' ', '-', '_']: slug.append('_') res = re.sub(r'_+', '_', ''.join(slug)).strip('_') return res[:30] if res else 'symbol' def extract_images_from_page(doc: fitz.Document, p_num: int, prefix: str, symbol_name: str, caption_hint: str) -> List[Dict[str, Any]]: """Извлекает крупные качественные изображения со страницы и сохраняет их в assets/illustrations/.""" results = [] imgs = doc[p_num].get_images() if not imgs: return results slug = transliterate_to_slug(symbol_name) for idx, img_info in enumerate(imgs): try: xref = img_info[0] base_img = doc.extract_image(xref) w, h = base_img["width"], base_img["height"] ext = base_img["ext"] # Фильтр мелких иконок/полос/декора if w < 160 or h < 160: continue fname = f"{prefix}_{p_num:03d}_{slug}_{idx}.{ext}" file_path = ILLUSTRATIONS_DIR / fname rel_asset_path = f"assets/illustrations/{fname}" with open(file_path, "wb") as f_out: f_out.write(base_img["image"]) caption = caption_hint if caption_hint else f"Иллюстрация: {symbol_name.title()}" results.append({ "id": f"{prefix}_{p_num}_{idx}", "file": rel_asset_path, "symbol": symbol_name, "caption": caption, "page": p_num + 1, "width": w, "height": h, }) except Exception: continue return results def main(): print("=" * 80) print("🚀 ГЛУБОКИЙ ПАРСИНГ НОВОЙ БИБЛИОТЕКИ NEWSOURCES") print("=" * 80) ILLUSTRATIONS_DIR.mkdir(parents=True, exist_ok=True) # 1. Загрузка баз данных with open(STAGING_DB_PATH, "r", encoding="utf-8") as f: staging_db = json.load(f) with open(EXTERNAL_DB_PATH, "r", encoding="utf-8") as f: external_db = json.load(f) catalog = [] if CATALOG_PATH.exists(): with open(CATALOG_PATH, "r", encoding="utf-8") as f: catalog = json.load(f) # Индекс канонических слов 1000 знаков core_index = {} for idx_str, entry in staging_db.items(): w = entry.get("word", "").lower().strip() if w: core_index[w] = idx_str print(f"Загружен канон 1000 знаков. Внешних символов в базе: {len(external_db)}.") timestamp_str = datetime.now().isoformat() def route_symbol(term: str, author: str, work: str, text: str, default_target_field: str, illustrations: List[Dict[str, Any]] = None): """Маршрутизирует символ: если в 1000 -> staging, иначе -> external_db.""" norm = term.strip().lower() if len(norm) < 2 or not text or len(text) < 25: return full_text = clean_spaces(text) excerpt = (full_text[:380] + "...") if len(full_text) > 380 else full_text citation = f"[{author} · {work}]" quoted = f"{citation}: «{excerpt}»" matched_idx = None if norm in core_index: matched_idx = core_index[norm] elif norm.endswith("ы") and norm[:-1] in core_index: matched_idx = core_index[norm[:-1]] elif norm.endswith("и") and norm[:-1] in core_index: matched_idx = core_index[norm[:-1]] elif norm.endswith("а") and norm[:-1] in core_index: matched_idx = core_index[norm[:-1]] # Привязка иллюстраций if illustrations: for ill in illustrations: ill["source"] = f"{author} · {work}" ill["normalized"] = norm ill["is_core_1000"] = (matched_idx is not None) ill["core_id"] = matched_idx if not any(x.get("file") == ill["file"] for x in catalog): catalog.append(ill) if matched_idx: # ОБОГАЩАЕМ ТЕСТОВУЮ БАЗУ (STAGING) entry = staging_db[matched_idx] fields = entry.setdefault("fields", {}) # Выбираем целевую ячейку, если default занята field_to_fill = default_target_field if fields.get(field_to_fill): alternatives = ["meta_synthesis", "semantic_core", "culture_antiquity_light", "culture_china_light", "culture_india_light", "culture_slavic_light", "culture_hermetic_light"] for alt in alternatives: if not fields.get(alt): field_to_fill = alt break if not fields.get(field_to_fill): fields[field_to_fill] = quoted filled, _ = count_filled_fields(entry) entry["filled_count"] = filled entry["is_complete"] = (filled >= 78) # Привязываем картинки к карточке if illustrations: target_ills = entry.setdefault("illustrations", []) for ill in illustrations: if not any(x.get("file") == ill["file"] for x in target_ills): target_ills.append(ill) else: # ПИШЕМ В БАЗУ ВНЕШНИХ СИМВОЛОВ if norm not in external_db: external_db[norm] = { "term": term.strip(), "normalized": norm, "category_guess": "Внешний академический символ", "sources": [], "illustrations": [], "created_at": timestamp_str, "updated_at": timestamp_str, } # Добавляем источник sources_list = external_db[norm]["sources"] if not any(s.get("author") == author for s in sources_list): sources_list.append({ "author": author, "work": work, "excerpt": excerpt, "full_text": full_text, }) # Добавляем картинки if illustrations: ext_ills = external_db[norm].setdefault("illustrations", []) for ill in illustrations: if not any(x.get("file") == ill["file"] for x in ext_ills): ext_ills.append(ill) external_db[norm]["updated_at"] = timestamp_str # ========================================================================= # ========================================================================= # 1. М. БАТТИСТИНИ — «Символы и аллегории» (2008) # ========================================================================= # Интеграция через канонический парсер sanitize_battistini_corpus.py print(" 📖 Обработка: Матильда Баттистини («Символы и аллегории») через канон...") from sanitize_battistini_corpus import CANONICAL_SYMBOLS, extract_clean_battistini_symbol bat_pdf = NEWSOURCES_DIR / "Battistini - SIMVOLY I ALLEGORII (Enziklopedija iskusstva) - 2008, pdf.pdf" if bat_pdf.exists(): doc_bat = fitz.open(str(bat_pdf)) for name, sp, ep in CANONICAL_SYMBOLS: item = extract_clean_battistini_symbol(doc_bat, name, sp, ep) route_symbol(name, "М. Баттистини", "«Символы и аллегории» (2008)", item["excerpt"], "culture_antiquity_light", []) print(" ✓ Баттистини (канон 71 символ) успешно обработана!") # ========================================================================= # 2. Ч. УИЛЬЯМС — «Китайская культура. Мифы, герои, символы» (2011) # ========================================================================= wil_pdf = NEWSOURCES_DIR / "Уильямс Ч. - Китайская культура. Мифы, герои, символы - 2011.pdf" if wil_pdf.exists(): print("\n📖 Обработка: Ч. Уильямс («Китайская культура: символы и мифы»)...") doc_wil = fitz.open(str(wil_pdf)) cur_title = None cur_lines = [] cur_imgs = [] for p in range(12, len(doc_wil) - 10): txt = doc_wil[p].get_text() lines = [l.strip() for l in txt.splitlines() if l.strip()] for l in lines: if l.isupper() and 2 <= len(l) <= 35 and re.match(r'^[А-ЯЁ\s\-]+$', l): if not any(skip in l for skip in ['КИТАЙСКАЯ КУЛЬТУРА', 'СОДЕРЖАНИЕ', 'ВВЕДЕНИЕ', 'УКАЗАТЕЛЬ', 'ЧАСТЬ']): if cur_title and cur_lines: route_symbol(cur_title, "Ч. Уильямс", "«Китайская культура: символы и мифы»", ' '.join(cur_lines), "culture_china_light", cur_imgs) cur_title = l cur_lines = [] cur_imgs = extract_images_from_page(doc_wil, p, "wil", cur_title, f"Китайский символ: {cur_title.title()}") else: if cur_title and not l.isdigit(): cur_lines.append(l) if cur_title and cur_lines: route_symbol(cur_title, "Ч. Уильямс", "«Китайская культура: символы и мифы»", ' '.join(cur_lines), "culture_china_light", cur_imgs) print(" ✓ Уильямс успешно обработан!") # ========================================================================= # 3. РОБЕРТ БИР — «Тибетские буддийские символы. Справочник» (2016) # ========================================================================= beer_pdf = NEWSOURCES_DIR / "Роберт Бир - Тибетские буддийские символы. Справочник (Самадхи) - 2016.pdf" if beer_pdf.exists(): print("\n📖 Обработка: Роберт Бир («Тибетские буддийские символы»)...") doc_beer = fitz.open(str(beer_pdf)) # Извлекаем оглавление toc_entries = [] cur_title_parts = [] for p in range(4, 9): lines = doc_beer[p].get_text().splitlines() for l in lines: s = l.strip() if not s: continue if s.isdigit(): page_num = int(s) if cur_title_parts: full_title = ' '.join(cur_title_parts).strip().rstrip('\t') if len(full_title) >= 3: toc_entries.append((full_title, page_num)) cur_title_parts = [] else: cur_title_parts.append(s) # Парсим текст между статьями for i in range(len(toc_entries)): sym_title, start_p = toc_entries[i] end_p = toc_entries[i + 1][1] if i + 1 < len(toc_entries) else start_p + 4 end_p = min(end_p, len(doc_beer)) entry_lines = [] entry_imgs = [] for p_idx in range(start_p - 1, end_p - 1): if 0 <= p_idx < len(doc_beer): entry_lines.append(doc_beer[p_idx].get_text()) imgs = extract_images_from_page(doc_beer, p_idx, "beer", sym_title, f"Тибетский сакральный знак: {sym_title}") entry_imgs.extend(imgs) entry_text = ' '.join(entry_lines) route_symbol(sym_title, "Роберт Бир", "«Тибетские буддийские символы»", entry_text, "culture_india_light", entry_imgs) print(" ✓ Роберт Бир успешно обработан!") # ========================================================================= # 4. П.Д. УСПЕНСКИЙ — «Символы Таро» (2023) # ========================================================================= tarot_pdf = NEWSOURCES_DIR / "Успенский П. - Символы Таро. Разговоры с дьяволом (Библиотека мировой литературы) - 2023.pdf" if tarot_pdf.exists(): print("\n📖 Обработка: П.Д. Успенский («Символы Таро»)...") doc_tarot = fitz.open(str(tarot_pdf)) cur_card = None cur_lines = [] for p in range(25, 115): txt = doc_tarot[p].get_text() lines = [l.strip() for l in txt.splitlines() if l.strip()] for idx_l, l in enumerate(lines): if l.startswith("Карта ") and idx_l + 1 < len(lines): next_l = lines[idx_l + 1] if next_l.isupper() and 2 <= len(next_l) <= 30: if cur_card and cur_lines: route_symbol(cur_card, "П.Д. Успенский", "«Символы Таро»", ' '.join(cur_lines), "mantic_tarot") cur_card = next_l.strip() cur_lines = [] else: if cur_card and not l.isdigit() and not l.startswith("П. УСПЕНСКИЙ"): cur_lines.append(l) if cur_card and cur_lines: route_symbol(cur_card, "П.Д. Успенский", "«Символы Таро»", ' '.join(cur_lines), "mantic_tarot") print(" ✓ Успенский (Таро) успешно обработан!") # ========================================================================= # 5. Г. ЦИММЕР — «Мифы и символы в индийской культуре» (2020) # ========================================================================= zim_pdf = NEWSOURCES_DIR / "Циммер Генрих-Мифы и символы в индийской культуре-(Философские технологии.Религиоведение)-2020.pdf" if zim_pdf.exists(): print("\n📖 Обработка: Генрих Циммер («Мифы и символы в индийской культуре»)...") doc_zim = fitz.open(str(zim_pdf)) cur_title = None cur_lines = [] for p in range(15, len(doc_zim) - 5): txt = doc_zim[p].get_text() lines = [l.strip() for l in txt.splitlines() if l.strip()] for l in lines: if l.isupper() and 3 <= len(l) <= 35 and re.match(r'^[А-ЯЁ\s\-]+$', l): if not any(skip in l for skip in ['ГЕНРИХ ЦИММЕР', 'МИФЫ И СИМВОЛЫ', 'СОДЕРЖАНИЕ', 'ЧАСТЬ', 'ГЛАВА']): if cur_title and cur_lines: route_symbol(cur_title, "Генрих Циммер", "«Мифы и символы в индийской культуре»", ' '.join(cur_lines), "culture_india_light") cur_title = l cur_lines = [] else: if cur_title and not l.isdigit(): cur_lines.append(l) if cur_title and cur_lines: route_symbol(cur_title, "Генрих Циммер", "«Мифы и символы в индийской культуре»", ' '.join(cur_lines), "culture_india_light") print(" ✓ Циммер успешно обработан!") # ========================================================================= # 6. В.Г. РОХМИСТРОВ — «Книга алхимии» (2008) # ========================================================================= rokh_pdf = NEWSOURCES_DIR / "Рохмистров В.Г. Книга алхимии (Александрийская библиотека) - 2008" / "Рохмистров В.Г. Книга алхимии (Александрийская библиотека) - 2008.pdf" if rokh_pdf.exists(): print("\n📖 Обработка: В.Г. Рохмистров («Книга алхимии»)...") doc_rokh = fitz.open(str(rokh_pdf)) # Алхимические главы alch_terms = { "алхимия": (1, 15), "изумрудная скрижаль": (8, 14), "герметизм": (8, 13), "философский камень": (6, 12), "ртуть": (6, 11), "сера": (10, 12), "золото": (6, 9), "пятый элемент": (11, 13) } for term, (p1, p2) in alch_terms.items(): lines = [] for p in range(p1, min(p2 + 1, len(doc_rokh))): lines.append(doc_rokh[p].get_text()) route_symbol(term, "В.Г. Рохмистров", "«Книга алхимии»", ' '.join(lines), "mantic_alchemy") print(" ✓ Рохмистров (Алхимия) успешно обработан!") # ========================================================================= # 7. МИРЧА ЭЛИАДЕ — «Образы и символы» (2000) # ========================================================================= eliade_pdf = NEWSOURCES_DIR / "Элиаде М. Избранные сочинения, 2000" / "Элиаде М. - Избранные сочинения. Миф о вечном возвращении. Образы и символы. Священное и мирское. - 2000.pdf" if eliade_pdf.exists(): print("\n📖 Обработка: Мирча Элиаде («Образы и символы»)...") doc_eli = fitz.open(str(eliade_pdf)) eliade_themes = { "центр": (25, 45), "узел": (80, 110), "раковина": (120, 145), "жемчуг": (130, 145), "вода": (146, 165), "древо": (166, 190) } for term, (p1, p2) in eliade_themes.items(): lines = [] for p in range(p1, min(p2 + 1, len(doc_eli))): lines.append(doc_eli[p].get_text()) route_symbol(term, "Мирча Элиаде", "«Образы и символы»", ' '.join(lines), "meta_synthesis") print(" ✓ Элиаде успешно обработан!") # ========================================================================= # 8. Н.Н. ВЕЛЕЦКАЯ — «Символы славянского язычества» (2009) # ========================================================================= vel_pdf = NEWSOURCES_DIR / "Велецкая Н.Н. - Символы славянского язычества (ТЗР) - 2009.pdf" if vel_pdf.exists(): print("\n📖 Обработка: Н.Н. Велецкая («Символы славянского язычества»)...") doc_vel = fitz.open(str(vel_pdf)) slavic_themes = { "ладья": (30, 60), "огонь": (65, 95), "земля": (100, 130), "колесо": (140, 165), "предки": (170, 200) } for term, (p1, p2) in slavic_themes.items(): lines = [] for p in range(p1, min(p2 + 1, len(doc_vel))): lines.append(doc_vel[p].get_text()) route_symbol(term, "Н.Н. Велецкая", "«Символы славянского язычества»", ' '.join(lines), "culture_slavic_light") print(" ✓ Велецкая (Славянское язычество) успешно обработана!") # ========================================================================= # 9. СОХРАНЕНИЕ ВСЕХ БАЗ И КАТАЛОГА ИЛЛЮСТРАЦИЙ # ========================================================================= print(f"\n💾 Сохранение тестовой базы: {STAGING_DB_PATH}...") with open(STAGING_DB_PATH, "w", encoding="utf-8") as f: json.dump(staging_db, f, ensure_ascii=False, indent=2) print(f"💾 Сохранение базы внешних символов: {EXTERNAL_DB_PATH}...") with open(EXTERNAL_DB_PATH, "w", encoding="utf-8") as f: json.dump(external_db, f, ensure_ascii=False, indent=2) with open(ENRICHER_EXTERNAL_PATH, "w", encoding="utf-8") as f: json.dump(external_db, f, ensure_ascii=False, indent=2) print(f"💾 Сохранение реестра иллюстраций: {CATALOG_PATH}...") with open(CATALOG_PATH, "w", encoding="utf-8") as f: json.dump(catalog, f, ensure_ascii=False, indent=2) # Итоговые метрики total_fields_staging = sum(e.get("filled_count", 0) for e in staging_db.values()) total_illustrations = len(list(ILLUSTRATIONS_DIR.iterdir())) print("\n" + "=" * 80) print("📊 ФИНАЛЬНЫЙ ОТЧЕТ ИНГЕСТИИ NEWSOURCES:") print(f"• Всего внешних символов в реестре: {len(external_db):,} понятий") print(f"• Общая заполненность тестовой базы (staging): {total_fields_staging:,} / 78,000 ячеек") print(f"• Всего файлов иллюстраций в assets/illustrations/: {total_illustrations} шт.") print(f"• Записей в каталоге иллюстраций: {len(catalog)} шт.") print(f"• База внешних символов: {EXTERNAL_DB_PATH.name} ({EXTERNAL_DB_PATH.stat().st_size / (1024*1024):.2f} MB)") print("=" * 80) if __name__ == "__main__": main()