"""使用Playwright加载页面的加载器，然后使用unstructured加载HTML。
"""
import logging
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING, AsyncIterator, Dict, Iterator, List, Optional
from langchain_core.documents import Document
from langchain_community.document_loaders.base import BaseLoader
if TYPE_CHECKING:
from playwright.async_api import Browser as AsyncBrowser
from playwright.async_api import Page as AsyncPage
from playwright.async_api import Response as AsyncResponse
from playwright.sync_api import Browser, Page, Response
logger = logging.getLogger(__name__)
[docs]class PlaywrightEvaluator(ABC):
"""抽象基类，用于所有评估器。
每个评估器应该接收一个页面、一个浏览器实例和一个响应对象，根据需要处理页面，并返回结果文本。"""
[docs] @abstractmethod
def evaluate(self, page: "Page", browser: "Browser", response: "Response") -> str:
"""同步处理页面并返回结果文本。
参数：
page: 需要处理的页面。
browser: 浏览器实例。
response: 来自page.goto()的响应。
返回值：
text: 页面的文本内容。
"""
pass
[docs] @abstractmethod
async def evaluate_async(
self, page: "AsyncPage", browser: "AsyncBrowser", response: "AsyncResponse"
) -> str:
"""异步处理页面并返回结果文本。
参数：
page：要处理的页面。
browser：浏览器实例。
response：来自page.goto()的响应。
返回：
text：页面的文本内容。
"""
pass
[docs]class UnstructuredHtmlEvaluator(PlaywrightEvaluator):
"""使用`unstructured`库评估页面的HTML内容。"""
[docs] def __init__(self, remove_selectors: Optional[List[str]] = None):
"""初始化UnstructuredHtmlEvaluator。"""
try:
import unstructured # noqa:F401
except ImportError:
raise ImportError(
"unstructured package not found, please install it with "
"`pip install unstructured`"
)
self.remove_selectors = remove_selectors
[docs] def evaluate(self, page: "Page", browser: "Browser", response: "Response") -> str:
"""同步处理页面的HTML内容。"""
from unstructured.partition.html import partition_html
for selector in self.remove_selectors or []:
elements = page.locator(selector).all()
for element in elements:
if element.is_visible():
element.evaluate("element => element.remove()")
page_source = page.content()
elements = partition_html(text=page_source)
return "\n\n".join([str(el) for el in elements])
[docs] async def evaluate_async(
self, page: "AsyncPage", browser: "AsyncBrowser", response: "AsyncResponse"
) -> str:
"""异步处理页面的HTML内容。"""
from unstructured.partition.html import partition_html
for selector in self.remove_selectors or []:
elements = await page.locator(selector).all()
for element in elements:
if await element.is_visible():
await element.evaluate("element => element.remove()")
page_source = await page.content()
elements = partition_html(text=page_source)
return "\n\n".join([str(el) for el in elements])
[docs]class PlaywrightURLLoader(BaseLoader):
"""使用`Playwright`加载`HTML`页面，并使用`Unstructured`进行解析。
这对于加载需要JavaScript渲染的页面非常有用。
属性:
urls (List[str]): 要加载的URL列表。
continue_on_failure (bool): 如果为True，则在失败时继续加载其他URL。
headless (bool): 如果为True，则浏览器将以无头模式运行。
proxy (Optional[Dict[str, str]]): 如果设置，则浏览器将通过指定的代理访问URL。
示例:
.. code-block:: python
from langchain_community.document_loaders import PlaywrightURLLoader
urls = ["https://api.ipify.org/?format=json",]
proxy={
"server": "https://xx.xx.xx:15818", # https://<host>:<port>
"username": "username",
"password": "password"
}
loader = PlaywrightURLLoader(urls, proxy=proxy)
data = loader.load()"""
[docs] def __init__(
self,
urls: List[str],
continue_on_failure: bool = True,
headless: bool = True,
remove_selectors: Optional[List[str]] = None,
evaluator: Optional[PlaywrightEvaluator] = None,
proxy: Optional[Dict[str, str]] = None,
):
"""使用Playwright加载URL列表。"""
try:
import playwright # noqa:F401
except ImportError:
raise ImportError(
"playwright package not found, please install it with "
"`pip install playwright`"
)
self.urls = urls
self.continue_on_failure = continue_on_failure
self.headless = headless
self.proxy = proxy
if remove_selectors and evaluator:
raise ValueError(
"`remove_selectors` and `evaluator` cannot be both not None"
)
# Use the provided evaluator, if any, otherwise, use the default.
self.evaluator = evaluator or UnstructuredHtmlEvaluator(remove_selectors)
[docs] def lazy_load(self) -> Iterator[Document]:
"""使用Playwright加载指定的URL并创建Document实例。
返回：
一个包含已加载内容的Document实例列表。
"""
from playwright.sync_api import sync_playwright
with sync_playwright() as p:
browser = p.chromium.launch(headless=self.headless, proxy=self.proxy)
for url in self.urls:
try:
page = browser.new_page()
response = page.goto(url)
if response is None:
raise ValueError(f"page.goto() returned None for url {url}")
text = self.evaluator.evaluate(page, browser, response)
metadata = {"source": url}
yield Document(page_content=text, metadata=metadata)
except Exception as e:
if self.continue_on_failure:
logger.error(
f"Error fetching or processing {url}, exception: {e}"
)
else:
raise e
browser.close()
[docs] async def aload(self) -> List[Document]:
"""使用Playwright异步加载指定的URL，并创建文档。
在jupyter notebook环境中使用此函数。
返回：
一个包含已加载内容的Document实例列表。
"""
return [doc async for doc in self.alazy_load()]
[docs] async def alazy_load(self) -> AsyncIterator[Document]:
"""使用Playwright异步加载指定的URL，并创建文档。
在jupyter notebook环境中使用此函数。
返回：
一个包含已加载内容的Document实例列表。
"""
from playwright.async_api import async_playwright
async with async_playwright() as p:
browser = await p.chromium.launch(headless=self.headless, proxy=self.proxy)
for url in self.urls:
try:
page = await browser.new_page()
response = await page.goto(url)
if response is None:
raise ValueError(f"page.goto() returned None for url {url}")
text = await self.evaluator.evaluate_async(page, browser, response)
metadata = {"source": url}
yield Document(page_content=text, metadata=metadata)
except Exception as e:
if self.continue_on_failure:
logger.error(
f"Error fetching or processing {url}, exception: {e}"
)
else:
raise e
await browser.close()