# -*- coding: utf-8 -*- """ Утилиты для загрузки веб-страниц и PDF, извлечения основного текста, рекурсивного обхода ссылок. Использует aiohttp для асинхронных запросов, BeautifulSoup для парсинга, readability-lxml для извлечения контента (опционально). Добавлено ограничение размера загружаемого контента (max_size_mb). """ 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, max_size_mb: int = 5) -> Tuple[Optional[str], Optional[str]]: """ Универсальная загрузка контента по URL. Аргументы: url (str): целевой адрес. timeout (int): таймаут запроса в секундах. max_size_mb (int): максимальный размер загружаемого контента в МБ. Возвращает: Tuple[Optional[str], Optional[str]]: (заголовок, текст). Для HTML: заголовок страницы и очищенный основной текст. Для PDF: имя файла (базовая часть URL) и извлечённый текст. В случае ошибки возвращает (None, None). Примечания: - Для PDF используется временный файл, который удаляется после чтения. - Поддерживаются только HTTP/HTTPS. - Таймаут применяется к запросу в целом (чтение + соединение). - Размер контента ограничен параметром max_size_mb. """ 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() limit = max_size_mb * 1024 * 1024 # Ограничение размера: читаем только limit байт try: data = await resp.content.read(limit=limit) except aiohttp.ContentTooLong: logger.error(f"Содержимое {url} превышает лимит {max_size_mb} МБ") return None, None # Обработка 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(data) 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 = data.decode('utf-8', errors='ignore') 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, max_size_mb: int = 5, ) -> List[Tuple[str, str, str]]: """ Рекурсивно обходит ссылки на том же домене (или не ограничивая, если domain_restrict=False). Аргументы: start_url (str): начальная страница. max_depth (int): максимальная глубина перехода по ссылкам. max_pages (int): максимальное количество страниц для обработки. domain_restrict (bool): ограничиваться ли тем же доменом (по умолчанию True). max_size_mb (int): максимальный размер загружаемой страницы (МБ). Возвращает: 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, max_size_mb=max_size_mb) 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 []