#!/usr/bin/env python3.11 # -*- coding: utf-8 -*- """ fetch_corpus_old_ru.py Скачивает корпус подлинных текстов в дореформенной орфографии с Викитеки (ru.wikisource.org, категория "Дореформенная орфография") - для использования как обучающие данные при дообучении Tesseract (rus -> rus_old). Сначала пытается набрать тематически близкие статьи (Средняя Азия, Туркестан, этнография, история) - они дадут более подходящий словарь для OCR вашей коллекции. Затем добирает недостающий объём обычными текстами из категории. """ import os import re import csv import time import requests API_URL = "https://ru.wikisource.org/w/api.php" CATEGORY = "Категория:Дореформенная орфография" OUTPUT_DIR = "/home/dayhanbiz/public_html/3-damja.science/ТЕСТ_образцы_языков/OCR_дообучение_дореформенный/Корпус_wikisource" MANIFEST_FILE = os.path.join(OUTPUT_DIR, "manifest.csv") TARGET_PAGES = 400 # сколько страниц набрать всего MIN_WORDS = 150 # пропускаем слишком короткие страницы (заготовки, короткие стихи) REQUEST_DELAY = 1.5 # вежливая пауза между запросами к API TOPICAL_KEYWORDS = [ "Туркестанъ", "Туркестан", "Средняя Азія", "Средняя Азия", "этнографія", "этнография", "Бухара", "Хива", "Коканд", "Закаспійск", "Закаспийск", "туркмен", "киргиз", "сартъ", ] HEADERS = { "User-Agent": "OldRuOcrCorpusBot/1.0 (personal non-commercial project, " "collecting pre-reform Russian orthography training corpus)" } WORD_RE = re.compile(r"\w+", re.UNICODE) def api_get(params, max_retries=4): params = dict(params) params["format"] = "json" for attempt in range(1, max_retries + 1): resp = requests.get(API_URL, params=params, headers=HEADERS, timeout=30) if resp.status_code == 429: wait = min(int(resp.headers.get("Retry-After", 3 * attempt)), 15) print(f" 429 Too Many Requests - жду {wait}с (попытка {attempt}/{max_retries})", flush=True) time.sleep(wait) continue if resp.status_code >= 500: wait = min(3 * attempt, 15) print(f" Ошибка сервера {resp.status_code} - жду {wait}с (попытка {attempt}/{max_retries})", flush=True) time.sleep(wait) continue resp.raise_for_status() return resp.json() raise RuntimeError(f"Не удалось получить ответ после {max_retries} попыток: {params}") import json CACHE_FILE_NAME = "_titles_cache.json" def get_category_members(category, namespace=0): """Возвращает список заголовков страниц в категории (только основное пространство имён). Сохраняет прогресс на диск - при повторном запуске продолжает с места остановки.""" cache_path = os.path.join(OUTPUT_DIR, CACHE_FILE_NAME) titles = [] cmcontinue = None if os.path.exists(cache_path): with open(cache_path, encoding="utf-8") as f: cache = json.load(f) titles = cache.get("titles", []) cmcontinue = cache.get("cmcontinue") if cmcontinue: print(f" продолжаю сбор списка с сохранённого места ({len(titles)} заголовков уже есть)", flush=True) else: print(f" список уже был полностью собран ранее ({len(titles)} заголовков)", flush=True) return titles while True: params = { "action": "query", "list": "categorymembers", "cmtitle": category, "cmnamespace": namespace, "cmlimit": 500, } if cmcontinue: params["cmcontinue"] = cmcontinue data = api_get(params) members = data.get("query", {}).get("categorymembers", []) titles.extend(m["title"] for m in members) print(f" получено заголовков: {len(titles)}", flush=True) cont = data.get("continue", {}).get("cmcontinue") # сохраняем прогресс после каждой страницы - при обрыве не теряем накопленное with open(cache_path, "w", encoding="utf-8") as f: json.dump({"titles": titles, "cmcontinue": cont}, f, ensure_ascii=False) time.sleep(REQUEST_DELAY) if not cont: break cmcontinue = cont if len(titles) % 5000 == 0: print(" небольшая дополнительная пауза (5с) для вежливости к API...", flush=True) time.sleep(5) return titles def get_plain_text(title): """Получает чистый текст страницы (без вики-разметки) через TextExtracts.""" params = { "action": "query", "prop": "extracts", "explaintext": 1, "titles": title, } data = api_get(params) pages = data.get("query", {}).get("pages", {}) for page in pages.values(): return page.get("extract", "") return "" def sanitize_filename(title, max_bytes=150): name = re.sub(r'[\\/*?:"<>|]', "", title) name = re.sub(r"\s+", " ", name).strip() encoded = name.encode("utf-8") if len(encoded) > max_bytes: encoded = encoded[:max_bytes] while encoded: try: name = encoded.decode("utf-8") break except UnicodeDecodeError: encoded = encoded[:-1] return name + ".txt" def main(): os.makedirs(OUTPUT_DIR, exist_ok=True) print("Получаю список страниц категории...", flush=True) all_titles = get_category_members(CATEGORY) print(f"Всего страниц в категории (основное пространство имён): {len(all_titles)}", flush=True) # Сортируем: сначала тематически близкие, потом остальные (в исходном порядке) def is_topical(title): return any(kw.lower() in title.lower() for kw in TOPICAL_KEYWORDS) topical = [t for t in all_titles if is_topical(t)] other = [t for t in all_titles if not is_topical(t)] print(f"Из них тематически близких (по ключевым словам в заголовке): {len(topical)}", flush=True) ordered_titles = topical + other manifest_rows = [] saved = 0 used_names = set() file_exists = os.path.exists(MANIFEST_FILE) if file_exists: with open(MANIFEST_FILE, encoding="utf-8", newline="") as f: reader = csv.DictReader(f, delimiter=";") for row in reader: used_names.add(row["title"]) saved = len(used_names) print(f"Уже скачано ранее (будет пропущено): {saved}", flush=True) csv_f = open(MANIFEST_FILE, mode="a", encoding="utf-8", newline="") writer = csv.DictWriter(csv_f, fieldnames=["title", "topical", "word_count", "filename"], delimiter=";") if not file_exists: writer.writeheader() used_filenames = set(os.listdir(OUTPUT_DIR)) consecutive_failures = 0 FAILURE_COOLDOWN_AFTER = 3 # после скольких неудач подряд делать долгую паузу COOLDOWN_SECONDS = 300 # 5 минут for title in ordered_titles: if saved >= TARGET_PAGES: break if title in used_names: continue try: text = get_plain_text(title) consecutive_failures = 0 except Exception as e: consecutive_failures += 1 print(f" ошибка при получении '{title}': {e} (неудач подряд: {consecutive_failures})", flush=True) if consecutive_failures >= FAILURE_COOLDOWN_AFTER: print(f" похоже на длительное ограничение API - жду {COOLDOWN_SECONDS}с перед следующей попыткой...", flush=True) time.sleep(COOLDOWN_SECONDS) consecutive_failures = 0 else: time.sleep(REQUEST_DELAY) continue time.sleep(REQUEST_DELAY) time.sleep(REQUEST_DELAY) word_count = len(WORD_RE.findall(text)) if word_count < MIN_WORDS: continue filename = sanitize_filename(title) base, ext = os.path.splitext(filename) counter = 2 while filename in used_filenames: filename = f"{base} ({counter}){ext}" counter += 1 used_filenames.add(filename) with open(os.path.join(OUTPUT_DIR, filename), "w", encoding="utf-8") as f: f.write(text) writer.writerow({ "title": title, "topical": is_topical(title), "word_count": word_count, "filename": filename, }) csv_f.flush() saved += 1 print(f"[{saved}/{TARGET_PAGES}] {title} ({word_count} слов)", flush=True) csv_f.close() print(f"\n🎉 Готово! Сохранено страниц: {saved}", flush=True) print(f"📁 Корпус: {OUTPUT_DIR}") print(f"📋 Манифест: {MANIFEST_FILE}") if __name__ == "__main__": main()