mirror of https://github.com/xtekky/gpt4free
You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
108 lines
3.9 KiB
Python
108 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
try:
|
|
from curl_cffi.requests import Session, Response
|
|
from .curl_cffi import StreamResponse, StreamSession, FormData
|
|
has_curl_cffi = True
|
|
except ImportError:
|
|
from typing import Type as Session, Type as Response
|
|
from .aiohttp import StreamResponse, StreamSession, FormData
|
|
has_curl_cffi = False
|
|
try:
|
|
import webview
|
|
import asyncio
|
|
has_webview = True
|
|
except ImportError:
|
|
has_webview = False
|
|
|
|
from .raise_for_status import raise_for_status
|
|
from ..webdriver import WebDriver, WebDriverSession
|
|
from ..webdriver import bypass_cloudflare, get_driver_cookies
|
|
from ..errors import MissingRequirementsError
|
|
from .defaults import DEFAULT_HEADERS, WEBVIEW_HAEDERS
|
|
|
|
async def get_args_from_webview(url: str) -> dict:
|
|
if not has_webview:
|
|
raise MissingRequirementsError('Install "webview" package')
|
|
window = webview.create_window("", url, hidden=True)
|
|
await asyncio.sleep(2)
|
|
body = None
|
|
while body is None:
|
|
try:
|
|
await asyncio.sleep(1)
|
|
body = window.dom.get_element("body:not(.no-js)")
|
|
except:
|
|
...
|
|
headers = {
|
|
**WEBVIEW_HAEDERS,
|
|
"User-Agent": window.evaluate_js("this.navigator.userAgent"),
|
|
"Accept-Language": window.evaluate_js("this.navigator.language"),
|
|
"Referer": window.real_url
|
|
}
|
|
cookies = [list(*cookie.items()) for cookie in window.get_cookies()]
|
|
cookies = {name: cookie.value for name, cookie in cookies}
|
|
window.destroy()
|
|
return {"headers": headers, "cookies": cookies}
|
|
|
|
def get_args_from_browser(
|
|
url: str,
|
|
webdriver: WebDriver = None,
|
|
proxy: str = None,
|
|
timeout: int = 120,
|
|
do_bypass_cloudflare: bool = True,
|
|
virtual_display: bool = False
|
|
) -> dict:
|
|
"""
|
|
Create a Session object using a WebDriver to handle cookies and headers.
|
|
|
|
Args:
|
|
url (str): The URL to navigate to using the WebDriver.
|
|
webdriver (WebDriver, optional): The WebDriver instance to use.
|
|
proxy (str, optional): Proxy server to use for the Session.
|
|
timeout (int, optional): Timeout in seconds for the WebDriver.
|
|
|
|
Returns:
|
|
Session: A Session object configured with cookies and headers from the WebDriver.
|
|
"""
|
|
with WebDriverSession(webdriver, "", proxy=proxy, virtual_display=virtual_display) as driver:
|
|
if do_bypass_cloudflare:
|
|
bypass_cloudflare(driver, url, timeout)
|
|
headers = {
|
|
**DEFAULT_HEADERS,
|
|
'referer': url,
|
|
}
|
|
if not hasattr(driver, "requests"):
|
|
headers["user-agent"] = driver.execute_script("return navigator.userAgent")
|
|
else:
|
|
for request in driver.requests:
|
|
if request.url.startswith(url):
|
|
for key, value in request.headers.items():
|
|
if key in (
|
|
"accept-encoding",
|
|
"accept-language",
|
|
"user-agent",
|
|
"sec-ch-ua",
|
|
"sec-ch-ua-platform",
|
|
"sec-ch-ua-arch",
|
|
"sec-ch-ua-full-version",
|
|
"sec-ch-ua-platform-version",
|
|
"sec-ch-ua-bitness"
|
|
):
|
|
headers[key] = value
|
|
break
|
|
cookies = get_driver_cookies(driver)
|
|
return {
|
|
'cookies': cookies,
|
|
'headers': headers,
|
|
}
|
|
|
|
def get_session_from_browser(url: str, webdriver: WebDriver = None, proxy: str = None, timeout: int = 120) -> Session:
|
|
if not has_curl_cffi:
|
|
raise MissingRequirementsError('Install "curl_cffi" package')
|
|
args = get_args_from_browser(url, webdriver, proxy, timeout)
|
|
return Session(
|
|
**args,
|
|
proxies={"https": proxy, "http": proxy},
|
|
timeout=timeout,
|
|
impersonate="chrome"
|
|
) |