Files
999/scripts/ingest_newsources_corpus.py

470 lines
25 KiB
Python
Executable File
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/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()