126 lines
3.8 KiB
Python
126 lines
3.8 KiB
Python
"""
|
||
Модуль для обхода страниц 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
|