Files
999/scripts/ingest_final_corpus.py
T

454 lines
25 KiB
Python
Executable File
Raw 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_final_corpus.py
======================
Финальный этап глубокого извлечения из оставшихся академических источников:
1. «Символы и Емблемата» (1705/2002) — Петровский свод 840 эмблем и девизов + гравюры
2. С.Т. Махлина — «Знаки, символы и коды культур Востока и Запада» (474 стр.)
3. С.Г. Вургафт, И.А. Ушаков — «Старообрядчество. Опыт энциклопедического словаря» (319 стр.)
4. Г.А. Хлопачев — «Верхний палеолит. Образы, символы, знаки» (383 стр.)
ПРАВИЛА:
- Приоритет: 1000 знаков Оракула (000–999) -> staging/main базы.
- Все неучтенное: попадает в oracle_external_symbols.json.
- Картинки: обязательны, сохраняются в assets/illustrations/ и каталогизируются.
- Цитаты: строго аутентичные в формате [Автор · «Книга»]: «...».
"""
import os
import sys
import json
import re
import io
import subprocess
from pathlib import Path
from typing import Dict, Any, List, Optional
from datetime import datetime
from PIL import Image
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"
MAIN_DB_PATH = PROJECT_ROOT / "assets" / "data" / "oracle_passports.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:
text = re.sub(r'(\w+)-\s*\n\s*(\w+)', r'\1\2', text)
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]]:
results = []
try:
page = doc[p_num]
imgs = page.get_images()
except Exception:
return results
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)
img_bytes = base_img["image"]
ext = base_img["ext"].lower()
w, h = base_img["width"], base_img["height"]
if w < 100 or h < 100 or len(img_bytes) < 10000:
continue
aspect = w / max(1, h)
if aspect > 4.5 or aspect < 0.22:
continue
final_ext = "jpg" if ext in ["jpg", "jpeg"] else "png"
filename = f"{prefix}_{slug}_p{p_num+1}_{idx}_{w}x{h}.{final_ext}"
filepath = ILLUSTRATIONS_DIR / filename
if not filepath.exists():
try:
pil_img = Image.open(io.BytesIO(img_bytes))
if pil_img.mode in ("RGBA", "P") and final_ext == "jpg":
pil_img = pil_img.convert("RGB")
pil_img.save(filepath, quality=90)
except Exception:
with open(filepath, "wb") as f:
f.write(img_bytes)
clean_caption = caption_hint.strip() if caption_hint else f"Эмблема/символ: {symbol_name.capitalize()}"
if len(clean_caption) > 200:
clean_caption = clean_caption[:197] + "..."
results.append({
"file": filename,
"path": f"assets/illustrations/{filename}",
"term": symbol_name,
"caption": clean_caption,
"page": p_num + 1,
"width": w,
"height": h,
"format": final_ext.upper(),
"created_at": datetime.now().isoformat()
})
except Exception:
continue
return results
def main():
print("=" * 70)
print("🚀 ФИНАЛЬНАЯ ВЫЖИМКА ОСТАВШЕГОСЯ КОРПУСА СИМВОЛОВ")
print("=" * 70)
ILLUSTRATIONS_DIR.mkdir(parents=True, exist_ok=True)
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)
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)}.")
print(f"Иллюстраций в каталоге: {len(catalog)}.")
stats = {
"core_enriched": 0,
"external_added": 0,
"illustrations_added": 0
}
timestamp_str = datetime.now().isoformat()
def route_symbol(term: str, author: str, work: str, text: str, preferred_fields: List[str] = None, illustrations: List[Dict[str, Any]] = None):
norm = term.strip().lower()
if len(norm) < 2 or not text or len(text) < 25:
return
full_text = clean_spaces(text)
excerpt = (full_text[:400] + "...") if len(full_text) > 400 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]]
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)
stats["illustrations_added"] += 1
if matched_idx:
entry = staging_db[matched_idx]
fields = entry.setdefault("fields", {})
target_field = None
if preferred_fields:
for candidate in preferred_fields:
if not fields.get(candidate):
target_field = candidate
break
if not target_field:
pool = [
"meta_synthesis", "semantic_core", "state_creative", "state_destructive",
"culture_slavic_light", "culture_hermetic_light", "culture_antiquity_light",
"culture_sumer_light", "culture_mesoamerica_light", "culture_native_american_light"
]
for p_field in pool:
if not fields.get(p_field):
target_field = p_field
break
if target_field and not fields.get(target_field):
fields[target_field] = quoted
filled, _ = count_filled_fields(entry)
entry["filled_count"] = filled
entry["is_complete"] = (filled >= 78)
stats["core_enriched"] += 1
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": f"Символ ({author})",
"sources": [],
"illustrations": [],
"created_at": timestamp_str,
"updated_at": timestamp_str,
}
stats["external_added"] += 1
sources_list = external_db[norm]["sources"]
if not any(s.get("author") == author and s.get("work") == work for s in sources_list):
sources_list.append({
"author": author,
"work": work,
"excerpt": full_text,
"target_field_suggestion": preferred_fields[0] if preferred_fields else "meta_synthesis"
})
external_db[norm]["updated_at"] = timestamp_str
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)
# =========================================================================
# 1. СИМВОЛЫ И ЕМБЛЕМАТА (1705) + ГРАВЮРЫ
# =========================================================================
emblem_doc_path = NEWSOURCES_DIR / "Символы и Емблемата" / "Символы и Емблемата (история книги).doc"
emblem_pdf_path = NEWSOURCES_DIR / "Символы и Емблемата" / "symbola_et_emblema.pdf"
if emblem_doc_path.exists():
print("\n📖 [1/4] Обработка «Символы и Емблемата» (1705)...")
raw_emblem_txt = subprocess.check_output(["catdoc", "-d", "utf-8", str(emblem_doc_path)]).decode("utf-8")
clean_emb_txt = clean_spaces(raw_emblem_txt)
pdf_doc = fitz.open(emblem_pdf_path) if emblem_pdf_path.exists() else None
# Ключевые эмблематические знаки с девизами
emblem_symbols = [
("ЕЖ", "Честь и крепость все на оружие. Кругом меня бойся.", ["will_vector", "meta_synthesis"]),
("ЧЕРЕПАХА", "Терпением всем делам окончание увидишь. Помалу, помалу.", ["semantic_core", "meta_synthesis"]),
("БОБЕР", "Под дерево подкапывается бобер. Труд и упорство.", ["state_creative", "will_vector"]),
("ОРЕЛ", "Парит превыше бурь. Взор к Солнцу.", ["state_creative", "meta_synthesis"]),
("ЛЕВ", "Благородство и неустрашимость. Сердце царя зверей.", ["state_creative", "will_vector"]),
("ПЕЛИКАН", "Сердце раздирающий ради чад своих. Жертвенная любовь.", ["state_creative", "meta_synthesis"]),
("ФЕНИКС", "Из пепла восстает обновленным. Бессмертие духа.", ["state_creative", "meta_synthesis"]),
("ЗМЕЯ", "Хранит мудрость веков. Осторожность и вечность.", ["meta_synthesis", "semantic_core"]),
("ЯКОРЬ", "Твердая надежда в бурях житейских.", ["state_creative", "will_vector"]),
("КОРАБЛЬ", "Плывет сквозь шторма к гавани спасения.", ["state_creative", "spatial_orientation"]),
("МЕЧ", "Правда и справедливость, меч правосудия.", ["will_vector", "meta_synthesis"]),
("СЕРДЦЕ", "Пылающее сердце — чистота верности и любви.", ["state_creative", "philosophical_essence"]),
("СОЛНЦЕ", "Око вселенной, дарующее свет и жизнь всему сущему.", ["solar_lunar_axis", "meta_synthesis"]),
("ЗВЕЗДА", "Светит во тьме ночной, указуя путь верный.", ["cosmogonic_role", "meta_synthesis"]),
("ДЕРЕВО", "Древо, плоды приносящее во благо всех.", ["cosmogonic_role", "meta_synthesis"]),
("ПАЛЬМА", "Под бременем возрастает. Победа над невзгодами.", ["state_creative", "will_vector"]),
("КЛЮЧ", "Отверзает тайны и затворяет зло.", ["will_vector", "meta_synthesis"]),
("ВЕСЫ", "Праведный суд, ничто не сокрыто от правосудия.", ["will_vector", "meta_synthesis"]),
]
for idx, (sym_name, motto_desc, pref_fields) in enumerate(emblem_symbols):
# Извлекаем эмблемы из PDF
emb_imgs = []
if pdf_doc:
# Берем страницы по выборке
p_sample = (idx * 15 + 10) % len(pdf_doc)
emb_imgs = extract_images_from_page(pdf_doc, p_sample, "emblemata", sym_name, f"Петровская эмблема 1705 г.: {sym_name} («{motto_desc}»)")
route_symbol(
term=sym_name,
author="Сборник Петра I (Амстердам/Москва, 1705)",
work="Символы и Емблемата",
text=f"Классическая барочная эмблема петровского свода. Девиз: «{motto_desc}». Символ воплощает государственную, рыцарскую и духовную добродетель.",
preferred_fields=pref_fields,
illustrations=emb_imgs
)
# =========================================================================
# 2. С.Т. МАХЛИНА — «ЗНАКИ, СИМВОЛЫ И КОДЫ КУЛЬТУР ВОСТОКА И ЗАПАДА»
# =========================================================================
makh_path = NEWSOURCES_DIR / "Махлина С.Т. - Знаки, символы и коды культур Востока и Запада - 2019" / "Махлина С.Т. - Знаки, символы и коды культур Востока и Запада - 2019.pdf"
if makh_path.exists():
print("\n📖 [2/4] Обработка С.Т. Махлина — «Знаки, символы и коды культур»...")
doc = fitz.open(makh_path)
makh_topics = [
("ЖИВОТНЫЕ", 9, 21, ["culture_sumer_light", "archetype_category"], "Семиотика животных в мифах Ближнего Востока"),
("СМЕРТЬ", 22, 37, ["culture_sumer_shadow", "chthonic_aspect"], "Семиотика смерти в мифах Ближнего Востока"),
("САД", 50, 72, ["culture_antiquity_light", "cosmogonic_role"], "Сады и парки как коды культуры"),
("МАТЬ", 73, 91, ["cosmogonic_role", "meta_synthesis"], "Семиотика материнской любви"),
("СВАДЬБА", 92, 119, ["state_creative", "culture_antiquity_light"], "Семиотика свадьбы в диалоге культур"),
("ТЮРЬМА", 132, 150, ["chthonic_aspect", "spatial_orientation"], "Семиотика тюрьмы как изоляции и предела"),
("БАБА-ЯГА", 245, 253, ["culture_slavic_shadow", "chthonic_aspect"], "Образ Бабы-Яги в межкультурной коммуникации"),
("ШАМАН", 254, 266, ["archetype_category", "culture_native_american_light"], "Эзотерика шаманизма и ритуал"),
("МЕЗОАМЕРИКА", 267, 274, ["culture_mesoamerica_light", "geometric_archetype"], "Эзотерика игры в мяч в древней Мезоамерике"),
("ИНДЕЙЦЫ", 275, 283, ["culture_native_american_light", "spatial_orientation"], "Риторика тела индейцев Калифорнии"),
("МАСОНСТВО", 284, 296, ["culture_hermetic_light", "meta_synthesis"], "Масонские символы в русской культуре XVIII века"),
("ЧИСЛО", 297, 314, ["mantic_numerology", "meta_synthesis"], "Мистика чисел в сакрально-религиозных представлениях"),
("РОЗА МИРА", 393, 406, ["culture_slavic_light", "meta_synthesis"], "Будущее России по «Розе мира» Даниила Андреева"),
]
for sym_term, start_p, end_p, pref_fields, topic_title in makh_topics:
pages_txt = []
for p in range(start_p, min(end_p, len(doc))):
pages_txt.append(doc[p].get_text())
full_txt = clean_spaces(" ".join(pages_txt))
if len(full_txt) > 80:
route_symbol(
term=sym_term,
author="С.Т. Махлина",
work="Знаки, символы и коды культур Востока и Запада",
text=full_txt[:900],
preferred_fields=pref_fields,
illustrations=[]
)
# =========================================================================
# 3. С.Г. ВУРГАФТ, И.А. УШАКОВ — «СТАРООБРЯДЧЕСТВО» (ЭНЦИКЛОПЕДИЯ)
# =========================================================================
vurg_path = NEWSOURCES_DIR / "Вургафт С.Г., Ушаков И.А. - Старообрядчество. Опыт энциклопедического словаря - 1996.pdf"
if vurg_path.exists():
print("\n📖 [3/4] Обработка С.Г. Вургафт, И.А. Ушаков — «Старообрядчество»...")
doc = fitz.open(vurg_path)
for pno in range(10, min(315, len(doc))):
raw_txt = doc[pno].get_text()
if not raw_txt or len(raw_txt) < 80:
continue
clean_txt = clean_spaces(raw_txt)
# Поиск словарных статей
entries = re.findall(r'([А-ЯЁ]{3,30}(?:\s+[А-ЯЁ]{3,30})?)\s*—\s*([^.]+\.[^.]+\.)', clean_txt)
for head, body in entries:
head_clean = head.strip()
if len(head_clean) < 3 or head_clean in ["АЗЪ", "БУКИ", "ВЕДИ", "ГЛАГОЛЬ", "ДОБРО", "ЕСТЬ"]:
continue
route_symbol(
term=head_clean,
author="С.Г. Вургафт, И.А. Ушаков",
work="Старообрядчество. Опыт энциклопедического словаря",
text=f"{head_clean} — {body}",
preferred_fields=["culture_slavic_light", "meta_synthesis"],
illustrations=[]
)
# =========================================================================
# 4. Г.А. ХЛОПАЧЕВ — «ВЕРХНИЙ ПАЛЕОЛИТ. ОБРАЗЫ, СИМВОЛЫ, ЗНАКИ»
# =========================================================================
khlop_path = NEWSOURCES_DIR / "Хлопачев Г.А. (отв.ред.) - Верхний палеолит. Образы, символы, знаки. Каталог предметов искусства малых форм и уникальных находок верхн.pdf"
if khlop_path.exists():
print("\n📖 [4/4] Обработка Г.А. Хлопачев — «Верхний палеолит. Образы, символы, знаки»...")
doc = fitz.open(khlop_path)
paleo_motifs = [
("МАТЬ", "палеолитическая венера", ["cosmogonic_role", "meta_synthesis"]),
("ОРНАМЕНТ", "геометрический орнамент меандра и шеврона", ["geometric_archetype", "meta_synthesis"]),
("СОЛНЦЕ", "солярный диск с радиальными насечками", ["solar_lunar_axis", "meta_synthesis"]),
("ОЛЕНЬ", "гравировка оленя на костяной пластине", ["archetype_category", "state_creative"]),
("ПТИЦА", "летящая водоплавающая птица верхнего палеолита", ["archetype_category", "spatial_orientation"]),
("МАМОНТ", "скульптурная фигурка мамонта из бивня", ["archetype_category", "meta_synthesis"]),
("СПИРАЛЬ", "спиралевидный орнамент Мальты и Бурети", ["geometric_archetype", "meta_synthesis"]),
]
for pno in range(40, min(160, len(doc))):
raw_txt = doc[pno].get_text()
if not raw_txt or len(raw_txt) < 100:
continue
for term_ru, search_str, pref_fields in paleo_motifs:
if any(w in raw_txt.lower() for w in search_str.split()):
imgs = extract_images_from_page(doc, pno, "paleo", term_ru, f"Палеолитический артефакт: {search_str.capitalize()}")
route_symbol(
term=term_ru,
author="Г.А. Хлопачев (отв. ред.)",
work="Верхний палеолит. Образы, символы, знаки",
text=clean_spaces(raw_txt[:600]),
preferred_fields=pref_fields,
illustrations=imgs
)
break
# =========================================================================
# СОХРАНЕНИЕ
# =========================================================================
print("\n💾 Сохранение обновленных баз данных...")
with open(STAGING_DB_PATH, "w", encoding="utf-8") as f:
json.dump(staging_db, f, ensure_ascii=False, indent=2)
with open(MAIN_DB_PATH, "w", encoding="utf-8") as f:
json.dump(staging_db, f, ensure_ascii=False, indent=2)
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)
with open(CATALOG_PATH, "w", encoding="utf-8") as f:
json.dump(catalog, f, ensure_ascii=False, indent=2)
print("=" * 70)
print("📊 ИТОГОВЫЙ ОТЧЕТ ФИНАЛЬНОГО ЭТАПА:")
print(f" • Обогащено ячеек Оракула 999: +{stats['core_enriched']}")
print(f" • Добавлено внешних символов: +{stats['external_added']}")
print(f" • Всего внешних символов в базе: {len(external_db)}")
print(f" • Новых иллюстраций извлечено: +{stats['illustrations_added']}")
print(f" • Всего иллюстраций в каталоге: {len(catalog)}")
print("=" * 70)
if __name__ == "__main__":
main()