File size: 1,926 Bytes
a4b70d9 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 |
from __future__ import annotations
import asyncio
import json
from ...typing import AsyncResult, Messages, Cookies
from ..base_provider import AsyncGeneratorProvider, ProviderModelMixin, AuthFileMixin, get_running_loop
from ...requests import Browser, get_nodriver, has_nodriver
from ...errors import MissingRequirementsError, ModelNotFoundError
from ... import debug
from ..helper import get_last_user_message
class GoogleSearch(AsyncGeneratorProvider, AuthFileMixin):
label = "Google Search"
url = "https://google.com"
working = has_nodriver
use_nodriver = True
@classmethod
async def create_async_generator(
cls,
model: str,
messages: Messages,
browser: Browser = None,
proxy: str = None,
timeout: int = 300,
**kwargs
) -> AsyncResult:
if not has_nodriver:
raise MissingRequirementsError("Google requires a browser to be installed.")
if not cls.working:
raise ModelNotFoundError(f"Model {model} not found.")
try:
stop_browser = None
if browser is None:
browser, stop_browser = await get_nodriver(proxy=proxy, timeout=timeout)
tab = await browser.get(cls.url)
await asyncio.sleep(3)
while True:
try:
await tab.wait_for('[aria-modal="true"]', timeout=10)
await tab.wait_for('[aria-modal="true"][style*="display: none"]', timeout=timeout)
except Exception as e:
break
break
element = await tab.wait_for('textarea')
await element.send_keys(get_last_user_message(messages))
button = await tab.find("Google Suche")
await button.click()
await asyncio.sleep(1000)
finally:
if stop_browser is not None:
stop_browser() |