File size: 2,788 Bytes
d2f3514 |
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 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 |
from __future__ import annotations
import json
import random
import re
from aiohttp import ClientSession
from .base_provider import AsyncProvider, format_prompt, get_cookies
class Bard(AsyncProvider):
url = "https://bard.google.com"
needs_auth = True
working = True
_snlm0e = None
@classmethod
async def create_async(
cls,
model: str,
messages: list[dict[str, str]],
proxy: str = None,
cookies: dict = None,
**kwargs
) -> str:
prompt = format_prompt(messages)
if proxy and "://" not in proxy:
proxy = f"http://{proxy}"
if not cookies:
cookies = get_cookies(".google.com")
headers = {
'authority': 'bard.google.com',
'origin': 'https://bard.google.com',
'referer': 'https://bard.google.com/',
'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/111.0.0.0 Safari/537.36',
'x-same-domain': '1',
}
async with ClientSession(
cookies=cookies,
headers=headers
) as session:
if not cls._snlm0e:
async with session.get(cls.url, proxy=proxy) as response:
text = await response.text()
match = re.search(r'SNlM0e\":\"(.*?)\"', text)
if not match:
raise RuntimeError("No snlm0e value.")
cls._snlm0e = match.group(1)
params = {
'bl': 'boq_assistant-bard-web-server_20230326.21_p0',
'_reqid': random.randint(1111, 9999),
'rt': 'c'
}
data = {
'at': cls._snlm0e,
'f.req': json.dumps([None, json.dumps([[prompt]])])
}
intents = '.'.join([
'assistant',
'lamda',
'BardFrontendService'
])
async with session.post(
f'{cls.url}/_/BardChatUi/data/{intents}/StreamGenerate',
data=data,
params=params,
proxy=proxy
) as response:
response = await response.text()
response = json.loads(response.splitlines()[3])[0][2]
response = json.loads(response)[4][0][1][0]
return response
@classmethod
@property
def params(cls):
params = [
("model", "str"),
("messages", "list[dict[str, str]]"),
("stream", "bool"),
("proxy", "str"),
]
param = ", ".join([": ".join(p) for p in params])
return f"g4f.provider.{cls.__name__} supports: ({param})"
|