""" Модуль для обхода страниц Yandex Wiki API """ import re import time import hashlib import requests import pendulum WIKI_API = 'https://api.wiki.yandex.net/v1' def _make_headers(api_token: str, org_id: str) -> dict: return { 'Authorization': f'OAuth {api_token}', 'X-Org-Id': org_id, } def get_page(slug: str, api_token: str, org_id: str) -> dict | None: """Получить страницу по slug. Возвращает None если не найдена.""" try: resp = requests.get( f'{WIKI_API}/pages', headers=_make_headers(api_token, org_id), params={'slug': slug, 'fields': 'content,attributes'}, timeout=15, ) if resp.status_code == 200: return resp.json() return None except requests.RequestException as e: print(f' ✗ Ошибка при запросе slug={slug}: {e}') return None def extract_slugs_from_content(content: str) -> list[str]: """Извлечь slug-ссылки из wiki-разметки страницы.""" slugs = [] # [[slug]] и [[slug|текст]] slugs += re.findall(r'\[\[([^\]|#]+)', content) # ((https://wiki.yandex.ru/slug текст)) или ((https://wiki.yandex.ru/slug)) url_matches = re.findall(r'https://wiki\.yandex\.ru/([^\s)#]+)', content) slugs += url_matches # Нормализуем: убираем пробелы и слэши по краям result = [] for s in slugs: s = s.strip().strip('/') if s and not s.startswith('http') and len(s) < 200: result.append(s) return result def content_hash(content: str) -> str: """MD5-хеш содержимого страницы для определения изменений.""" return hashlib.md5(content.encode('utf-8')).hexdigest() def crawl( root_slugs: list[str], api_token: str, org_id: str, max_depth: int = 4, delay: float = 0.3, ) -> list[dict]: """ Обойти дерево страниц начиная с root_slugs. Args: root_slugs: Список стартовых slug-ов (например ['upravlenie-analitiki']) api_token: OAuth-токен org_id: ID организации max_depth: Максимальная глубина обхода delay: Пауза между запросами в секундах Returns: Список словарей с данными страниц """ dt_load = pendulum.now(tz='Europe/Moscow') visited: set[str] = set() pages: list[dict] = [] def _crawl(slug: str, depth: int): if depth > max_depth or slug in visited: return visited.add(slug) print(f' {" " * depth}→ {slug}') page = get_page(slug, api_token, org_id) time.sleep(delay) if not page: return raw_content = page.get('content', '') attrs = page.get('attributes', {}) modified_at = attrs.get('modified_at') or attrs.get('updated_at') pages.append({ 'pg_load_dttm': dt_load, 'slug': page.get('slug', slug), 'wiki_page_id': page.get('id'), 'title': page.get('title', ''), 'page_type': page.get('page_type', ''), 'modified_at': modified_at, 'content': raw_content, 'content_hash': content_hash(raw_content), }) # Рекурсивно обходим ссылки for child_slug in extract_slugs_from_content(raw_content): _crawl(child_slug, depth + 1) print(f'Начинаем обход wiki, корневые разделы: {root_slugs}') for root in root_slugs: _crawl(root, 0) print(f'✓ Обход завершён. Найдено страниц: {len(pages)}') return pages