import os import json import re import shutil from pathlib import Path import fitz # PyMuPDF import pytesseract from PIL import Image from bs4 import BeautifulSoup from datetime import datetime from concurrent.futures import ProcessPoolExecutor, as_completed BASE_DIR = Path("/home/dayhanbiz/public_html/Библиотека 3-DAMJA") CATALOG_DIR = BASE_DIR / "catalog" TEXTS_DIR = BASE_DIR / "extracted_texts" OLD_RUS_DIR = BASE_DIR / "documents_in_old_russian" EURO_DIR = BASE_DIR / "documents_in_european_languages" ME_DIR = BASE_DIR / "documents_in_ME_languages" for d in [TEXTS_DIR, OLD_RUS_DIR, EURO_DIR, ME_DIR]: d.mkdir(parents=True, exist_ok=True) def is_old_russian_text(text: str) -> bool: """ Проверяет, является ли текст дореформенным русским. Условие: сумма букв і, ъ, ѣ составляет > 2% от общего числа букв (кириллицы). """ if not text or len(text.strip()) < 50: return False # Считаем общее количество букв кириллицы cyrillic_letters = re.findall(r'[а-яёіѣъА-ЯЁІѢЪ]', text) total_cyr = len(cyrillic_letters) if total_cyr == 0: return False # Считаем специфические дореформенные символы (включая регистр) old_chars = len(re.findall(r'[іѣъІѢЪ]', text)) percentage = (old_chars / total_cyr) * 100 return percentage > 2.0 def classify_and_route_pdf(pdf_path: Path) -> str: """ Проверяет страницы 5, 6, 7, 8, 9 (индексы 4, 5, 6, 7, 8). Если текстовый слой содержит дореформенный русский (>2%), проверяем дальше. """ try: doc = fitz.open(pdf_path) total_pages = len(doc) if total_pages == 0: doc.close() return 'modern_rus' # Целевые страницы: 5, 6, 7, 8, 9 (если есть) target_pages = [p for p in [4, 5, 6, 7, 8] if p < total_pages] if not target_pages: target_pages = [p for p in range(min(5, total_pages))] sample_text = "" for p_idx in target_pages: page = doc[p_idx] txt = page.get_text() # Если на этих страницах мало текста (например, сканы), делаем быстрый OCR для проверки if len(txt.strip()) < 30: pix = page.get_pixmap(dpi=100) img_path = BASE_DIR / f"search_databases/temp_route_{os.getpid()}.png" pix.save(str(img_path)) try: image = Image.open(img_path) txt = pytesseract.image_to_string(image, lang='rus', config='--psm 6') except: pass finally: if img_path.exists(): img_path.unlink() sample_text += txt + "\n" doc.close() # Проверяем собранный текст на дореформенность if is_old_russian_text(sample_text): return 'old_rus' # Проверка на латиницу / европейские языки lat_count = len(re.findall(r'[a-zA-Z]', sample_text)) cyr_count = len(re.findall(r'[а-яА-ЯёЁіѣъ]', sample_text)) if lat_count > 100 and lat_count > (cyr_count * 3): path_str = str(pdf_path).lower() if any(me_keyword in path_str for me_keyword in ['persian', 'arabic', 'urdu', 'farsi', 'me_', 'oriental']): return 'me' return 'euro' return 'modern_rus' except Exception as e: return 'modern_rus' def extract_text_from_pdf(pdf_path: Path, output_book_dir: Path): try: doc = fitz.open(pdf_path) page_records = [] for page_num in range(len(doc)): page = doc[page_num] text = page.get_text() # Если слой текста пуст или слишком бедный, подключаем OCR if len(text.strip()) < 50: pix = page.get_pixmap(dpi=150) img_path = BASE_DIR / f"search_databases/temp_page_{os.getpid()}.png" pix.save(str(img_path)) try: image = Image.open(img_path) text = pytesseract.image_to_string(image, lang='rus', config='--psm 6') finally: if img_path.exists(): img_path.unlink() page_file_path = output_book_dir / f"page_{page_num + 1:04d}.txt" with open(page_file_path, "w", encoding="utf-8") as f: f.write(text) page_records.append({"page_number": page_num + 1, "txt_path": str(page_file_path)}) doc.close() return page_records except Exception as e: return [] def extract_text_from_html(html_path: Path, output_book_dir: Path): try: with open(html_path, "r", encoding="utf-8", errors="ignore") as f: soup = BeautifulSoup(f.read(), "lxml") for script in soup(["script", "style"]): script.decompose() text = soup.get_text(separator="\n", strip=True) page_file_path = output_book_dir / "page_0001.txt" with open(page_file_path, "w", encoding="utf-8") as f: f.write(text) return [{"page_number": 1, "txt_path": str(page_file_path)}] except Exception as e: return [] def process_single_item(item): file_path = Path(item["path"]) file_hash = item["hash"] ext = item["extension"] if ext == '.pdf' and file_path.exists(): category = classify_and_route_pdf(file_path) if category == 'old_rus': # Дополнительная проверка: открываем документ полностью и смотрим уже имеющийся текстовый слой try: doc = fitz.open(file_path) full_extracted_text = "" for page in doc: full_extracted_text += page.get_text() doc.close() # Если в документе ЕСТЬ готовый цифровой текст И он тоже удовлетворяет критерию дореформенного (>2%) if len(full_extracted_text.strip()) > 200 and is_old_russian_text(full_extracted_text): # Индексируем штатно (как современный русский) в extracted_texts pass # проваливаемся ниже в стандартную обработку извлечения текстов else: # Слой плохой или отсутствует — копируем файл в папку для отдельной работы dest_file = OLD_RUS_DIR / file_path.name if not dest_file.exists(): shutil.copy2(file_path, dest_file) return 1 except: dest_file = OLD_RUS_DIR / file_path.name if not dest_file.exists(): shutil.copy2(file_path, dest_file) return 1 elif category == 'euro': dest_file = EURO_DIR / file_path.name if not dest_file.exists(): shutil.copy2(file_path, dest_file) return 1 elif category == 'me': dest_file = ME_DIR / file_path.name if not dest_file.exists(): shutil.copy2(file_path, dest_file) return 1 # Стандартная обработка (современный русский или дореформенный с хорошим слоем) book_text_dir = TEXTS_DIR / file_hash book_text_dir.mkdir(parents=True, exist_ok=True) if any(book_text_dir.glob("page_*.txt")): return 1 if ext == '.pdf': extract_text_from_pdf(file_path, book_text_dir) elif ext in ['.html', '.htm']: extract_text_from_html(file_path, book_text_dir) return 1 def run_extraction(): print("=== Запуск обновленного интеллектуального экстрактора (с уточнением по стр. 5-9) ===") print(f"Время: {datetime.now()}\n") max_workers = 12 for cat_name in ["list_history.json", "list_vam.json"]: cat_file = CATALOG_DIR / cat_name if not cat_file.exists(): continue print(f"Обработка каталога: {cat_name} в {max_workers} потоков") with open(cat_file, "r", encoding="utf-8") as f: items = json.load(f) processed_count = 0 with ProcessPoolExecutor(max_workers=max_workers) as executor: futures = [executor.submit(process_single_item, item) for item in items] for future in as_completed(futures): processed_count += future.result() if processed_count % 500 == 0: print(f" → Обработано документов: {processed_count} / {len(items)}") print(f"✅ Каталог {cat_name} успешно обработан.\n") if __name__ == "__main__": run_extraction()