210 lines
8.6 KiB
Python
210 lines
8.6 KiB
Python
# -*- 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 [] |