147 lines
4.8 KiB
Python
147 lines
4.8 KiB
Python
"""
|
||
Обход {% tree %} страниц Яндекс Вики через Playwright.
|
||
|
||
Находит дочерние страницы, которые не видны через API (рендерятся динамически).
|
||
Возвращает список slug-ов которых нет в wiki_pages — их нужно добавить в ROOT_SLUGS.
|
||
|
||
Запуск:
|
||
python wiki_tree_crawler.py
|
||
"""
|
||
import re
|
||
import time
|
||
from pathlib import Path
|
||
from dotenv import load_dotenv
|
||
|
||
load_dotenv(Path(__file__).parent / '.env')
|
||
|
||
from playwright.sync_api import sync_playwright
|
||
|
||
AUTH_FILE = Path(__file__).parent / 'wiki_auth.json'
|
||
WIKI_BASE = 'https://wiki.yandex.ru'
|
||
DELAY = 0.5 # пауза между запросами
|
||
|
||
|
||
def slug_from_url(url: str) -> str | None:
|
||
"""Извлечь slug из URL wiki.yandex.ru/slug/..."""
|
||
match = re.match(r'https?://wiki\.yandex\.ru/([^?#]+?)/?$', url)
|
||
if not match:
|
||
return None
|
||
slug = match.group(1).strip('/')
|
||
# Отфильтровать служебные страницы
|
||
if any(slug.startswith(p) for p in ['.files', 'homepage/', 'cdp-']):
|
||
return None
|
||
return slug
|
||
|
||
|
||
def get_child_slugs(page, parent_slug: str) -> list[str]:
|
||
"""Открыть страницу и собрать все ссылки на дочерние страницы."""
|
||
url = f'{WIKI_BASE}/{parent_slug}'
|
||
page.goto(url, wait_until='networkidle', timeout=30000)
|
||
time.sleep(DELAY)
|
||
|
||
# Ищем ссылки в левом навигационном дереве и в содержимом страницы
|
||
links = page.eval_on_selector_all(
|
||
'a[href]',
|
||
'els => els.map(el => el.href)'
|
||
)
|
||
|
||
child_slugs = []
|
||
for link in links:
|
||
slug = slug_from_url(link)
|
||
if slug and slug.startswith(parent_slug + '/') and slug != parent_slug:
|
||
child_slugs.append(slug)
|
||
|
||
return list(set(child_slugs))
|
||
|
||
|
||
def find_tree_pages_in_db() -> list[tuple[str, str]]:
|
||
"""Получить из Supabase все страницы с {% tree %} в контенте."""
|
||
import sys
|
||
sys.path.insert(0, str(Path(__file__).parent))
|
||
from supabase import SupabaseManager
|
||
|
||
db = SupabaseManager()
|
||
db.connect()
|
||
db.cursor.execute("""
|
||
SELECT slug, value->>'title' AS title
|
||
FROM wiki_pages
|
||
WHERE value->>'content' LIKE '%{%% tree%%}%'
|
||
ORDER BY slug
|
||
""")
|
||
rows = db.cursor.fetchall()
|
||
db.close()
|
||
return rows
|
||
|
||
|
||
def get_all_known_slugs() -> set[str]:
|
||
"""Все slug-и которые уже есть в wiki_pages."""
|
||
import sys
|
||
sys.path.insert(0, str(Path(__file__).parent))
|
||
from supabase import SupabaseManager
|
||
|
||
db = SupabaseManager()
|
||
db.connect()
|
||
db.cursor.execute("SELECT slug FROM wiki_pages")
|
||
slugs = {row[0] for row in db.cursor.fetchall()}
|
||
db.close()
|
||
return slugs
|
||
|
||
|
||
def main():
|
||
if not AUTH_FILE.exists():
|
||
print(f'✗ Файл авторизации не найден: {AUTH_FILE}')
|
||
print(' Сначала запусти: python wiki_auth.py')
|
||
return
|
||
|
||
ROOT_PREFIX = 'upravlenie-analitiki'
|
||
|
||
print('Загружаем список страниц с {% tree %} из Supabase...')
|
||
all_tree_pages = find_tree_pages_in_db()
|
||
tree_pages = [
|
||
(s, t) for s, t in all_tree_pages
|
||
if s == ROOT_PREFIX or s.startswith(ROOT_PREFIX + '/')
|
||
]
|
||
known_slugs = get_all_known_slugs()
|
||
print(f'Страниц с {{% tree %}} (всего): {len(all_tree_pages)}, в разделе УА: {len(tree_pages)}')
|
||
print()
|
||
|
||
new_slugs = []
|
||
|
||
with sync_playwright() as p:
|
||
browser = p.chromium.launch(headless=True)
|
||
context = browser.new_context(storage_state=str(AUTH_FILE))
|
||
page = context.new_page()
|
||
|
||
for slug, title in tree_pages:
|
||
print(f'→ {slug} «{title}»')
|
||
try:
|
||
children = get_child_slugs(page, slug)
|
||
new = [s for s in children if s not in known_slugs]
|
||
print(f' Найдено дочерних: {len(children)}, новых: {len(new)}')
|
||
for s in sorted(children):
|
||
marker = ' ← NEW' if s in new else ''
|
||
print(f' {s}{marker}')
|
||
new_slugs.extend(new)
|
||
except Exception as e:
|
||
print(f' ✗ Ошибка: {e}')
|
||
print()
|
||
|
||
browser.close()
|
||
|
||
new_slugs = list(set(new_slugs))
|
||
|
||
if new_slugs:
|
||
print('=' * 60)
|
||
print(f'Найдено {len(new_slugs)} новых slug-ов для ROOT_SLUGS:')
|
||
print()
|
||
for s in sorted(new_slugs):
|
||
print(f" '{s}',")
|
||
print()
|
||
print('Добавь их в ROOT_SLUGS в wiki_sync.py и запусти wiki_sync.py')
|
||
else:
|
||
print('✓ Новых страниц не найдено')
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|