diff --git a/core/utils/web_utils.py b/core/utils/web_utils.py new file mode 100644 index 0000000..48b1e2d --- /dev/null +++ b/core/utils/web_utils.py @@ -0,0 +1,242 @@ +# -*- coding: utf-8 -*- +""" +Утилиты для загрузки веб-страниц и PDF, извлечения основного текста, +рекурсивного обхода ссылок. + +Использует aiohttp для асинхронных запросов, BeautifulSoup для парсинга, +readability-lxml для извлечения контента (опционально). + +Параметры (max_pages, max_depth) берутся из конфига при вызове из xmpp_client. + +Импортируется в: +- indexing_worker.py (fetch_any_url, crawl_url) +- возможно, в других функциях при обработке URL. +""" + +import asyncio +import aiohttp +import logging +from typing import Optional, List, Tuple +from urllib.parse import urljoin, urlparse + +from bs4 import BeautifulSoup +import aiofiles +import tempfile +import os + +# Опциональная библиотека readability-lxml улучшает извлечение основного текста +try: + from readability import Document + HAS_READABILITY = True +except ImportError: + HAS_READABILITY = False + logging.warning("readability-lxml не установлен, используем BeautifulSoup без очистки") + +logger = logging.getLogger(__name__) + + +async def fetch_any_url(url: str, timeout: int = 30) -> Tuple[Optional[str], Optional[str]]: + """ + Универсальная загрузка контента по URL. + + Аргументы: + url (str): целевой адрес. + timeout (int): таймаут запроса в секундах. + + Возвращает: + Tuple[Optional[str], Optional[str]]: (заголовок, текст). + Для HTML: заголовок страницы и очищенный основной текст. + Для PDF: имя файла (базовая часть URL) и извлечённый текст. + В случае ошибки возвращает (None, None). + + Примечания: + - Для PDF используется временный файл, который удаляется после чтения. + - Поддерживаются только HTTP/HTTPS. + - Таймаут применяется к запросу в целом (чтение + соединение). + """ + try: + async with aiohttp.ClientSession() as session: + async with session.get(url, timeout=timeout) as resp: + if resp.status != 200: + logger.error(f"HTTP {resp.status} для {url}") + return None, None + content_type = resp.headers.get('Content-Type', '').lower() + # Обработка PDF + if 'application/pdf' in content_type or url.lower().endswith('.pdf'): + # Сохраняем PDF во временный файл + with tempfile.NamedTemporaryFile(delete=False, suffix='.pdf') as tmp: + tmp_path = tmp.name + async with aiofiles.open(tmp_path, 'wb') as f: + await f.write(await resp.read()) + try: + from pypdf import PdfReader + reader = PdfReader(tmp_path) + text = "\n".join([page.extract_text() or "" for page in reader.pages]) + title = os.path.basename(url) + return title, text + except ImportError: + logger.error("pypdf не установлен") + return None, None + finally: + os.unlink(tmp_path) + else: + # Обычная HTML-страница + html = await resp.text() + title, text = extract_text_from_html(html, url) + return title, text + except asyncio.TimeoutError: + logger.error(f"Таймаут при загрузке {url}") + return None, None + except Exception as e: + logger.error(f"Ошибка загрузки {url}: {e}") + return None, None + + +def extract_text_from_html(html: str, url: str = "") -> Tuple[Optional[str], str]: + """ + Извлекает заголовок и основной текст из HTML. + + Аргументы: + html (str): исходный HTML-код. + url (str): URL страницы (используется для формирования заголовка, если теги отсутствуют). + + Возвращает: + Tuple[Optional[str], str]: (заголовок, очищенный текст). + Заголовок может быть None, если не удалось извлечь. + + Алгоритм извлечения заголовка (по приоритету): + 1. + 2. <meta property="og:title"> + 3. <h1> + 4. Элементы с классами 'title', 'article-title', 'page-title', 'document-title' + 5. Последний сегмент пути URL + 6. Домен (как запасной вариант) + """ + soup = BeautifulSoup(html, 'lxml') + + # ---- 1. Извлечение заголовка ---- + title = None + if soup.title and soup.title.string: + title = soup.title.string.strip() + if not title: + og_title = soup.find('meta', property='og:title') + if og_title and og_title.get('content'): + title = og_title['content'].strip() + if not title: + h1 = soup.find('h1') + if h1: + title = h1.get_text(strip=True) + if not title: + # Поиск по распространённым классам + for cls in ['title', 'article-title', 'page-title', 'document-title']: + elem = soup.find(class_=cls) + if elem: + title = elem.get_text(strip=True) + break + if not title: + # Последняя надежда: использовать последний сегмент URL + parsed = urlparse(url) + if parsed.path and parsed.path != '/': + last_part = parsed.path.rstrip('/').split('/')[-1] + title = last_part.replace('-', ' ').replace('_', ' ').title() + else: + title = parsed.netloc + + # ---- 2. Извлечение основного текста ---- + if HAS_READABILITY: + # Используем readability для выделения «читаемой» части + doc = Document(html) + text = doc.summary() + # Если readability дал заголовок, но мы его не получили ранее – используем + if not title and doc.title(): + title = doc.title() + # Очищаем полученный HTML от лишних тегов + text_soup = BeautifulSoup(text, 'lxml') + text = text_soup.get_text(separator='\n', strip=True) + else: + # Удаляем скрипты, стили, навигацию, футеры, хедеры + for script in soup(["script", "style", "nav", "footer", "header"]): + script.decompose() + text = soup.get_text(separator='\n', strip=True) + + logger.debug(f"Извлечён заголовок: {repr(title)} для {url}") + return title, text + + +async def crawl_url( + start_url: str, + max_depth: int, + max_pages: int, + domain_restrict: bool = True, +) -> List[Tuple[str, str, str]]: + """ + Рекурсивно обходит ссылки на том же домене (или не ограничивая, если domain_restrict=False). + + Аргументы: + start_url (str): начальная страница. + max_depth (int): максимальная глубина перехода по ссылкам. + max_pages (int): максимальное количество страниц для обработки. + domain_restrict (bool): ограничиваться ли тем же доменом (по умолчанию True). + + Возвращает: + List[Tuple[str, str, str]]: список кортежей (url, title, text). + """ + visited = set() + to_visit = [(start_url, 0)] + results = [] + visited.add(start_url) + + while to_visit and len(results) < max_pages: + url, depth = to_visit.pop(0) + if depth > max_depth: + continue + logger.info(f"Обработка {url}, глубина {depth}") + title, text = await fetch_any_url(url) + if text: + results.append((url, title or url, text)) + else: + continue + if depth < max_depth: + # Извлекаем ссылки из текущей страницы + links = await extract_links_from_url(url, domain_restrict) + for link in links: + if link not in visited: + visited.add(link) + to_visit.append((link, depth + 1)) + return results + + +async def extract_links_from_url(url: str, domain_restrict: bool = True) -> List[str]: + """ + Извлекает все ссылки (href) из HTML-страницы по заданному URL. + + Аргументы: + url (str): адрес страницы. + domain_restrict (bool): если True, оставлять только ссылки на том же домене. + + Возвращает: + List[str]: список абсолютных URL. + """ + try: + async with aiohttp.ClientSession() as session: + async with session.get(url, timeout=10) as resp: + if resp.status != 200: + return [] + html = await resp.text() + soup = BeautifulSoup(html, 'lxml') + base_domain = urlparse(url).netloc + links = set() + for a in soup.find_all('a', href=True): + href = a['href'] + full_url = urljoin(url, href) + parsed = urlparse(full_url) + if domain_restrict and parsed.netloc != base_domain: + continue + if parsed.scheme in ('http', 'https'): + # Удаляем фрагмент (#) + full_url = parsed._replace(fragment='').geturl() + links.add(full_url) + return list(links) + except Exception as e: + logger.error(f"Ошибка извлечения ссылок с {url}: {e}") + return [] \ No newline at end of file