#!/usr/bin/env python3.11 # -*- coding: utf-8 -*- """ detect_language_VAM.py Сканирует все PDF в коллекции "Хранилище библиотеки водного, сельского и мелиоративного хозяйства и ресурсов стран бывшего Туркестана и сопредельных стран": 1. определяет наличие текстового слоя; 2. распознаёт (OCR) страницы 4-9 и по морфологическим/орфографическим маркерам определяет язык/письменность документа (9 категорий); 3. записывает результат в CSV и в метаданные самого PDF. Логика проверена и откалибрована на 12 образцах разных языков (detect_language_TEST11.py) - финальная версия. """ import os import re import csv import time import traceback from concurrent.futures import ProcessPoolExecutor, as_completed # Ограничиваем Tesseract одним потоком на вызов - иначе он сам использует # несколько ядер на один файл, и MAX_WORKERS перестаёт соответствовать # реальной нагрузке на CPU (было обнаружено при load average ~30 на 18 ядер) os.environ["OMP_THREAD_LIMIT"] = "1" os.environ["OMP_NUM_THREADS"] = "1" import fitz # pymupdf import pytesseract from PIL import Image from langdetect import detect as langdetect_detect, LangDetectException import pikepdf # ======================= НАСТРОЙКИ ======================= INPUT_DIR = "/home/dayhanbiz/public_html/3-damja.science/Хранилище библиотеки водного, сельского и мелиоративного хозяйства и ресурсов стран бывшего Туркестана и сопредельных стран" CSV_DIR = "/home/dayhanbiz/public_html/3-damja.science/Обслуживание библиотеки водного, сельского и мелиоративного хозяйства и ресурсов стран бывшего Туркестана и сопредельных стран/Скрипты/Регистрация языка документов" CSV_FILE = os.path.join(CSV_DIR, "language_detection_VAM.csv") # Ограничение на число файлов за один запуск. None = вся коллекция. TEST_LIMIT = None # Сколько файлов обрабатывать одновременно (на сервере 18 ядер - не занимаем все) MAX_WORKERS = 6 # Страницы для OCR (нумерация с 1) - страницы 4-9 OCR_PAGE_RANGE_1INDEXED = (4, 9) OCR_DPI = 200 OLD_RU_THRESHOLD = 0.12 TURKIC_SUFFIX_THRESHOLD = 0.08 MIN_WORDS_FOR_TRUST = 15 # ========================================================================== TURKMEN_CYR_CHARS = set("ӘәҢңӨөҮүҖҗ") TURKMEN_LATIN_CHARS = set("çÇäÄžŽňŇöÖşŞüÜýÝ") CONSONANTS_RU = set("бвгджзклмнпрстфхцчшщБВГДЖЗКЛМНПРСТФХЦЧШЩ") TURKIC_SUFFIXES_CYR = ("дан", "ндан", "лар", "лары", "ыны", "нын", "ынын", "нин", "инин", "нда", "лен", "рин") # Грамматические окончания дореформенной орфографии, состоящие только из обычных # букв (аго/яго/ыя) - переживают любой OCR, даже тот, что не умеет печатать ѣ/ѳ/і/ъ OLD_RU_GRAMMAR_SUFFIXES = ("аго", "яго", "ыя") OLD_RU_GRAMMAR_THRESHOLD = 0.01 TURKIC_SUFFIXES_LATIN = ("nda", "dan", "den", "lar", "ler", "yny", "yna", "yan", "aky", "ynda", "ilen", "daky", "leri", "lary", "inde", "ynyn", "syna", "olan", "aryn") # '-tagat'/'-tagatant' - причастие прошедшего времени (то же явление, что '-تگ' в белуджском на # фарси, только латиницей); 'ent'/'atant' - связка "есть/были", частое отдельное слово BALOCHI_LATIN_SUFFIXES = ("tagat", "tagatant") BALOCHI_LATIN_WORDS = {"ent", "atant"} FARSI_FUNCTION_TOKENS = {"است", "می", "های", "این", "آن", "را"} # Подтверждено данными на образце туркменского на фарси (аблатив/мн.число/локатив + лексика) TURKIC_FARSI_SUFFIXES = ("دان", "لار", "ندا", "لهر") TURKIC_FARSI_WORDS = {"بو", "او", "ایکی"} URDU_MARKERS = {"کا", "کی", "کے", "اور", "ہے", "تھا", "تھی", "تھے", "گیا"} # '-تگ' - причастие прошедшего времени, очень частое в белуджском тексте (не в фарси/урду). BALOCHI_SUFFIXES = ("تگ",) BALOCHI_MARKERS = {"ات"} WORD_RE = re.compile(r"\w+", re.UNICODE) DIGIT_CHARS = set("0123456789۰۱۲۳۴۵۶۷۸۹") def tokenize(text): tokens = WORD_RE.findall(text) return [t for t in tokens if not all(ch in DIGIT_CHARS for ch in t)] def has_text_layer(doc, max_pages_to_check=10, min_chars=200): total_chars = 0 for i in range(min(doc.page_count, max_pages_to_check)): try: total_chars += len(doc.load_page(i).get_text().strip()) except Exception: continue if total_chars >= min_chars: return True return total_chars >= min_chars def render_pages(doc, page_range_1indexed, dpi): start, end = page_range_1indexed indices = [i - 1 for i in range(start, end + 1) if 0 <= i - 1 < doc.page_count] if not indices: indices = list(range(min(doc.page_count, 6))) images = [] zoom = dpi / 72 mat = fitz.Matrix(zoom, zoom) for idx in indices: try: page = doc.load_page(idx) pix = page.get_pixmap(matrix=mat) img = Image.frombytes("RGB", [pix.width, pix.height], pix.samples) images.append(img) except Exception: continue return images def ocr_images(images, lang): all_words = [] all_confs = [] for img in images: try: data = pytesseract.image_to_data(img, lang=lang, output_type=pytesseract.Output.DICT) except Exception: continue for text, conf in zip(data.get("text", []), data.get("conf", [])): text = text.strip() if not text: continue try: c = float(conf) except (ValueError, TypeError): c = -1 if c < 0: continue all_words.append(text) all_confs.append(c) full_text = " ".join(all_words) mean_conf = sum(all_confs) / len(all_confs) if all_confs else 0.0 return full_text, mean_conf, len(all_words) def classify_cyrillic(text): words = tokenize(text) if not words: return "русский современный", 0.0, "нет распознанных слов" old_ru_hits = 0 grammar_hits = 0 turkic_suffix_hits = 0 turkmen_char_hits = 0 for w in words: lw = w.lower() if len(w) >= 3 and w[-1] == "ъ" and w[-2] in CONSONANTS_RU: old_ru_hits += 1 if len(lw) >= 5 and lw.endswith(OLD_RU_GRAMMAR_SUFFIXES): grammar_hits += 1 if any(ch in TURKMEN_CYR_CHARS for ch in w): turkmen_char_hits += 1 if len(w) >= 5 and lw.endswith(TURKIC_SUFFIXES_CYR): turkic_suffix_hits += 1 old_ru_ratio = old_ru_hits / len(words) grammar_ratio = grammar_hits / len(words) turkic_suffix_ratio = turkic_suffix_hits / len(words) evidence = (f"old_ru_ratio={old_ru_ratio:.2f};grammar_ratio={grammar_ratio:.3f};" f"turkmen_chars={turkmen_char_hits};" f"turkic_suffix_ratio={turkic_suffix_ratio:.2f};слов={len(words)}") # Метод исключения: 1. дореформенный (буквы ИЛИ окончания) -> 2. туркменский кириллица -> 3. современный # ВАЖНО: буквы Ә/Ң/Ө/Ү/Җ НЕ используются как решающий признак сами по себе - # они общие для казахского/киргизского/татарского/башкирского/якутского алфавитов # тоже, и одно случайное слово-цитата в лингвистической статье может дать ложное # срабатывание. Решает только устойчивая плотность суффиксов (turkic_suffix_ratio). if old_ru_ratio >= OLD_RU_THRESHOLD or grammar_ratio >= OLD_RU_GRAMMAR_THRESHOLD: return "русский дореформенный", max(old_ru_ratio, grammar_ratio), evidence if turkic_suffix_ratio >= TURKIC_SUFFIX_THRESHOLD: return "туркменский (кириллица)", turkic_suffix_ratio, evidence return "русский современный", 1 - old_ru_ratio, evidence def classify_latin(text): words = tokenize(text) if not words: return "европейский язык (не определён)", 0.0, "нет распознанных слов" turkmen_char_hits = 0 turkic_suffix_hits = 0 balochi_hits = 0 for w in words: lw = w.lower() if any(ch in TURKMEN_LATIN_CHARS for ch in w): turkmen_char_hits += 1 if len(w) >= 5 and lw.endswith(TURKIC_SUFFIXES_LATIN): turkic_suffix_hits += 1 if lw.endswith(BALOCHI_LATIN_SUFFIXES) or lw in BALOCHI_LATIN_WORDS: balochi_hits += 1 n = len(words) turkic_suffix_ratio = turkic_suffix_hits / n balochi_ratio = balochi_hits / n evidence = (f"turkmen_chars={turkmen_char_hits};turkic_suffix_ratio={turkic_suffix_ratio:.2f};" f"balochi_latin_ratio={balochi_ratio:.2f};слов={n}") # Метод исключения: 1. туркменский латиница -> 2. белуджский латиница -> 3. европейский if turkmen_char_hits > 0 or turkic_suffix_ratio >= TURKIC_SUFFIX_THRESHOLD: return "туркменский (латиница)", max(turkic_suffix_ratio, 0.5 if turkmen_char_hits else 0), evidence if balochi_ratio >= 0.02: return "белуджский (латиница)", min(1.0, balochi_ratio * 15), evidence try: lang_code = langdetect_detect(text) except LangDetectException: lang_code = "не определён" return f"европейский язык ({lang_code})", 0.5, evidence def classify_persian_script(text): words = tokenize(text) if not words: return "персидское письмо (не определено уверенно)", 0.0, "нет распознанных слов" n = len(words) turkic_hits = sum(1 for w in words if w.endswith(TURKIC_FARSI_SUFFIXES) or w in TURKIC_FARSI_WORDS) urdu_hits = sum(1 for w in words if w in URDU_MARKERS) balochi_hits = sum(1 for w in words if w in BALOCHI_MARKERS or w.endswith(BALOCHI_SUFFIXES)) farsi_func_hits = sum(1 for w in words if w in FARSI_FUNCTION_TOKENS or w.startswith("می")) turkic_ratio = turkic_hits / n urdu_ratio = urdu_hits / n balochi_ratio = balochi_hits / n farsi_ratio = farsi_func_hits / n evidence = (f"turkic_ratio={turkic_ratio:.3f};urdu_ratio={urdu_ratio:.3f};" f"balochi_ratio={balochi_ratio:.3f};farsi_ratio={farsi_ratio:.3f};слов={n}") # Метод исключения: 1. туркменский -> 2. урду -> 3. фарси -> 4. белуджский (слабый сигнал) if turkic_ratio >= 0.03: return "туркменский (фарси/арабское письмо)", min(1.0, turkic_ratio * 10), evidence if urdu_ratio >= 0.008: return "урду", min(1.0, urdu_ratio * 30), evidence if farsi_ratio >= 0.03: return "фарси", min(1.0, farsi_ratio * 10), evidence if balochi_ratio >= 0.01: return "белуджский (низкая уверенность, проверить вручную)", min(0.4, balochi_ratio * 15), evidence return "персидское письмо (не определено уверенно, возможен белуджский)", 0.2, evidence def extract_pages_text(doc, page_range_1indexed): start, end = page_range_1indexed indices = [i - 1 for i in range(start, end + 1) if 0 <= i - 1 < doc.page_count] if not indices: indices = list(range(min(doc.page_count, 6))) parts = [] for idx in indices: try: parts.append(doc.load_page(idx).get_text()) except Exception: continue return " ".join(parts) def detect_script_family_from_text(text): cyr = sum(1 for ch in text if "\u0400" <= ch <= "\u04FF") lat = sum(1 for ch in text if ch.isascii() and ch.isalpha()) ar = sum(1 for ch in text if "\u0600" <= ch <= "\u06FF" or "\u0750" <= ch <= "\u077F") counts = {"кириллица": cyr, "латиница": lat, "персидское письмо": ar} best = max(counts, key=counts.get) if counts[best] < 100: # мало текста на этих страницах - недостаточно для уверенного решения return None return best def detect_language_from_text_layer(doc): """Быстрый путь: если в PDF уже есть текстовый слой, читаем текст напрямую (без рендеринга и OCR) - это в разы быстрее и точнее, т.к. специальные буквы (Ә, Ń, Ö, ٹ и т.д.) в настоящем тексте не искажены.""" text = extract_pages_text(doc, OCR_PAGE_RANGE_1INDEXED) script_family = detect_script_family_from_text(text) if script_family is None: # текста на этих страницах мало (например, картинки/таблицы) - подстрахуемся через OCR return detect_language(doc) if script_family == "кириллица": lang, conf, evidence = classify_cyrillic(text) elif script_family == "латиница": lang, conf, evidence = classify_latin(text) else: lang, conf, evidence = classify_persian_script(text) return lang, script_family, conf, evidence def detect_language(doc): images = render_pages(doc, OCR_PAGE_RANGE_1INDEXED, OCR_DPI) if not images: return "не определён", "нет страниц для OCR", 0.0, "" text_rus, conf_rus, words_rus = ocr_images(images, "rus") text_fas, conf_fas, words_fas = ocr_images(images, "fas") text_eng, conf_eng, words_eng = ocr_images(images, "eng") candidates = [] if words_rus >= MIN_WORDS_FOR_TRUST: candidates.append(("кириллица", conf_rus, text_rus)) if words_fas >= MIN_WORDS_FOR_TRUST: candidates.append(("персидское письмо", conf_fas, text_fas)) if words_eng >= MIN_WORDS_FOR_TRUST: candidates.append(("латиница", conf_eng, text_eng)) if not candidates: best_guess = max( [("кириллица", conf_rus, text_rus, words_rus), ("персидское письмо", conf_fas, text_fas, words_fas), ("латиница", conf_eng, text_eng, words_eng)], key=lambda x: x[3] ) return "не определён (мало текста на странице)", best_guess[0], 0.0, "" script_family, _, best_text = max(candidates, key=lambda x: x[1]) if script_family == "кириллица": lang, conf, evidence = classify_cyrillic(best_text) elif script_family == "латиница": lang, conf, evidence = classify_latin(best_text) else: lang, conf, evidence = classify_persian_script(best_text) return lang, script_family, conf, evidence def write_pdf_metadata(pdf_path, result): try: with pikepdf.open(pdf_path, allow_overwriting_input=True) as pdf: pdf.docinfo["/HasTextLayer"] = pikepdf.String(str(result["has_text_layer"])) pdf.docinfo["/DetectedLanguage"] = pikepdf.String(result["detected_language"]) pdf.docinfo["/ScriptFamily"] = pikepdf.String(result["script_family"]) pdf.docinfo["/LanguageConfidence"] = pikepdf.String(f'{result["confidence"]:.2f}') pdf.docinfo["/LanguageEvidence"] = pikepdf.String(result["evidence"]) pdf.save() return True, "" except Exception as e: return False, str(e) def process_file(pdf_path, input_dir): rel_path = os.path.relpath(pdf_path, input_dir) result = { "relative_path": rel_path, "absolute_path": pdf_path, "page_count": "", "has_text_layer": "", "script_family": "", "detected_language": "", "confidence": 0.0, "evidence": "", "status": "ok", "error_message": "", "scan_date": time.strftime("%Y-%m-%d %H:%M:%S"), } try: doc = fitz.open(pdf_path) if doc.is_encrypted: doc.authenticate("") result["page_count"] = doc.page_count result["has_text_layer"] = has_text_layer(doc) if result["has_text_layer"]: lang, script_family, conf, evidence = detect_language_from_text_layer(doc) else: lang, script_family, conf, evidence = detect_language(doc) result["detected_language"] = lang result["script_family"] = script_family result["confidence"] = conf result["evidence"] = evidence doc.close() ok, err = write_pdf_metadata(pdf_path, result) if not ok: result["status"] = "ошибка записи метаданных" result["error_message"] = err except Exception as e: result["status"] = "ошибка" result["error_message"] = f"{e}\n{traceback.format_exc(limit=1)}" return result CSV_FIELDS = ["relative_path", "absolute_path", "page_count", "has_text_layer", "script_family", "detected_language", "confidence", "evidence", "status", "error_message", "scan_date"] def load_already_processed(csv_path): processed = set() if os.path.exists(csv_path): with open(csv_path, encoding="utf-8", newline="") as f: reader = csv.DictReader(f, delimiter=";") for row in reader: processed.add(row.get("relative_path", "")) return processed def main(): os.makedirs(CSV_DIR, exist_ok=True) already = load_already_processed(CSV_FILE) print(f"Уже обработано ранее (будет пропущено): {len(already)}") all_pdfs = [] for root, _, files in os.walk(INPUT_DIR): for fn in files: if fn.lower().endswith(".pdf"): all_pdfs.append(os.path.join(root, fn)) to_process = [p for p in all_pdfs if os.path.relpath(p, INPUT_DIR) not in already] print(f"Всего PDF найдено: {len(all_pdfs)}. К обработке: {len(to_process)}") if TEST_LIMIT is not None: to_process = to_process[:TEST_LIMIT] print(f"⚠️ Ограничение TEST_LIMIT={TEST_LIMIT}: обрабатываем только {len(to_process)} файлов") file_exists = os.path.exists(CSV_FILE) csv_f = open(CSV_FILE, mode="a", encoding="utf-8", newline="") writer = csv.DictWriter(csv_f, fieldnames=CSV_FIELDS, delimiter=";") if not file_exists: writer.writeheader() csv_f.flush() done = 0 total = len(to_process) start_time = time.time() with ProcessPoolExecutor(max_workers=MAX_WORKERS) as executor: futures = {executor.submit(process_file, p, INPUT_DIR): p for p in to_process} for fut in as_completed(futures): result = fut.result() writer.writerow(result) csv_f.flush() done += 1 elapsed = time.time() - start_time print(f"[{done}/{total}] {result['relative_path']} -> " f"{result['detected_language']} (текст.слой={result['has_text_layer']}, " f"статус={result['status']}) [{elapsed:.0f}s]") csv_f.close() print(f"\n🎉 Готово! Обработано в этом запуске: {done}") print(f"📁 CSV: {CSV_FILE}") if __name__ == "__main__": main()