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
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
|
from __future__ import annotations
import sys
import asyncio
import webbrowser
from os import path
from asyncio import AbstractEventLoop
from platformdirs import user_config_dir
from browser_cookie3 import (
chrome,
chromium,
opera,
opera_gx,
brave,
edge,
vivaldi,
firefox,
BrowserCookieError
)
try:
from selenium.webdriver.remote.webdriver import WebDriver
except ImportError:
class WebDriver():
pass
try:
from undetected_chromedriver import Chrome, ChromeOptions
except ImportError:
class Chrome():
def __init__():
raise RuntimeError('Please install the "undetected_chromedriver" package')
class ChromeOptions():
def add_argument():
pass
try:
from pyvirtualdisplay import Display
except ImportError:
class Display():
def start():
pass
def stop():
pass
from ..typing import Dict, Messages, Union, Tuple
from .. import debug
# Change event loop policy on windows
if sys.platform == 'win32':
if isinstance(
asyncio.get_event_loop_policy(), asyncio.WindowsProactorEventLoopPolicy
):
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
# Local Cookie Storage
_cookies: Dict[str, Dict[str, str]] = {}
# If event loop is already running, handle nested event loops
# If "nest_asyncio" is installed, patch the event loop.
def get_event_loop() -> AbstractEventLoop:
try:
asyncio.get_running_loop()
except RuntimeError:
try:
return asyncio.get_event_loop()
except RuntimeError:
asyncio.set_event_loop(asyncio.new_event_loop())
return asyncio.get_event_loop()
try:
event_loop = asyncio.get_event_loop()
if not hasattr(event_loop.__class__, "_nest_patched"):
import nest_asyncio
nest_asyncio.apply(event_loop)
return event_loop
except ImportError:
raise RuntimeError(
'Use "create_async" instead of "create" function in a running event loop. Or install the "nest_asyncio" package.'
)
def init_cookies():
urls = [
'https://chat-gpt.org',
'https://www.aitianhu.com',
'https://chatgptfree.ai',
'https://gptchatly.com',
'https://bard.google.com',
'https://huggingface.co/chat',
'https://open-assistant.io/chat'
]
browsers = ['google-chrome', 'chrome', 'firefox', 'safari']
def open_urls_in_browser(browser):
b = webbrowser.get(browser)
for url in urls:
b.open(url, new=0, autoraise=True)
for browser in browsers:
try:
open_urls_in_browser(browser)
break
except webbrowser.Error:
continue
# Load cookies for a domain from all supported browsers.
# Cache the results in the "_cookies" variable.
def get_cookies(domain_name=''):
if domain_name in _cookies:
return _cookies[domain_name]
def g4f(domain_name):
user_data_dir = user_config_dir("g4f")
cookie_file = path.join(user_data_dir, "Default", "Cookies")
return [] if not path.exists(cookie_file) else chrome(cookie_file, domain_name)
cookies = {}
for cookie_fn in [g4f, chrome, chromium, opera, opera_gx, brave, edge, vivaldi, firefox]:
try:
cookie_jar = cookie_fn(domain_name=domain_name)
if len(cookie_jar) and debug.logging:
print(f"Read cookies from {cookie_fn.__name__} for {domain_name}")
for cookie in cookie_jar:
if cookie.name not in cookies:
cookies[cookie.name] = cookie.value
except BrowserCookieError as e:
pass
_cookies[domain_name] = cookies
return _cookies[domain_name]
def format_prompt(messages: Messages, add_special_tokens=False) -> str:
if not add_special_tokens and len(messages) <= 1:
return messages[0]["content"]
formatted = "\n".join(
[
f'{message["role"].capitalize()}: {message["content"]}'
for message in messages
]
)
return f"{formatted}\nAssistant:"
def get_browser(
user_data_dir: str = None,
hidden_display: bool = False,
proxy: str = None,
options: ChromeOptions = None
) -> Union[Chrome, Tuple[Chrome, Display]] :
if user_data_dir == None:
user_data_dir = user_config_dir("g4f")
if hidden_display:
display = Display(visible=0, size=(1920, 1080))
display.start()
if proxy:
if not options:
options = ChromeOptions()
options.add_argument(f'--proxy-server={proxy}')
browser = Chrome(user_data_dir=user_data_dir, options=options)
if hidden_display:
return browser, display
return browser
|
