# -*- coding: utf-8 -*- """ Утилиты для загрузки веб-страниц и PDF, извлечения основного текста, рекурсивного обхода ссылок. Добавлено ограничение размера загружаемого контента. """ 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: целевой адрес. timeout: таймаут запроса в секундах. max_size_mb: максимальный размер загружаемого контента в МБ. Возвращает: Tuple[Optional[str], Optional[str]]: (заголовок, текст). Для HTML: заголовок страницы и очищенный основной текст. Для PDF: имя файла (базовая часть URL) и извлечённый текст. В случае ошибки возвращает (None, None). """ 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 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'): 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 (без изменений).""" soup = BeautifulSoup(html, 'lxml') 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: 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 if HAS_READABILITY: doc = Document(html) text = doc.summary() if not title and doc.title(): title = doc.title() 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 []