Add new file
This commit is contained in:
Markov Andrey
242
core/utils/web_utils.py
Normal file
242
core/utils/web_utils.py
Normal file
@@ -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. <title>
|
||||
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 []
|
||||
Reference in New Issue
Block a user