blackteay/DrissionPage
1
0
Fork 0
mirror of https://gitee.com/g1879/DrissionPage.git synced 2024-12-10 04:00:23 +08:00
Code Issues Projects Releases Wiki Activity

增加ChromiumOptions，修改项目结构，未完成

Browse Source
This commit is contained in:
g1879 2023-01-17 21:48:47 +08:00
parent 3692aba38f
commit 3c1e3b9690
20 changed files with 1735 additions and 1213 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
DrissionPage/__init__.py
View File

@ -12,5 +12,7 @@ from .web_page import WebPage
from .chromium_page import ChromiumPage
from .session_page import SessionPage
from .drission import Drission
from .config import DriverOptions, SessionOptions
from .configs.driver_options import DriverOptions
from .configs.chromium_options import ChromiumOptions
from .configs.session_options import SessionOptions
from .action_chains import ActionChains

3
DrissionPage/chromium_base.py
View File

@ -10,8 +10,7 @@ from requests import Session
from .base import BasePage
from .chromium_element import ChromiumElementWaiter, ChromiumScroll, ChromiumElement, run_js, make_chromium_ele
from .common import get_loc, offset_scroll
from .config import cookies_to_tuple
from .common import get_loc, offset_scroll, cookies_to_tuple
from .session_element import make_session_ele
from .chromium_driver import ChromiumDriver

2
DrissionPage/chromium_driver.py
View File

@ -174,7 +174,7 @@ class ChromiumDriver(object):
timeout = kwargs.pop("_timeout", None)
result = self._send({"method": _method, "params": kwargs}, timeout=timeout)
if 'result' not in result and 'error' in result:
raise CallMethodException(f"调用方法：{_method} 错误：{result['error']['message']}")
raise CallMethodException(f"\n调用方法：{_method}\n参数：{kwargs}\n错误：{result['error']['message']}")
return result['result']

5
DrissionPage/chromium_page.py
View File

@ -14,7 +14,7 @@ from .chromium_base import Timeout, ChromiumBase
from .chromium_driver import ChromiumDriver
from .chromium_tab import ChromiumTab
from .common import connect_browser
from .config import DriverOptions
from .configs.driver_options import DriverOptions
from .session_page import DownloadSetter
@ -418,6 +418,7 @@ class WindowSetter(object):
def normal(self):
"""设置窗口为常规模式"""
self._perform({'windowState': 'normal'})
self._perform({'windowState': 'normal'})
def size(self, width=None, height=None):
"""设置窗口大小 \n
@ -446,7 +447,7 @@ class WindowSetter(object):
def _get_info(self):
"""获取窗口位置及大小信息"""
return self.driver.Browser.getWindowBounds()
return self.driver.Browser.getWindowForTarget()
def _perform(self, bounds):
"""执行改变窗口大小操作

194
DrissionPage/common.py
View File

@ -4,17 +4,21 @@
@Contact : g1879@qq.com
"""
from html import unescape
from http.cookiejar import Cookie
from pathlib import Path
from platform import system
from re import split, search, sub
from shutil import rmtree
from subprocess import Popen
from time import perf_counter, sleep
from zipfile import ZipFile
from urllib.parse import urlparse, urljoin, urlunparse
from requests import get as requests_get
from zipfile import ZipFile
from .config import DriverOptions
from requests import get as requests_get
from requests.cookies import RequestsCookieJar
# from .configs.chromium_options import ChromiumOptions
from .configs.driver_options import DriverOptions
def get_ele_txt(e):
@ -329,6 +333,61 @@ def format_html(text):
return unescape(text).replace('\xa0', ' ') if text else text
def cookie_to_dict(cookie):
"""把Cookie对象转为dict格式 \n
:param cookie: Cookie对象
:return: cookie字典
"""
if isinstance(cookie, Cookie):
cookie_dict = cookie.__dict__.copy()
cookie_dict.pop('rfc2109')
cookie_dict.pop('_rest')
return cookie_dict
elif isinstance(cookie, dict):
cookie_dict = cookie
elif isinstance(cookie, str):
cookie = cookie.split(',' if ',' in cookie else ';')
cookie_dict = {}
for key, attr in enumerate(cookie):
attr_val = attr.lstrip().split('=')
if key == 0:
cookie_dict['name'] = attr_val[0]
cookie_dict['value'] = attr_val[1] if len(attr_val) == 2 else ''
else:
cookie_dict[attr_val[0]] = attr_val[1] if len(attr_val) == 2 else ''
return cookie_dict
else:
raise TypeError('cookie参数必须为Cookie、str或dict类型。')
return cookie_dict
def cookies_to_tuple(cookies):
"""把cookies转为tuple格式 \n
:param cookies: cookies信息，可为CookieJar, list, tuple, str, dict
:return: 返回tuple形式的cookies
"""
if isinstance(cookies, (list, tuple, RequestsCookieJar)):
cookies = tuple(cookie_to_dict(cookie) for cookie in cookies)
elif isinstance(cookies, str):
cookies = tuple(cookie_to_dict(cookie.lstrip()) for cookie in cookies.split(";"))
elif isinstance(cookies, dict):
cookies = tuple({'name': cookie, 'value': cookies[cookie]} for cookie in cookies)
else:
raise TypeError('cookies参数必须为RequestsCookieJar、list、tuple、str或dict类型。')
return cookies
def clean_folder(folder_path, ignore=None):
"""清空一个文件夹，除了ignore里的文件和文件夹 \n
:param folder_path: 要清空的文件夹路径
@ -524,8 +583,8 @@ def connect_browser(option):
chrome_path = get_exe_from_port(port) if chrome_path == 'chrome' and system_type == 'windows' else chrome_path
return chrome_path, None
args = _get_launch_args(option)
_set_prefs(option)
args = get_launch_args(option)
set_prefs(option)
# ----------创建浏览器进程----------
try:
@ -546,6 +605,71 @@ def connect_browser(option):
return chrome_path, debugger
def get_launch_args(opt):
"""从DriverOptions获取命令行启动参数"""
sys = system().lower()
result = []
# ----------处理arguments-----------
args = opt.arguments
for arg in args:
index = arg.find('=') + 1
if index == 0:
result.append(arg)
else:
a = arg[index:].strip()
if a.startswith('"') and a.endswith('"'):
result.append(arg)
else:
result.append(f'{arg[:index]}"{a}"')
# ----------处理插件extensions-------------
ext = opt._extension_files if isinstance(opt, DriverOptions) else opt.extensions
if ext:
ext = set(ext)
if sys == 'windows':
ext = '","'.join(ext)
ext = f'"{ext}"'
else:
ext = ','.join(ext)
ext = f'--load-extension={ext}'
result.append(ext)
return result
def set_prefs(opt):
"""处理启动配置中的prefs项，目前只能对已存在文件夹配置"""
# todo: 支持删除pref项
prefs = opt.experimental_options.get('prefs', None) if isinstance(opt, DriverOptions) else opt.preferences
if prefs and opt.user_data_path:
args = opt.arguments
user = 'Default'
for arg in args:
if arg.startswith('--profile-directory'):
user = arg.split('=')[-1].strip()
break
prefs_file = Path(opt.user_data_path) / user / 'Preferences'
if not prefs_file.exists():
prefs_file.parent.mkdir(parents=True, exist_ok=True)
with open(prefs_file, 'w') as f:
f.write('{}')
from json import load, dump
with open(prefs_file, "r", encoding='utf-8') as f:
j = load(f)
for pref in prefs:
value = prefs[pref]
pref = pref.split('.')
_make_leave_in_dict(j, pref, 0, len(pref))
_set_value_to_dict(j, pref, value)
with open(prefs_file, 'w', encoding='utf-8') as f:
dump(j, f)
def _run_browser(port, path: str, args) -> Popen:
"""创建chrome进程 \n
:param port: 端口号
@ -574,66 +698,6 @@ def _run_browser(port, path: str, args) -> Popen:
raise ConnectionError('无法连接浏览器。')
def _get_launch_args(opt: DriverOptions) -> list:
"""从DriverOptions获取命令行启动参数"""
sys = system().lower()
result = []
# ----------处理arguments-----------
args = opt.arguments
for arg in args:
if arg.startswith(('--user-data-dir', '--disk-cache-dir', '--user-agent')) and sys == 'windows':
index = arg.find('=') + 1
result.append(f'{arg[:index]}"{arg[index:].strip()}"')
else:
result.append(arg)
# ----------处理插件extensions-------------
ext = opt._extension_files
if ext:
ext = set(ext)
if sys == 'windows':
ext = '","'.join(ext)
ext = f'"{ext}"'
else:
ext = ','.join(ext)
ext = f'--load-extension={ext}'
result.append(ext)
return result
def _set_prefs(opt: DriverOptions) -> None:
"""处理启动配置中的prefs项，目前只能对已存在文件夹配置"""
prefs = opt.experimental_options.get('prefs', None)
if prefs and opt.user_data_path:
args = opt.arguments
profile = 'Default'
for arg in args:
if arg.startswith('--profile-directory'):
profile = arg.split('=')[-1].strip()
break
prefs_file = Path(opt.user_data_path) / profile / 'Preferences'
if not prefs_file.exists():
prefs_file.parent.mkdir(parents=True, exist_ok=True)
with open(prefs_file, 'w') as f:
f.write('{}')
from json import load, dump
with open(prefs_file, "r", encoding='utf-8') as f:
j = load(f)
for pref in prefs:
value = prefs[pref]
pref = pref.split('.')
_make_leave_in_dict(j, pref, 0, len(pref))
_set_value_to_dict(j, pref, value)
with open(prefs_file, 'w', encoding='utf-8') as f:
dump(j, f)
def _make_leave_in_dict(target_dict: dict, src: list, num: int, end: int) -> None:
"""把prefs中a.b.c形式的属性转为a['b']['c']形式
:param target_dict: 要处理的dict

21
DrissionPage/common.pyi
View File

@ -3,13 +3,16 @@
@Author : g1879
@Contact : g1879@qq.com
"""
from http.cookiejar import Cookie
from pathlib import Path
from typing import Union
from requests import get as requests_get
from requests.cookies import RequestsCookieJar
from .base import BasePage, DrissionElement
from .chromium_element import ChromiumElement
from .config import DriverOptions
from .configs.chromium_options import ChromiumOptions
def get_ele_txt(e: DrissionElement) -> str: ...
@ -27,6 +30,12 @@ def translate_loc(loc: tuple) -> tuple: ...
def format_html(text: str) -> str: ...
def cookie_to_dict(cookie: Union[Cookie, str, dict]) -> dict: ...
def cookies_to_tuple(cookies: Union[RequestsCookieJar, list, tuple, str, dict]) -> tuple: ...
def clean_folder(folder_path: str, ignore: list = None) -> None: ...
@ -54,10 +63,16 @@ def make_absolute_link(link, page: BasePage = None) -> str: ...
def is_js_func(func: str) -> bool: ...
def port_is_using(ip: str, port: str) -> bool: ...
def port_is_using(ip: str, port: Union[str, int]) -> bool: ...
def connect_browser(option: DriverOptions) -> tuple: ...
def connect_browser(option: Union[ChromiumOptions, DriverOptions]) -> tuple: ...
def get_launch_args(opt: Union[ChromiumOptions, DriverOptions]) -> list: ...
def set_prefs(opt: Union[ChromiumOptions, DriverOptions]) -> None: ...
def location_in_viewport(page, loc_x: int, loc_y: int) -> bool: ...

905
DrissionPage/config.py
View File

@ -1,905 +0,0 @@
# -*- coding:utf-8 -*-
"""
@Author : g1879
@Contact : g1879@qq.com
"""
from configparser import RawConfigParser, NoSectionError, NoOptionError
from http.cookiejar import Cookie
from pathlib import Path
from requests.cookies import RequestsCookieJar
from selenium.webdriver.chrome.options import Options
class OptionsManager(object):
"""管理配置文件内容的类"""
def __init__(self, path=None):
"""初始化，读取配置文件，如没有设置临时文件夹，则设置并新建 \n
:param path: ini文件的路径，默认读取模块文件夹下的
"""
self.ini_path = str(Path(__file__).parent / 'configs.ini') if path == 'default' or path is None else path
if not Path(self.ini_path).exists():
raise FileNotFoundError('ini文件不存在。')
self._conf = RawConfigParser()
self._conf.read(self.ini_path, encoding='utf-8')
def __getattr__(self, item):
"""以dict形似返回获取大项信息
:param item: 项名
:return: None
"""
return self.get_option(item)
def get_value(self, section, item):
"""获取配置的值 \n
:param section: 段名
:param item: 项名
:return: 项值
"""
try:
return eval(self._conf.get(section, item))
except (SyntaxError, NameError):
return self._conf.get(section, item)
except NoSectionError and NoOptionError:
return None
def get_option(self, section):
"""把section内容以字典方式返回 \n
:param section: 段名
:return: 段内容生成的字典
"""
items = self._conf.items(section)
option = dict()
for j in items:
try:
option[j[0]] = eval(self._conf.get(section, j[0]))
except Exception:
option[j[0]] = self._conf.get(section, j[0])
return option
def set_item(self, section, item, value):
"""设置配置值 \n
:param section: 段名
:param item: 项名
:param value: 项值
:return: None
"""
self._conf.set(section, item, str(value))
self.__setattr__(f'_{section}', None)
return self
def remove_item(self, section, item):
"""删除配置值 \n
:param section: 段名
:param item: 项名
:return: None
"""
self._conf.remove_option(section, item)
return self
def save(self, path=None):
"""保存配置文件 \n
:param path: ini文件的路径，传入 'default' 保存到默认ini文件
:return: 保存路径
"""
default_path = (Path(__file__).parent / 'configs.ini').absolute()
if path == 'default':
path = default_path
elif path is None:
path = Path(self.ini_path).absolute()
else:
path = Path(path).absolute()
path = path / 'config.ini' if path.is_dir() else path
path = str(path)
self._conf.write(open(path, 'w', encoding='utf-8'))
print(f'配置已保存到文件：{path}')
if path == str(default_path):
print('以后程序可自动从文件加载配置。')
return path
def save_to_default(self):
"""保存当前配置到默认ini文件"""
return self.save('default')
class SessionOptions(object):
"""requests的Session对象配置类"""
def __init__(self, read_file=True, ini_path=None):
"""
:param read_file: 是否从文件读取配置
:param ini_path: ini文件路径
"""
self.ini_path = None
self._download_path = None
self._headers = None
self._cookies = None
self._auth = None
self._proxies = None
self._hooks = None
self._params = None
self._verify = None
self._cert = None
self._adapters = None
self._stream = None
self._trust_env = None
self._max_redirects = None
self._timeout = 10
self._del_set = set() # 记录要从ini文件删除的参数
if read_file:
self.ini_path = ini_path or str(Path(__file__).parent / 'configs.ini')
om = OptionsManager(self.ini_path)
options_dict = om.session_options
if options_dict.get('headers', None) is not None:
self.set_headers(options_dict['headers'])
if options_dict.get('cookies', None) is not None:
self.set_cookies(options_dict['cookies'])
if options_dict.get('auth', None) is not None:
self._auth = options_dict['auth']
if options_dict.get('params', None) is not None:
self._params = options_dict['params']
if options_dict.get('verify', None) is not None:
self._verify = options_dict['verify']
if options_dict.get('cert', None) is not None:
self._cert = options_dict['cert']
if options_dict.get('stream', None) is not None:
self._stream = options_dict['stream']
if options_dict.get('trust_env', None) is not None:
self._trust_env = options_dict['trust_env']
if options_dict.get('max_redirects', None) is not None:
self._max_redirects = options_dict['max_redirects']
self.set_proxies(om.proxies.get('http', None), om.proxies.get('https', None))
self._timeout = om.timeouts.get('implicit', 10)
self._download_path = om.paths.get('download_path', None)
# ===========须独立处理的项开始============
@property
def download_path(self):
"""返回默认下载路径属性信息"""
return self._download_path
def set_paths(self, download_path=None):
"""设置默认下载路径 \n
:param download_path: 下载路径
:return: 返回当前对象
"""
if download_path is not None:
self._download_path = str(download_path)
return self
@property
def timeout(self):
"""返回timeout属性信息"""
return self._timeout
def set_timeout(self, second):
"""设置超时信息
:param second: 秒数
:return: 返回当前对象
"""
self._timeout = second
return self
@property
def proxies(self):
"""返回proxies设置信息"""
if self._proxies is None:
self._proxies = {}
return self._proxies
def set_proxies(self, http, https=None):
"""设置proxies参数 \n
:param http: http代理地址
:param https: https代理地址
:return: 返回当前对象
"""
proxies = None if http == https is None else {'http': http, 'https': https or http}
self._sets('proxies', proxies)
return self
# ===========须独立处理的项结束============
@property
def headers(self):
"""返回headers设置信息"""
if self._headers is None:
self._headers = {}
return self._headers
def set_headers(self, headers):
"""设置headers参数 \n
:param headers: 参数值，传入None可在ini文件标记删除
:return: 返回当前对象
"""
if headers is None:
self._headers = None
self._del_set.add('headers')
else:
self._headers = {key.lower(): headers[key] for key in headers}
return self
def set_a_header(self, attr, value):
"""设置headers中一个项 \n
:param attr: 设置名称
:param value: 设置值
:return: 返回当前对象
"""
if self._headers is None:
self._headers = {}
self._headers[attr.lower()] = value
return self
def remove_a_header(self, attr):
"""从headers中删除一个设置 \n
:param attr: 要删除的设置
:return: 返回当前对象
"""
if self._headers is None:
return self
attr = attr.lower()
if attr in self._headers:
self._headers.pop(attr)
return self
@property
def cookies(self):
"""以list形式返回cookies"""
if self._cookies is None:
self._cookies = []
return self._cookies
def set_cookies(self, cookies):
"""设置cookies信息 \n
:param cookies: cookies，可为CookieJar, list, tuple, str, dict，传入None可在ini文件标记删除
:return: 返回当前对象
"""
cookies = cookies if cookies is None else list(cookies_to_tuple(cookies))
self._sets('cookies', cookies)
return self
@property
def auth(self):
"""返回auth设置信息"""
return self._auth
def set_auth(self, auth):
"""设置认证元组或对象 \n
:param auth: 认证元组或对象
:return: 返回当前对象
"""
self._sets('auth', auth)
return self
@property
def hooks(self):
"""返回回调方法"""
if self._hooks is None:
self._hooks = {}
return self._hooks
def set_hooks(self, hooks):
"""设置回调方法 \n
:param hooks:
:return: 返回当前对象
"""
self._hooks = hooks
return self
@property
def params(self):
"""返回params设置信息"""
if self._params is None:
self._params = {}
return self._params
def set_params(self, params):
"""设置查询参数字典 \n
:param params: 查询参数字典
:return: 返回当前对象
"""
self._sets('params', params)
return self
@property
def verify(self):
"""返回是否验证SSL证书设置"""
return self._verify
def set_verify(self, on_off):
"""设置是否验证SSL证书 \n
:param on_off: 是否验证 SSL 证书
:return: 返回当前对象
"""
self._sets('verify', on_off)
return self
@property
def cert(self):
"""返回cert设置信息"""
return self._cert
def set_cert(self, cert):
"""SSL客户端证书文件的路径(.pem格式)，或(‘cert’, ‘key’)元组 \n
:param cert: 证书路径或元组
:return: 返回当前对象
"""
self._sets('cert', cert)
return self
@property
def adapters(self):
"""返回适配器设置信息"""
if self._adapters is None:
self._adapters = []
return self._adapters
def add_adapter(self, url, adapter):
"""添加适配器 \n
:param url: 适配器对应url
:param adapter: 适配器对象
:return: 返回当前对象
"""
self._adapters.append((url, adapter))
return self
@property
def stream(self):
"""返回stream设置信息"""
return self._stream
def set_stream(self, on_off):
"""设置是否使用流式响应内容 \n
:param on_off: 是否使用流式响应内容
:return: 返回当前对象
"""
self._sets('stream', on_off)
return self
@property
def trust_env(self):
"""返回trust_env设置信息"""
return self._trust_env
def set_trust_env(self, on_off):
"""设置是否信任环境 \n
:param on_off: 是否信任环境
:return: 返回当前对象
"""
self._sets('trust_env', on_off)
return self
@property
def max_redirects(self):
"""返回最大重定向次数"""
return self._max_redirects
def set_max_redirects(self, times):
"""设置最大重定向次数 \n
:param times: 最大重定向次数
:return: 返回当前对象
"""
self._sets('max_redirects', times)
return self
def _sets(self, arg, val):
"""给属性赋值或标记删除
:param arg: 属性名称
:param val: 参数值
:return: None
"""
if val is None:
self.__setattr__(f'_{arg}', None)
self._del_set.add(arg)
else:
self.__setattr__(f'_{arg}', val)
if arg in self._del_set:
self._del_set.remove(arg)
def save(self, path=None):
"""保存设置到文件 \n
:param path: ini文件的路径，传入 'default' 保存到默认ini文件
:return: 保存文件的绝对路径
"""
if path == 'default':
path = (Path(__file__).parent / 'configs.ini').absolute()
elif path is None:
if self.ini_path:
path = Path(self.ini_path).absolute()
else:
path = (Path(__file__).parent / 'configs.ini').absolute()
else:
path = Path(path).absolute()
path = path / 'config.ini' if path.is_dir() else path
if path.exists():
om = OptionsManager(str(path))
else:
om = OptionsManager(self.ini_path or str(Path(__file__).parent / 'configs.ini'))
options = session_options_to_dict(self)
for i in options:
if i not in ('download_path', 'timeout', 'proxies'):
om.set_item('session_options', i, options[i])
om.set_item('paths', 'download_path', self.download_path)
om.set_item('timeouts', 'implicit', self.timeout)
om.set_item('proxies', 'http', self.proxies.get('http', None))
om.set_item('proxies', 'https', self.proxies.get('https', None))
for i in self._del_set:
if i == 'download_path':
om.set_item('paths', 'download_path', '')
elif i == 'proxies':
om.set_item('proxies', 'http', '')
om.set_item('proxies', 'https', '')
else:
om.remove_item('session_options', i)
path = str(path)
om.save(path)
return path
def save_to_default(self):
"""保存当前配置到默认ini文件"""
return self.save('default')
def as_dict(self):
"""以字典形式返回本对象"""
return session_options_to_dict(self)
class DriverOptions(Options):
"""chrome浏览器配置类，继承自selenium.webdriver.chrome.options的Options类，
增加了删除配置和保存到文件方法。
"""
def __init__(self, read_file=True, ini_path=None):
"""初始化，默认从文件读取设置 \n
:param read_file: 是否从默认ini文件中读取配置信息
:param ini_path: ini文件路径，为None则读取默认ini文件
"""
super().__init__()
self._user_data_path = None
if read_file:
self.ini_path = ini_path or str(Path(__file__).parent / 'configs.ini')
om = OptionsManager(self.ini_path)
options_dict = om.chrome_options
self._driver_path = om.paths.get('chromedriver_path', None)
self._download_path = om.paths.get('download_path', None)
self._binary_location = options_dict.get('binary_location', '')
self._arguments = options_dict.get('arguments', [])
self._extensions = options_dict.get('extensions', [])
self._experimental_options = options_dict.get('experimental_options', {})
self._debugger_address = options_dict.get('debugger_address', None)
self.page_load_strategy = options_dict.get('page_load_strategy', 'normal')
for arg in self._arguments:
if arg.startswith('--user-data-dir='):
self.set_paths(user_data_path=arg[16:])
break
self.timeouts = options_dict.get('timeouts', {'implicit': 10, 'pageLoad': 30, 'script': 30})
return
self._driver_path = None
self._download_path = None
self.ini_path = None
self.timeouts = {'implicit': 10, 'pageLoad': 30, 'script': 30}
self._debugger_address = '127.0.0.1:9222'
@property
def driver_path(self):
"""chromedriver文件路径"""
return self._driver_path
@property
def download_path(self):
"""默认下载路径文件路径"""
return self._download_path
@property
def chrome_path(self):
"""浏览器启动文件路径"""
return self.browser_path
@property
def browser_path(self):
"""浏览器启动文件路径"""
return self.binary_location or 'chrome'
@property
def user_data_path(self):
"""返回用户文件夹路径"""
return self._user_data_path
# -------------重写父类方法，实现链式操作-------------
def add_argument(self, argument):
"""添加一个配置项 \n
:param argument: 配置项内容
:return: 当前对象
"""
super().add_argument(argument)
return self
def set_capability(self, name, value):
"""设置一个capability \n
:param name: capability名称
:param value: capability值
:return: 当前对象
"""
super().set_capability(name, value)
return self
def add_extension(self, extension):
"""添加插件 \n
:param extension: crx文件路径
:return: 当前对象
"""
super().add_extension(extension)
return self
def add_encoded_extension(self, extension):
"""将带有扩展数据的 Base64 编码字符串添加到将用于将其提取到 ChromeDriver 的列表中 \n
:param extension: 带有扩展数据的 Base64 编码字符串
:return: 当前对象
"""
super().add_encoded_extension(extension)
return self
def add_experimental_option(self, name, value):
"""添加一个实验选项到浏览器 \n
:param name: 选项名称
:param value: 选项值
:return: 当前对象
"""
super().add_experimental_option(name, value)
return self
# -------------重写父类方法结束-------------
def save(self, path=None):
"""保存设置到文件 \n
:param path: ini文件的路径， None 保存到当前读取的配置文件，传入 'default' 保存到默认ini文件
:return: 保存文件的绝对路径
"""
if path == 'default':
path = (Path(__file__).parent / 'configs.ini').absolute()
elif path is None:
if self.ini_path:
path = Path(self.ini_path).absolute()
else:
path = (Path(__file__).parent / 'configs.ini').absolute()
else:
path = Path(path).absolute()
path = path / 'config.ini' if path.is_dir() else path
if path.exists():
om = OptionsManager(str(path))
else:
om = OptionsManager(self.ini_path or str(Path(__file__).parent / 'configs.ini'))
options = self.as_dict()
for i in options:
if i == 'driver_path':
om.set_item('paths', 'chromedriver_path', options[i])
elif i == 'download_path':
om.set_item('paths', 'download_path', options[i])
else:
om.set_item('chrome_options', i, options[i])
path = str(path)
om.save(path)
return path
def save_to_default(self):
"""保存当前配置到默认ini文件"""
return self.save('default')
def remove_argument(self, value):
"""移除一个argument项 \n
:param value: 设置项名，有值的设置项传入设置名称即可
:return: 当前对象
"""
del_list = []
for argument in self._arguments:
if argument.startswith(value):
del_list.append(argument)
for del_arg in del_list:
self._arguments.remove(del_arg)
return self
def remove_experimental_option(self, key):
"""移除一个实验设置，传入key值删除 \n
:param key: 实验设置的名称
:return: 当前对象
"""
if key in self._experimental_options:
self._experimental_options.pop(key)
return self
def remove_all_extensions(self):
"""移除所有插件 \n
:return: 当前对象
"""
# 因插件是以整个文件储存，难以移除其中一个，故如须设置则全部移除再重设
self._extensions = []
return self
def set_argument(self, arg, value):
"""设置浏览器配置的argument属性 \n
:param arg: 属性名
:param value: 属性值，有值的属性传入值，没有的传入bool
:return: 当前对象
"""
self.remove_argument(arg)
if value:
arg_str = arg if isinstance(value, bool) else f'{arg}={value}'
self.add_argument(arg_str)
return self
def set_timeouts(self, implicit=None, pageLoad=None, script=None):
"""设置超时时间，设置单位为秒，selenium4以上版本有效 \n
:param implicit: 查找元素超时时间
:param pageLoad: 页面加载超时时间
:param script: 脚本运行超时时间
:return: 当前对象
"""
if implicit is not None:
self.timeouts['implicit'] = implicit
if pageLoad is not None:
self.timeouts['pageLoad'] = pageLoad
if script is not None:
self.timeouts['script'] = script
return self
def set_headless(self, on_off=True):
"""设置是否隐藏浏览器界面 \n
:param on_off: 开或关
:return: 当前对象
"""
on_off = True if on_off else False
return self.set_argument('--headless', on_off)
def set_no_imgs(self, on_off=True):
"""设置是否加载图片 \n
:param on_off: 开或关
:return: 当前对象
"""
on_off = True if on_off else False
return self.set_argument('--blink-settings=imagesEnabled=false', on_off)
def set_no_js(self, on_off=True):
"""设置是否禁用js \n
:param on_off: 开或关
:return: 当前对象
"""
on_off = True if on_off else False
return self.set_argument('--disable-javascript', on_off)
def set_mute(self, on_off=True):
"""设置是否静音 \n
:param on_off: 开或关
:return: 当前对象
"""
on_off = True if on_off else False
return self.set_argument('--mute-audio', on_off)
def set_user_agent(self, user_agent):
"""设置user agent \n
:param user_agent: user agent文本
:return: 当前对象
"""
return self.set_argument('--user-agent', user_agent)
def set_proxy(self, proxy):
"""设置代理 \n
:param proxy: 代理url和端口
:return: 当前对象
"""
return self.set_argument('--proxy-server', proxy)
def set_page_load_strategy(self, value):
"""设置page_load_strategy，可接收 'normal', 'eager', 'none' \n
selenium4以上版本才支持此功能
normal：默认情况下使用, 等待所有资源下载完成
eager：DOM访问已准备就绪, 但其他资源 (如图像) 可能仍在加载中
none：完全不阻塞WebDriver
:param value: 可接收 'normal', 'eager', 'none'
:return: 当前对象
"""
if value not in ('normal', 'eager', 'none'):
raise ValueError("只能选择'normal', 'eager', 'none'。")
self.page_load_strategy = value.lower()
return self
def set_paths(self, driver_path=None, chrome_path=None, browser_path=None, local_port=None,
debugger_address=None, download_path=None, user_data_path=None, cache_path=None):
"""快捷的路径设置函数 \n
:param driver_path: chromedriver.exe路径
:param chrome_path: chrome.exe路径
:param browser_path: 浏览器可执行文件路径
:param local_port: 本地端口号
:param debugger_address: 调试浏览器地址，例：127.0.0.1:9222
:param download_path: 下载文件路径
:param user_data_path: 用户数据路径
:param cache_path: 缓存路径
:return: 当前对象
"""
if driver_path is not None:
self._driver_path = str(driver_path)
if chrome_path is not None:
self.binary_location = str(chrome_path)
if browser_path is not None:
self.binary_location = str(browser_path)
if local_port is not None:
self.debugger_address = '' if local_port == '' else f'127.0.0.1:{local_port}'
if debugger_address is not None:
self.debugger_address = debugger_address
if download_path is not None:
self._download_path = str(download_path)
if user_data_path is not None:
self.set_argument('--user-data-dir', str(user_data_path))
self._user_data_path = user_data_path
if cache_path is not None:
self.set_argument('--disk-cache-dir', str(cache_path))
return self
def as_dict(self):
"""已dict方式返回所有配置信息"""
return chrome_options_to_dict(self)
def chrome_options_to_dict(options):
"""把chrome配置对象转换为字典 \n
:param options: chrome配置对象，字典或DriverOptions对象
:return: 配置字典
"""
if options in (False, None):
return DriverOptions(read_file=False).as_dict()
if isinstance(options, dict):
return options
re_dict = dict()
attrs = ['debugger_address', 'binary_location', 'arguments', 'extensions', 'experimental_options', 'driver_path',
'page_load_strategy', 'download_path']
options_dir = options.__dir__()
for attr in attrs:
try:
re_dict[attr] = options.__getattribute__(attr) if attr in options_dir else None
except Exception:
pass
if 'timeouts' in options_dir and 'timeouts' in options._caps:
timeouts = options.__getattribute__('timeouts')
re_dict['timeouts'] = timeouts
return re_dict
def session_options_to_dict(options):
"""把session配置对象转换为字典 \n
:param options: session配置对象或字典
:return: 配置字典
"""
if options in (False, None):
return SessionOptions(read_file=False).as_dict()
if isinstance(options, dict):
return options
re_dict = dict()
attrs = ['headers', 'cookies', 'proxies', 'params', 'verify', 'stream', 'trust_env',
'max_redirects', 'timeout', 'download_path']
for attr in attrs:
val = options.__getattribute__(f'_{attr}')
if val is not None:
re_dict[attr] = val
return re_dict
def cookie_to_dict(cookie):
"""把Cookie对象转为dict格式 \n
:param cookie: Cookie对象
:return: cookie字典
"""
if isinstance(cookie, Cookie):
cookie_dict = cookie.__dict__.copy()
cookie_dict.pop('rfc2109')
cookie_dict.pop('_rest')
return cookie_dict
elif isinstance(cookie, dict):
cookie_dict = cookie
elif isinstance(cookie, str):
cookie = cookie.split(',' if ',' in cookie else ';')
cookie_dict = {}
for key, attr in enumerate(cookie):
attr_val = attr.lstrip().split('=')
if key == 0:
cookie_dict['name'] = attr_val[0]
cookie_dict['value'] = attr_val[1] if len(attr_val) == 2 else ''
else:
cookie_dict[attr_val[0]] = attr_val[1] if len(attr_val) == 2 else ''
return cookie_dict
else:
raise TypeError('cookie参数必须为Cookie、str或dict类型。')
return cookie_dict
def cookies_to_tuple(cookies):
"""把cookies转为tuple格式 \n
:param cookies: cookies信息，可为CookieJar, list, tuple, str, dict
:return: 返回tuple形式的cookies
"""
if isinstance(cookies, (list, tuple, RequestsCookieJar)):
cookies = tuple(cookie_to_dict(cookie) for cookie in cookies)
elif isinstance(cookies, str):
cookies = tuple(cookie_to_dict(cookie.lstrip()) for cookie in cookies.split(";"))
elif isinstance(cookies, dict):
cookies = tuple({'name': cookie, 'value': cookies[cookie]} for cookie in cookies)
else:
raise TypeError('cookies参数必须为RequestsCookieJar、list、tuple、str或dict类型。')
return cookies

229
DrissionPage/config.pyi
View File

@ -1,229 +0,0 @@
# -*- coding:utf-8 -*-
"""
@Author : g1879
@Contact : g1879@qq.com
"""
from configparser import RawConfigParser
from http.cookiejar import Cookie
from pathlib import Path
from typing import Any, Union, List, Tuple
from requests.adapters import HTTPAdapter
from requests.auth import HTTPBasicAuth
from requests.cookies import RequestsCookieJar
from selenium.webdriver.chrome.options import Options
class OptionsManager(object):
ini_path: str = ...
_conf: RawConfigParser = ...
paths: dict = ...
chrome_options: dict = ...
session_options: dict = ...
def __init__(self, path: str = None): ...
def __getattr__(self, item) -> dict: ...
def get_value(self, section: str, item: str) -> Any: ...
def get_option(self, section: str) -> dict: ...
def set_item(self, section: str, item: str, value: Any) -> None: ...
def remove_item(self, section: str, item: str) -> None: ...
def save(self, path: str = None) -> str: ...
def save_to_default(self) -> str: ...
class SessionOptions(object):
def __init__(self, read_file: bool = True, ini_path: str = None):
self.ini_path: str = ...
self._download_path: str = ...
self._headers: dict = ...
self._cookies: list = ...
self._auth: tuple = ...
self._proxies: dict = ...
self._hooks: dict = ...
self._params: dict = ...
self._verify: bool = ...
self._cert: Union[str, tuple] = ...
self._adapters: list = ...
self._stream: bool = ...
self._trust_env: bool = ...
self._max_redirects: int = ...
self._timeout: float = ...
self._del_set: set = ...
@property
def download_path(self) -> str: ...
def set_paths(self, download_path: Union[str, Path]) -> SessionOptions: ...
@property
def timeout(self) -> Union[int, float]: ...
def set_timeout(self, second: Union[int, float]) -> SessionOptions: ...
@property
def headers(self) -> dict: ...
def set_headers(self, headers: Union[dict, None]) -> SessionOptions: ...
def set_a_header(self, attr: str, value: str) -> SessionOptions: ...
def remove_a_header(self, attr: str) -> SessionOptions: ...
@property
def cookies(self) -> list: ...
def set_cookies(self, cookies: Union[RequestsCookieJar, list, tuple, str, dict, None]) -> SessionOptions: ...
@property
def auth(self) -> Union[Tuple[str, str], HTTPBasicAuth]: ...
def set_auth(self, auth: Union[Tuple[str, str], HTTPBasicAuth, None]) -> SessionOptions: ...
@property
def proxies(self) -> dict: ...
def set_proxies(self, http: Union[str, None], https: Union[str, None] = None) -> SessionOptions: ...
@property
def hooks(self) -> dict: ...
def set_hooks(self, hooks: Union[dict, None]) -> SessionOptions: ...
@property
def params(self) -> dict: ...
def set_params(self, params: Union[dict, None]) -> SessionOptions: ...
@property
def verify(self) -> bool: ...
def set_verify(self, on_off: Union[bool, None]) -> SessionOptions: ...
@property
def cert(self) -> Union[str, tuple]: ...
def set_cert(self, cert: Union[str, Tuple[str, str], None]) -> SessionOptions: ...
@property
def adapters(self): list: ...
def add_adapter(self, url: str, adapter: HTTPAdapter) -> SessionOptions: ...
@property
def stream(self) -> bool: ...
def set_stream(self, on_off: Union[bool, None]) -> SessionOptions: ...
@property
def trust_env(self) -> bool: ...
def set_trust_env(self, on_off: Union[bool, None]) -> SessionOptions: ...
@property
def max_redirects(self) -> int: ...
def _sets(self, arg: str, val: Any) -> None: ...
def set_max_redirects(self, times: Union[int, None]) -> SessionOptions: ...
def save(self, path: str = None) -> str: ...
def save_to_default(self) -> str: ...
def as_dict(self) -> dict: ...
class DriverOptions(Options):
def __init__(self, read_file: bool = True, ini_path: str = None):
self.ini_path: str = ...
self._driver_path: str = ...
self._user_data_path: str = ...
self._download_path: str = ...
@property
def driver_path(self) -> str: ...
@property
def download_path(self) -> str: ...
@property
def chrome_path(self) -> str: ...
@property
def browser_path(self) -> str: ...
@property
def user_data_path(self) -> str: ...
# -------------重写父类方法，实现链式操作-------------
def add_argument(self, argument: str) -> DriverOptions: ...
def set_capability(self, name: str, value: str) -> DriverOptions: ...
def add_extension(self, extension: str) -> DriverOptions: ...
def add_encoded_extension(self, extension: str) -> DriverOptions: ...
def add_experimental_option(self, name: str, value: Union[str, int, dict, List[str]]) -> DriverOptions: ...
# -------------重写父类方法结束-------------
def save(self, path: str = None) -> str: ...
def save_to_default(self) -> str: ...
def remove_argument(self, value: str) -> DriverOptions: ...
def remove_experimental_option(self, key: str) -> DriverOptions: ...
def remove_all_extensions(self) -> DriverOptions: ...
def set_argument(self, arg: str, value: Union[bool, str]) -> DriverOptions: ...
def set_timeouts(self, implicit: float = None, pageLoad: float = None, script: float = None) -> DriverOptions: ...
def set_headless(self, on_off: bool = True) -> DriverOptions: ...
def set_no_imgs(self, on_off: bool = True) -> DriverOptions: ...
def set_no_js(self, on_off: bool = True) -> DriverOptions: ...
def set_mute(self, on_off: bool = True) -> DriverOptions: ...
def set_user_agent(self, user_agent: str) -> DriverOptions: ...
def set_proxy(self, proxy: str) -> DriverOptions: ...
def set_page_load_strategy(self, value: str) -> DriverOptions: ...
def set_paths(self,
driver_path: Union[str, Path] = None,
chrome_path: Union[str, Path] = None,
browser_path: Union[str, Path] = None,
local_port: Union[int, str] = None,
debugger_address: str = None,
download_path: str = None,
user_data_path: str = None,
cache_path: str = None) -> DriverOptions: ...
def as_dict(self) -> dict: ...
def chrome_options_to_dict(options: Union[dict, DriverOptions, Options, None, bool]) -> Union[dict, None]: ...
def session_options_to_dict(options: Union[dict, SessionOptions, None]) -> Union[dict, None]: ...
def cookie_to_dict(cookie: Union[Cookie, str, dict]) -> dict: ...
def cookies_to_tuple(cookies: Union[RequestsCookieJar, list, tuple, str, dict]) -> tuple: ...

388
DrissionPage/configs/chromium_options.py Normal file
View File

@ -0,0 +1,388 @@
# -*- coding:utf-8 -*-
from pathlib import Path
from shutil import rmtree
from DrissionPage.common import port_is_using
from .options_manage import OptionsManager
class ChromiumOptions(object):
def __init__(self, read_file=True, ini_path=None):
"""初始化，默认从文件读取设置 \n
:param read_file: 是否从默认ini文件中读取配置信息
:param ini_path: ini文件路径，为None则读取默认ini文件
"""
self._user_data_path = None
self._user = 'Default'
if read_file:
self.ini_path = ini_path or str(Path(__file__).parent / 'configs.ini')
om = OptionsManager(self.ini_path)
options = om.chrome_options
self._download_path = om.paths.get('download_path', None)
self._arguments = options.get('arguments', [])
self._binary_location = options.get('binary_location', '')
self._extensions = options.get('extensions', [])
self._prefs = options.get('experimental_options', {}).get('prefs', {})
self._debugger_address = options.get('debugger_address', None)
self._page_load_strategy = options.get('page_load_strategy', 'normal')
self._proxy = om.proxies.get('http', None)
user_path = user = False
for arg in self._arguments:
if arg.startswith('--user-data-dir='):
self.set_paths(user_data_path=arg[16:])
user_path = True
if arg.startswith('--profile-directory='):
self.set_user(arg[20:])
user = True
if user and user_path:
break
timeouts = om.timeouts
self._timeouts = {'implicit': timeouts['implicit'],
'pageLoad': timeouts['page_load'],
'script': timeouts['script']}
return
self.ini_path = None
self._binary_location = ""
self._arguments = []
self._download_path = None
self._extensions = []
self._prefs = {}
self._prefs_to_del = []
self._timeouts = {'implicit': 10, 'pageLoad': 30, 'script': 30}
self._debugger_address = '127.0.0.1:9222'
self._page_load_strategy = 'normal'
self._proxy = None
@property
def download_path(self):
"""默认下载路径文件路径"""
return self._download_path
@property
def browser_path(self):
"""浏览器启动文件路径"""
return self._binary_location
@property
def user_data_path(self):
"""返回用户数据文件夹路径"""
return self._user_data_path
@property
def user(self):
"""返回用户配置文件夹名称"""
return self._user
@property
def page_load_strategy(self):
"""返回页面加载策略设置"""
return self._page_load_strategy
@property
def timeouts(self):
"""返回timeouts设置"""
return self._timeouts
@property
def proxy(self):
"""返回代理设置"""
return self._proxy
@property
def debugger_address(self):
"""返回浏览器地址，ip:port"""
return self._debugger_address
@property
def extensions(self):
"""以list形式返回要加载的插件路径"""
return self._extensions
@property
def preferences(self):
"""返回用户首选项配置"""
return self._prefs
def set_argument(self, arg, value=None):
"""设置浏览器配置的argument属性 \n
:param arg: 属性名
:param value: 属性值，有值的属性传入值，没有的传入None，如传入False，删除该项
:return: 当前对象
"""
self.remove_argument(arg)
if value is not False:
arg_str = arg if value is None else f'{arg}={value}'
self._arguments.append(arg_str)
return self
def remove_argument(self, value):
"""移除一个argument项 \n
:param value: 设置项名，有值的设置项传入设置名称即可
:return: 当前对象
"""
del_list = []
for argument in self._arguments:
if argument == value or argument.startswith(f'{value}='):
del_list.append(argument)
for del_arg in del_list:
self._arguments.remove(del_arg)
return self
def add_extension(self, path):
"""添加插件 \n
:param path: 插件路径，可指向文件夹
:return: 当前对象
"""
path = Path(path)
if not path.exists():
raise OSError('插件路径不存在。')
self._extensions.append(str(path))
return self
def remove_extensions(self):
"""移除所有插件 \n
:return: 当前对象
"""
self._extensions = []
return self
def set_pref(self, arg, value):
"""设置Preferences文件中的用户设置项
:param arg: 设置项名称
:param value: 设置项值
:return: 当前对象
"""
self._prefs[arg] = value
return self
def remove_pref(self, arg):
"""删除用户首选项设置，不能删除已设置到文件中的项 \n
:param arg: 设置项名称
:return: 当前对象
"""
self._prefs.pop(arg)
return self
def remove_pref_from_file(self, arg):
"""删除用户配置文件中已设置的项 \n
:param arg: 设置项名称
:return: 当前对象
"""
self._prefs_to_del.append(arg)
return self
def set_timeouts(self, implicit=None, pageLoad=None, script=None):
"""设置超时时间，单位为秒 \n
:param implicit: 默认超时时间
:param pageLoad: 页面加载超时时间
:param script: 脚本运行超时时间
:return: 当前对象
"""
if implicit is not None:
self._timeouts['implicit'] = implicit
if pageLoad is not None:
self._timeouts['pageLoad'] = pageLoad
if script is not None:
self._timeouts['script'] = script
return self
def set_user(self, user='Default'):
"""设置使用哪个用户配置文件夹 \n
:param user: 用户文件夹名称
:return: 当前对象
"""
self.set_argument('--profile-directory', user)
self._user = user
return self
def set_headless(self, on_off=True):
"""设置是否隐藏浏览器界面 \n
:param on_off: 开或关
:return: 当前对象
"""
on_off = True if on_off else False
return self.set_argument('--headless', on_off)
def set_no_imgs(self, on_off=True):
"""设置是否加载图片 \n
:param on_off: 开或关
:return: 当前对象
"""
on_off = True if on_off else False
return self.set_argument('--blink-settings=imagesEnabled=false', on_off)
def set_no_js(self, on_off=True):
"""设置是否禁用js \n
:param on_off: 开或关
:return: 当前对象
"""
on_off = True if on_off else False
return self.set_argument('--disable-javascript', on_off)
def set_mute(self, on_off=True):
"""设置是否静音 \n
:param on_off: 开或关
:return: 当前对象
"""
on_off = True if on_off else False
return self.set_argument('--mute-audio', on_off)
def set_user_agent(self, user_agent):
"""设置user agent \n
:param user_agent: user agent文本
:return: 当前对象
"""
return self.set_argument('--user-agent', user_agent)
def set_proxy(self, proxy):
"""设置代理 \n
:param proxy: 代理url和端口
:return: 当前对象
"""
self._proxy = proxy
return self.set_argument('--proxy-server', proxy)
def set_page_load_strategy(self, value):
"""设置page_load_strategy，可接收 'normal', 'eager', 'none' \n
selenium4以上版本才支持此功能
normal：默认情况下使用, 等待所有资源下载完成
eager：DOM访问已准备就绪, 但其他资源 (如图像) 可能仍在加载中
none：完全不阻塞WebDriver
:param value: 可接收 'normal', 'eager', 'none'
:return: 当前对象
"""
if value not in ('normal', 'eager', 'none'):
raise ValueError("只能选择'normal', 'eager', 'none'。")
self._page_load_strategy = value.lower()
return self
def set_paths(self, browser_path=None, local_port=None, debugger_address=None, download_path=None,
user_data_path=None, cache_path=None):
"""快捷的路径设置函数 \n
:param browser_path: 浏览器可执行文件路径
:param local_port: 本地端口号
:param debugger_address: 调试浏览器地址，例：127.0.0.1:9222
:param download_path: 下载文件路径
:param user_data_path: 用户数据路径
:param cache_path: 缓存路径
:return: 当前对象
"""
if browser_path is not None:
self._binary_location = str(browser_path)
if local_port is not None:
self._debugger_address = f'127.0.0.1:{local_port}'
if debugger_address is not None:
self._debugger_address = debugger_address
if download_path is not None:
self._download_path = str(download_path)
if user_data_path is not None:
u = str(user_data_path)
self.set_argument('--user-data-dir', u)
self._user_data_path = u
if cache_path is not None:
self.set_argument('--disk-cache-dir', str(cache_path))
return self
def auto_port(self, data_path):
"""自动获取可用端口 \n
:param data_path: 用户文件夹保存路径
:return: 当前对象
"""
port, path = PortFinder().get_port(data_path)
self.set_paths(local_port=port, user_data_path=path)
return self
def save(self, path=None):
"""保存设置到文件 \n
:param path: ini文件的路径， None 保存到当前读取的配置文件，传入 'default' 保存到默认ini文件
:return: 保存文件的绝对路径
"""
if path == 'default':
path = (Path(__file__).parent / 'configs.ini').absolute()
elif path is None:
if self.ini_path:
path = Path(self.ini_path).absolute()
else:
path = (Path(__file__).parent / 'configs.ini').absolute()
else:
path = Path(path).absolute()
path = path / 'config.ini' if path.is_dir() else path
if path.exists():
om = OptionsManager(str(path))
else:
om = OptionsManager(self.ini_path or str(Path(__file__).parent / 'configs.ini'))
# 设置chrome_options
attrs = ('debugger_address', 'binary_location', 'arguments', 'extensions', 'user', 'page_load_strategy')
for i in attrs:
om.set_item('chrome_options', i, self.__getattribute__(f'_{i}'))
# 设置代理
om.set_item('proxies', 'http', self._proxy)
om.set_item('proxies', 'https', self._proxy)
# 设置路径
om.set_item('paths', 'download_path', self._download_path)
# 设置timeout
om.set_item('timeouts', 'implicit', self._timeouts['implicit'])
om.set_item('timeouts', 'page_load', self._timeouts['pageLoad'])
om.set_item('timeouts', 'script', self._timeouts['script'])
# 设置prefs
eo = om.chrome_options.get('experimental_options', {})
eo['prefs'] = self._prefs
om.set_item('chrome_options', 'experimental_options', eo)
path = str(path)
om.save(path)
return path
def save_to_default(self):
"""保存当前配置到默认ini文件"""
return self.save('default')
class PortFinder(object):
used_port = []
@staticmethod
def get_port(path):
"""查找一个可用端口 \n
:param path: 用户文件夹保存路径
:return: 可以使用的端口和用户文件夹路径组成的元组
"""
path = Path(path)
for i in range(9600, 9800):
if i in PortFinder.used_port or port_is_using('127.0.0.1', i):
continue
path = path / f'userData{i}'
if not path.exists():
PortFinder.used_port.append(i)
return i, str(path)
try:
rmtree(path)
except PermissionError:
continue
raise OSError('未找到可用端口。')

101
DrissionPage/configs/chromium_options.pyi Normal file
View File

@ -0,0 +1,101 @@
# -*- coding:utf-8 -*-
from pathlib import Path
from typing import Union, Tuple, Any
class ChromiumOptions(object):
def __init__(self, read_file: bool = True, ini_path: str = None):
self.ini_path: str = ...
self._driver_path: str = ...
self._user_data_path: str = ...
self._download_path: str = ...
self._arguments: list = ...
self._binary_location: str = ...
self._user: str = ...
self._page_load_strategy: str = ...
self._timeouts: dict = ...
self._proxy: str = ...
self._debugger_address: str = ...
self._extensions: list = ...
self._prefs: dict = ...
self._prefs_to_del: list = ...
@property
def download_path(self) -> str: ...
@property
def browser_path(self) -> str: ...
@property
def user_data_path(self) -> str: ...
@property
def user(self) -> str: ...
@property
def page_load_strategy(self) -> str: ...
@property
def timeouts(self) -> dict: ...
@property
def proxy(self) -> str: ...
@property
def debugger_address(self) -> str: ...
@property
def extensions(self) -> list: ...
@property
def preferences(self) -> dict: ...
def set_argument(self, arg: str, value: Union[str, None, bool] = None) -> ChromiumOptions: ...
def remove_argument(self, value: str) -> ChromiumOptions: ...
def add_extension(self, path: Union[str, Path]) -> ChromiumOptions: ...
def remove_extensions(self) -> ChromiumOptions: ...
def set_pref(self, arg: str, value: Any) -> ChromiumOptions: ...
def remove_pref(self, arg: str) -> ChromiumOptions: ...
def remove_pref_from_file(self, arg: str) -> ChromiumOptions: ...
def set_timeouts(self, implicit: Union[int, float] = None, pageLoad: Union[int, float] = None,
script: Union[int, float] = None) -> ChromiumOptions: ...
def set_user(self, user: str = 'Default') -> ChromiumOptions: ...
def set_headless(self, on_off: bool = True) -> ChromiumOptions: ...
def set_no_imgs(self, on_off: bool = True) -> ChromiumOptions: ...
def set_no_js(self, on_off: bool = True) -> ChromiumOptions: ...
def set_mute(self, on_off: bool = True) -> ChromiumOptions: ...
def set_user_agent(self, user_agent: str) -> ChromiumOptions: ...
def set_proxy(self, proxy: str) -> ChromiumOptions: ...
def set_page_load_strategy(self, value: str) -> ChromiumOptions: ...
def set_paths(self, browser_path: Union[str, Path] = None, local_port: Union[int, str] = None,
debugger_address: str = None, download_path: Union[str, Path] = None,
user_data_path: Union[str, Path] = None, cache_path: Union[str, Path] = None) -> ChromiumOptions: ...
def auto_port(self, data_path: Union[str, Path]) -> ChromiumOptions: ...
def save(self, path: Union[str, Path] = None) -> str: ...
def save_to_default(self) -> str: ...
class PortFinder(object):
used_port: list = ...
@staticmethod
def get_port(path: Union[str, Path]) -> Tuple[int, str]: ...

358
DrissionPage/configs/driver_options.py Normal file
View File

@ -0,0 +1,358 @@
# -*- coding:utf-8 -*-
from pathlib import Path
from selenium.webdriver.chrome.options import Options
from .options_manage import OptionsManager
class DriverOptions(Options):
"""chrome浏览器配置类，继承自selenium.webdriver.chrome.options的Options类，
增加了删除配置和保存到文件方法。
"""
def __init__(self, read_file=True, ini_path=None):
"""初始化，默认从文件读取设置 \n
:param read_file: 是否从默认ini文件中读取配置信息
:param ini_path: ini文件路径，为None则读取默认ini文件
"""
super().__init__()
self._user_data_path = None
if read_file:
self.ini_path = ini_path or str(Path(__file__).parent / 'configs.ini')
om = OptionsManager(self.ini_path)
options_dict = om.chrome_options
self._driver_path = om.paths.get('chromedriver_path', None)
self._download_path = om.paths.get('download_path', None)
self._binary_location = options_dict.get('binary_location', '')
self._arguments = options_dict.get('arguments', [])
self._extensions = options_dict.get('extensions', [])
self._experimental_options = options_dict.get('experimental_options', {})
self._debugger_address = options_dict.get('debugger_address', None)
self.page_load_strategy = options_dict.get('page_load_strategy', 'normal')
for arg in self._arguments:
if arg.startswith('--user-data-dir='):
self.set_paths(user_data_path=arg[16:])
break
self.timeouts = options_dict.get('timeouts', {'implicit': 10, 'pageLoad': 30, 'script': 30})
return
self._driver_path = None
self._download_path = None
self.ini_path = None
self.timeouts = {'implicit': 10, 'pageLoad': 30, 'script': 30}
self._debugger_address = '127.0.0.1:9222'
@property
def driver_path(self):
"""chromedriver文件路径"""
return self._driver_path
@property
def download_path(self):
"""默认下载路径文件路径"""
return self._download_path
@property
def chrome_path(self):
"""浏览器启动文件路径"""
return self.browser_path
@property
def browser_path(self):
"""浏览器启动文件路径"""
return self.binary_location or 'chrome'
@property
def user_data_path(self):
"""返回用户文件夹路径"""
return self._user_data_path
# -------------重写父类方法，实现链式操作-------------
def add_argument(self, argument):
"""添加一个配置项 \n
:param argument: 配置项内容
:return: 当前对象
"""
super().add_argument(argument)
return self
def set_capability(self, name, value):
"""设置一个capability \n
:param name: capability名称
:param value: capability值
:return: 当前对象
"""
super().set_capability(name, value)
return self
def add_extension(self, extension):
"""添加插件 \n
:param extension: crx文件路径
:return: 当前对象
"""
super().add_extension(extension)
return self
def add_encoded_extension(self, extension):
"""将带有扩展数据的 Base64 编码字符串添加到将用于将其提取到 ChromeDriver 的列表中 \n
:param extension: 带有扩展数据的 Base64 编码字符串
:return: 当前对象
"""
super().add_encoded_extension(extension)
return self
def add_experimental_option(self, name, value):
"""添加一个实验选项到浏览器 \n
:param name: 选项名称
:param value: 选项值
:return: 当前对象
"""
super().add_experimental_option(name, value)
return self
# -------------重写父类方法结束-------------
def save(self, path=None):
"""保存设置到文件 \n
:param path: ini文件的路径， None 保存到当前读取的配置文件，传入 'default' 保存到默认ini文件
:return: 保存文件的绝对路径
"""
if path == 'default':
path = (Path(__file__).parent / 'configs.ini').absolute()
elif path is None:
if self.ini_path:
path = Path(self.ini_path).absolute()
else:
path = (Path(__file__).parent / 'configs.ini').absolute()
else:
path = Path(path).absolute()
path = path / 'config.ini' if path.is_dir() else path
if path.exists():
om = OptionsManager(str(path))
else:
om = OptionsManager(self.ini_path or str(Path(__file__).parent / 'configs.ini'))
options = self.as_dict()
for i in options:
if i == 'driver_path':
om.set_item('paths', 'chromedriver_path', options[i])
elif i == 'download_path':
om.set_item('paths', 'download_path', options[i])
else:
om.set_item('chrome_options', i, options[i])
path = str(path)
om.save(path)
return path
def save_to_default(self):
"""保存当前配置到默认ini文件"""
return self.save('default')
def remove_argument(self, value):
"""移除一个argument项 \n
:param value: 设置项名，有值的设置项传入设置名称即可
:return: 当前对象
"""
del_list = []
for argument in self._arguments:
if argument.startswith(value):
del_list.append(argument)
for del_arg in del_list:
self._arguments.remove(del_arg)
return self
def remove_experimental_option(self, key):
"""移除一个实验设置，传入key值删除 \n
:param key: 实验设置的名称
:return: 当前对象
"""
if key in self._experimental_options:
self._experimental_options.pop(key)
return self
def remove_all_extensions(self):
"""移除所有插件 \n
:return: 当前对象
"""
# 因插件是以整个文件储存，难以移除其中一个，故如须设置则全部移除再重设
self._extensions = []
return self
def set_argument(self, arg, value):
"""设置浏览器配置的argument属性 \n
:param arg: 属性名
:param value: 属性值，有值的属性传入值，没有的传入bool
:return: 当前对象
"""
self.remove_argument(arg)
if value:
arg_str = arg if isinstance(value, bool) else f'{arg}={value}'
self.add_argument(arg_str)
return self
def set_timeouts(self, implicit=None, pageLoad=None, script=None):
"""设置超时时间，设置单位为秒，selenium4以上版本有效 \n
:param implicit: 查找元素超时时间
:param pageLoad: 页面加载超时时间
:param script: 脚本运行超时时间
:return: 当前对象
"""
if implicit is not None:
self.timeouts['implicit'] = implicit
if pageLoad is not None:
self.timeouts['pageLoad'] = pageLoad
if script is not None:
self.timeouts['script'] = script
return self
def set_headless(self, on_off=True):
"""设置是否隐藏浏览器界面 \n
:param on_off: 开或关
:return: 当前对象
"""
on_off = True if on_off else False
return self.set_argument('--headless', on_off)
def set_no_imgs(self, on_off=True):
"""设置是否加载图片 \n
:param on_off: 开或关
:return: 当前对象
"""
on_off = True if on_off else False
return self.set_argument('--blink-settings=imagesEnabled=false', on_off)
def set_no_js(self, on_off=True):
"""设置是否禁用js \n
:param on_off: 开或关
:return: 当前对象
"""
on_off = True if on_off else False
return self.set_argument('--disable-javascript', on_off)
def set_mute(self, on_off=True):
"""设置是否静音 \n
:param on_off: 开或关
:return: 当前对象
"""
on_off = True if on_off else False
return self.set_argument('--mute-audio', on_off)
def set_user_agent(self, user_agent):
"""设置user agent \n
:param user_agent: user agent文本
:return: 当前对象
"""
return self.set_argument('--user-agent', user_agent)
def set_proxy(self, proxy):
"""设置代理 \n
:param proxy: 代理url和端口
:return: 当前对象
"""
return self.set_argument('--proxy-server', proxy)
def set_page_load_strategy(self, value):
"""设置page_load_strategy，可接收 'normal', 'eager', 'none' \n
selenium4以上版本才支持此功能
normal：默认情况下使用, 等待所有资源下载完成
eager：DOM访问已准备就绪, 但其他资源 (如图像) 可能仍在加载中
none：完全不阻塞WebDriver
:param value: 可接收 'normal', 'eager', 'none'
:return: 当前对象
"""
if value not in ('normal', 'eager', 'none'):
raise ValueError("只能选择'normal', 'eager', 'none'。")
self.page_load_strategy = value.lower()
return self
def set_paths(self, driver_path=None, chrome_path=None, browser_path=None, local_port=None,
debugger_address=None, download_path=None, user_data_path=None, cache_path=None):
"""快捷的路径设置函数 \n
:param driver_path: chromedriver.exe路径
:param chrome_path: chrome.exe路径
:param browser_path: 浏览器可执行文件路径
:param local_port: 本地端口号
:param debugger_address: 调试浏览器地址，例：127.0.0.1:9222
:param download_path: 下载文件路径
:param user_data_path: 用户数据路径
:param cache_path: 缓存路径
:return: 当前对象
"""
if driver_path is not None:
self._driver_path = str(driver_path)
if chrome_path is not None:
self.binary_location = str(chrome_path)
if browser_path is not None:
self.binary_location = str(browser_path)
if local_port is not None:
self.debugger_address = '' if local_port == '' else f'127.0.0.1:{local_port}'
if debugger_address is not None:
self.debugger_address = debugger_address
if download_path is not None:
self._download_path = str(download_path)
if user_data_path is not None:
self.set_argument('--user-data-dir', str(user_data_path))
self._user_data_path = user_data_path
if cache_path is not None:
self.set_argument('--disk-cache-dir', str(cache_path))
return self
def as_dict(self):
"""已dict方式返回所有配置信息"""
return chrome_options_to_dict(self)
def chrome_options_to_dict(options):
"""把chrome配置对象转换为字典 \n
:param options: chrome配置对象，字典或DriverOptions对象
:return: 配置字典
"""
if options in (False, None):
return DriverOptions(read_file=False).as_dict()
if isinstance(options, dict):
return options
re_dict = dict()
attrs = ['debugger_address', 'binary_location', 'arguments', 'extensions', 'experimental_options', 'driver_path',
'page_load_strategy', 'download_path']
options_dir = options.__dir__()
for attr in attrs:
try:
re_dict[attr] = options.__getattribute__(attr) if attr in options_dir else None
except Exception:
pass
if 'timeouts' in options_dir and 'timeouts' in options._caps:
timeouts = options.__getattribute__('timeouts')
re_dict['timeouts'] = timeouts
return re_dict

85
DrissionPage/configs/driver_options.pyi Normal file
View File

@ -0,0 +1,85 @@
# -*- coding:utf-8 -*-
from pathlib import Path
from typing import Union, List
from selenium.webdriver.chrome.options import Options
class DriverOptions(Options):
def __init__(self, read_file: bool = True, ini_path: str = None):
self.ini_path: str = ...
self._driver_path: str = ...
self._user_data_path: str = ...
self._download_path: str = ...
@property
def driver_path(self) -> str: ...
@property
def download_path(self) -> str: ...
@property
def chrome_path(self) -> str: ...
@property
def browser_path(self) -> str: ...
@property
def user_data_path(self) -> str: ...
# -------------重写父类方法，实现链式操作-------------
def add_argument(self, argument: str) -> DriverOptions: ...
def set_capability(self, name: str, value: str) -> DriverOptions: ...
def add_extension(self, extension: str) -> DriverOptions: ...
def add_encoded_extension(self, extension: str) -> DriverOptions: ...
def add_experimental_option(self, name: str, value: Union[str, int, dict, List[str]]) -> DriverOptions: ...
# -------------重写父类方法结束-------------
def save(self, path: str = None) -> str: ...
def save_to_default(self) -> str: ...
def remove_argument(self, value: str) -> DriverOptions: ...
def remove_experimental_option(self, key: str) -> DriverOptions: ...
def remove_all_extensions(self) -> DriverOptions: ...
def set_argument(self, arg: str, value: Union[bool, str]) -> DriverOptions: ...
def set_timeouts(self, implicit: float = None, pageLoad: float = None, script: float = None) -> DriverOptions: ...
def set_headless(self, on_off: bool = True) -> DriverOptions: ...
def set_no_imgs(self, on_off: bool = True) -> DriverOptions: ...
def set_no_js(self, on_off: bool = True) -> DriverOptions: ...
def set_mute(self, on_off: bool = True) -> DriverOptions: ...
def set_user_agent(self, user_agent: str) -> DriverOptions: ...
def set_proxy(self, proxy: str) -> DriverOptions: ...
def set_page_load_strategy(self, value: str) -> DriverOptions: ...
def set_paths(self,
driver_path: Union[str, Path] = None,
chrome_path: Union[str, Path] = None,
browser_path: Union[str, Path] = None,
local_port: Union[int, str] = None,
debugger_address: str = None,
download_path: str = None,
user_data_path: str = None,
cache_path: str = None) -> DriverOptions: ...
def as_dict(self) -> dict: ...
def chrome_options_to_dict(options: Union[dict, DriverOptions, Options, None, bool]) -> Union[dict, None]: ...

102
DrissionPage/configs/options_manage.py Normal file
View File

@ -0,0 +1,102 @@
# -*- coding:utf-8 -*-
from configparser import RawConfigParser, NoSectionError, NoOptionError
from pathlib import Path
class OptionsManager(object):
"""管理配置文件内容的类"""
def __init__(self, path=None):
"""初始化，读取配置文件，如没有设置临时文件夹，则设置并新建 \n
:param path: ini文件的路径，默认读取模块文件夹下的
"""
self.ini_path = str(Path(__file__).parent / 'configs.ini') if path == 'default' or path is None else path
if not Path(self.ini_path).exists():
raise FileNotFoundError('ini文件不存在。')
self._conf = RawConfigParser()
self._conf.read(self.ini_path, encoding='utf-8')
def __getattr__(self, item):
"""以dict形似返回获取大项信息
:param item: 项名
:return: None
"""
return self.get_option(item)
def get_value(self, section, item):
"""获取配置的值 \n
:param section: 段名
:param item: 项名
:return: 项值
"""
try:
return eval(self._conf.get(section, item))
except (SyntaxError, NameError):
return self._conf.get(section, item)
except NoSectionError and NoOptionError:
return None
def get_option(self, section):
"""把section内容以字典方式返回 \n
:param section: 段名
:return: 段内容生成的字典
"""
items = self._conf.items(section)
option = dict()
for j in items:
try:
option[j[0]] = eval(self._conf.get(section, j[0]))
except Exception:
option[j[0]] = self._conf.get(section, j[0])
return option
def set_item(self, section, item, value):
"""设置配置值 \n
:param section: 段名
:param item: 项名
:param value: 项值
:return: None
"""
self._conf.set(section, item, str(value))
self.__setattr__(f'_{section}', None)
return self
def remove_item(self, section, item):
"""删除配置值 \n
:param section: 段名
:param item: 项名
:return: None
"""
self._conf.remove_option(section, item)
return self
def save(self, path=None):
"""保存配置文件 \n
:param path: ini文件的路径，传入 'default' 保存到默认ini文件
:return: 保存路径
"""
default_path = (Path(__file__).parent / 'configs.ini').absolute()
if path == 'default':
path = default_path
elif path is None:
path = Path(self.ini_path).absolute()
else:
path = Path(path).absolute()
path = path / 'config.ini' if path.is_dir() else path
path = str(path)
self._conf.write(open(path, 'w', encoding='utf-8'))
print(f'配置已保存到文件：{path}')
if path == str(default_path):
print('以后程序可自动从文件加载配置。')
return path
def save_to_default(self):
"""保存当前配置到默认ini文件"""
return self.save('default')

30
DrissionPage/configs/options_manage.pyi Normal file
View File

@ -0,0 +1,30 @@
# -*- coding:utf-8 -*-
from configparser import RawConfigParser
from typing import Any
class OptionsManager(object):
ini_path: str = ...
_conf: RawConfigParser = ...
paths: dict = ...
chrome_options: dict = ...
session_options: dict = ...
timeouts: dict = ...
proxies: dict = ...
def __init__(self, path: str = None): ...
def __getattr__(self, item) -> dict: ...
def get_value(self, section: str, item: str) -> Any: ...
def get_option(self, section: str) -> dict: ...
def set_item(self, section: str, item: str, value: Any) -> None: ...
def remove_item(self, section: str, item: str) -> None: ...
def save(self, path: str = None) -> str: ...
def save_to_default(self) -> str: ...

395
DrissionPage/configs/session_options.py Normal file
View File

@ -0,0 +1,395 @@
# -*- coding:utf-8 -*-
from pathlib import Path
from DrissionPage.common import cookies_to_tuple
from .options_manage import OptionsManager
class SessionOptions(object):
"""requests的Session对象配置类"""
def __init__(self, read_file=True, ini_path=None):
"""
:param read_file: 是否从文件读取配置
:param ini_path: ini文件路径
"""
self.ini_path = None
self._download_path = None
self._headers = None
self._cookies = None
self._auth = None
self._proxies = None
self._hooks = None
self._params = None
self._verify = None
self._cert = None
self._adapters = None
self._stream = None
self._trust_env = None
self._max_redirects = None
self._timeout = 10
self._del_set = set() # 记录要从ini文件删除的参数
if read_file:
self.ini_path = ini_path or str(Path(__file__).parent / 'configs.ini')
om = OptionsManager(self.ini_path)
options_dict = om.session_options
if options_dict.get('headers', None) is not None:
self.set_headers(options_dict['headers'])
if options_dict.get('cookies', None) is not None:
self.set_cookies(options_dict['cookies'])
if options_dict.get('auth', None) is not None:
self._auth = options_dict['auth']
if options_dict.get('params', None) is not None:
self._params = options_dict['params']
if options_dict.get('verify', None) is not None:
self._verify = options_dict['verify']
if options_dict.get('cert', None) is not None:
self._cert = options_dict['cert']
if options_dict.get('stream', None) is not None:
self._stream = options_dict['stream']
if options_dict.get('trust_env', None) is not None:
self._trust_env = options_dict['trust_env']
if options_dict.get('max_redirects', None) is not None:
self._max_redirects = options_dict['max_redirects']
self.set_proxies(om.proxies.get('http', None), om.proxies.get('https', None))
self._timeout = om.timeouts.get('implicit', 10)
self._download_path = om.paths.get('download_path', None)
# ===========须独立处理的项开始============
@property
def download_path(self):
"""返回默认下载路径属性信息"""
return self._download_path
def set_paths(self, download_path=None):
"""设置默认下载路径 \n
:param download_path: 下载路径
:return: 返回当前对象
"""
if download_path is not None:
self._download_path = str(download_path)
return self
@property
def timeout(self):
"""返回timeout属性信息"""
return self._timeout
def set_timeout(self, second):
"""设置超时信息
:param second: 秒数
:return: 返回当前对象
"""
self._timeout = second
return self
@property
def proxies(self):
"""返回proxies设置信息"""
if self._proxies is None:
self._proxies = {}
return self._proxies
def set_proxies(self, http, https=None):
"""设置proxies参数 \n
:param http: http代理地址
:param https: https代理地址
:return: 返回当前对象
"""
proxies = None if http == https is None else {'http': http, 'https': https or http}
self._sets('proxies', proxies)
return self
# ===========须独立处理的项结束============
@property
def headers(self):
"""返回headers设置信息"""
if self._headers is None:
self._headers = {}
return self._headers
def set_headers(self, headers):
"""设置headers参数 \n
:param headers: 参数值，传入None可在ini文件标记删除
:return: 返回当前对象
"""
if headers is None:
self._headers = None
self._del_set.add('headers')
else:
self._headers = {key.lower(): headers[key] for key in headers}
return self
def set_a_header(self, attr, value):
"""设置headers中一个项 \n
:param attr: 设置名称
:param value: 设置值
:return: 返回当前对象
"""
if self._headers is None:
self._headers = {}
self._headers[attr.lower()] = value
return self
def remove_a_header(self, attr):
"""从headers中删除一个设置 \n
:param attr: 要删除的设置
:return: 返回当前对象
"""
if self._headers is None:
return self
attr = attr.lower()
if attr in self._headers:
self._headers.pop(attr)
return self
@property
def cookies(self):
"""以list形式返回cookies"""
if self._cookies is None:
self._cookies = []
return self._cookies
def set_cookies(self, cookies):
"""设置cookies信息 \n
:param cookies: cookies，可为CookieJar, list, tuple, str, dict，传入None可在ini文件标记删除
:return: 返回当前对象
"""
cookies = cookies if cookies is None else list(cookies_to_tuple(cookies))
self._sets('cookies', cookies)
return self
@property
def auth(self):
"""返回auth设置信息"""
return self._auth
def set_auth(self, auth):
"""设置认证元组或对象 \n
:param auth: 认证元组或对象
:return: 返回当前对象
"""
self._sets('auth', auth)
return self
@property
def hooks(self):
"""返回回调方法"""
if self._hooks is None:
self._hooks = {}
return self._hooks
def set_hooks(self, hooks):
"""设置回调方法 \n
:param hooks:
:return: 返回当前对象
"""
self._hooks = hooks
return self
@property
def params(self):
"""返回params设置信息"""
if self._params is None:
self._params = {}
return self._params
def set_params(self, params):
"""设置查询参数字典 \n
:param params: 查询参数字典
:return: 返回当前对象
"""
self._sets('params', params)
return self
@property
def verify(self):
"""返回是否验证SSL证书设置"""
return self._verify
def set_verify(self, on_off):
"""设置是否验证SSL证书 \n
:param on_off: 是否验证 SSL 证书
:return: 返回当前对象
"""
self._sets('verify', on_off)
return self
@property
def cert(self):
"""返回cert设置信息"""
return self._cert
def set_cert(self, cert):
"""SSL客户端证书文件的路径(.pem格式)，或(‘cert’, ‘key’)元组 \n
:param cert: 证书路径或元组
:return: 返回当前对象
"""
self._sets('cert', cert)
return self
@property
def adapters(self):
"""返回适配器设置信息"""
if self._adapters is None:
self._adapters = []
return self._adapters
def add_adapter(self, url, adapter):
"""添加适配器 \n
:param url: 适配器对应url
:param adapter: 适配器对象
:return: 返回当前对象
"""
self._adapters.append((url, adapter))
return self
@property
def stream(self):
"""返回stream设置信息"""
return self._stream
def set_stream(self, on_off):
"""设置是否使用流式响应内容 \n
:param on_off: 是否使用流式响应内容
:return: 返回当前对象
"""
self._sets('stream', on_off)
return self
@property
def trust_env(self):
"""返回trust_env设置信息"""
return self._trust_env
def set_trust_env(self, on_off):
"""设置是否信任环境 \n
:param on_off: 是否信任环境
:return: 返回当前对象
"""
self._sets('trust_env', on_off)
return self
@property
def max_redirects(self):
"""返回最大重定向次数"""
return self._max_redirects
def set_max_redirects(self, times):
"""设置最大重定向次数 \n
:param times: 最大重定向次数
:return: 返回当前对象
"""
self._sets('max_redirects', times)
return self
def _sets(self, arg, val):
"""给属性赋值或标记删除
:param arg: 属性名称
:param val: 参数值
:return: None
"""
if val is None:
self.__setattr__(f'_{arg}', None)
self._del_set.add(arg)
else:
self.__setattr__(f'_{arg}', val)
if arg in self._del_set:
self._del_set.remove(arg)
def save(self, path=None):
"""保存设置到文件 \n
:param path: ini文件的路径，传入 'default' 保存到默认ini文件
:return: 保存文件的绝对路径
"""
if path == 'default':
path = (Path(__file__).parent / 'configs.ini').absolute()
elif path is None:
if self.ini_path:
path = Path(self.ini_path).absolute()
else:
path = (Path(__file__).parent / 'configs.ini').absolute()
else:
path = Path(path).absolute()
path = path / 'config.ini' if path.is_dir() else path
if path.exists():
om = OptionsManager(str(path))
else:
om = OptionsManager(self.ini_path or str(Path(__file__).parent / 'configs.ini'))
options = session_options_to_dict(self)
for i in options:
if i not in ('download_path', 'timeout', 'proxies'):
om.set_item('session_options', i, options[i])
om.set_item('paths', 'download_path', self.download_path)
om.set_item('timeouts', 'implicit', self.timeout)
om.set_item('proxies', 'http', self.proxies.get('http', None))
om.set_item('proxies', 'https', self.proxies.get('https', None))
for i in self._del_set:
if i == 'download_path':
om.set_item('paths', 'download_path', '')
elif i == 'proxies':
om.set_item('proxies', 'http', '')
om.set_item('proxies', 'https', '')
else:
om.remove_item('session_options', i)
path = str(path)
om.save(path)
return path
def save_to_default(self):
"""保存当前配置到默认ini文件"""
return self.save('default')
def as_dict(self):
"""以字典形式返回本对象"""
return session_options_to_dict(self)
def session_options_to_dict(options):
"""把session配置对象转换为字典 \n
:param options: session配置对象或字典
:return: 配置字典
"""
if options in (False, None):
return SessionOptions(read_file=False).as_dict()
if isinstance(options, dict):
return options
re_dict = dict()
attrs = ['headers', 'cookies', 'proxies', 'params', 'verify', 'stream', 'trust_env',
'max_redirects', 'timeout', 'download_path']
for attr in attrs:
val = options.__getattribute__(f'_{attr}')
if val is not None:
re_dict[attr] = val
return re_dict

111
DrissionPage/configs/session_options.pyi Normal file
View File

@ -0,0 +1,111 @@
# -*- coding:utf-8 -*-
from pathlib import Path
from typing import Any, Union, Tuple
from requests.adapters import HTTPAdapter
from requests.auth import HTTPBasicAuth
from requests.cookies import RequestsCookieJar
class SessionOptions(object):
def __init__(self, read_file: bool = True, ini_path: str = None):
self.ini_path: str = ...
self._download_path: str = ...
self._headers: dict = ...
self._cookies: list = ...
self._auth: tuple = ...
self._proxies: dict = ...
self._hooks: dict = ...
self._params: dict = ...
self._verify: bool = ...
self._cert: Union[str, tuple] = ...
self._adapters: list = ...
self._stream: bool = ...
self._trust_env: bool = ...
self._max_redirects: int = ...
self._timeout: float = ...
self._del_set: set = ...
@property
def download_path(self) -> str: ...
def set_paths(self, download_path: Union[str, Path]) -> SessionOptions: ...
@property
def timeout(self) -> Union[int, float]: ...
def set_timeout(self, second: Union[int, float]) -> SessionOptions: ...
@property
def headers(self) -> dict: ...
def set_headers(self, headers: Union[dict, None]) -> SessionOptions: ...
def set_a_header(self, attr: str, value: str) -> SessionOptions: ...
def remove_a_header(self, attr: str) -> SessionOptions: ...
@property
def cookies(self) -> list: ...
def set_cookies(self, cookies: Union[RequestsCookieJar, list, tuple, str, dict, None]) -> SessionOptions: ...
@property
def auth(self) -> Union[Tuple[str, str], HTTPBasicAuth]: ...
def set_auth(self, auth: Union[Tuple[str, str], HTTPBasicAuth, None]) -> SessionOptions: ...
@property
def proxies(self) -> dict: ...
def set_proxies(self, http: Union[str, None], https: Union[str, None] = None) -> SessionOptions: ...
@property
def hooks(self) -> dict: ...
def set_hooks(self, hooks: Union[dict, None]) -> SessionOptions: ...
@property
def params(self) -> dict: ...
def set_params(self, params: Union[dict, None]) -> SessionOptions: ...
@property
def verify(self) -> bool: ...
def set_verify(self, on_off: Union[bool, None]) -> SessionOptions: ...
@property
def cert(self) -> Union[str, tuple]: ...
def set_cert(self, cert: Union[str, Tuple[str, str], None]) -> SessionOptions: ...
@property
def adapters(self): list: ...
def add_adapter(self, url: str, adapter: HTTPAdapter) -> SessionOptions: ...
@property
def stream(self) -> bool: ...
def set_stream(self, on_off: Union[bool, None]) -> SessionOptions: ...
@property
def trust_env(self) -> bool: ...
def set_trust_env(self, on_off: Union[bool, None]) -> SessionOptions: ...
@property
def max_redirects(self) -> int: ...
def _sets(self, arg: str, val: Any) -> None: ...
def set_max_redirects(self, times: Union[int, None]) -> SessionOptions: ...
def save(self, path: str = None) -> str: ...
def save_to_default(self) -> str: ...
def as_dict(self) -> dict: ...
def session_options_to_dict(options: Union[dict, SessionOptions, None]) -> Union[dict, None]: ...

5
DrissionPage/drission.py
View File

@ -14,8 +14,9 @@ from selenium.webdriver.chrome.options import Options
from selenium.webdriver.remote.webdriver import WebDriver as RemoteWebDriver
from tldextract import extract
from .common import get_pid_from_port, connect_browser
from .config import SessionOptions, DriverOptions, cookies_to_tuple, session_options_to_dict
from .common import get_pid_from_port, connect_browser, cookies_to_tuple
from .configs.session_options import SessionOptions, session_options_to_dict
from .configs.driver_options import DriverOptions
class Drission(object):

3
DrissionPage/easy_set.py
View File

@ -12,7 +12,8 @@ from typing import Union
from selenium import webdriver
from .common import unzip
from .config import OptionsManager, DriverOptions
from .configs.options_manage import OptionsManager
from .configs.driver_options import DriverOptions
from .drission import Drission
from .session_page import SessionPage

3
DrissionPage/session_page.py
View File

@ -13,7 +13,8 @@ from requests.structures import CaseInsensitiveDict
from tldextract import extract
from .base import BasePage
from .config import SessionOptions, cookies_to_tuple, cookie_to_dict
from .configs.session_options import SessionOptions
from .common import cookies_to_tuple, cookie_to_dict
from .session_element import SessionElement, make_session_ele

4
DrissionPage/web_page.py
View File

@ -13,7 +13,9 @@ from .base import BasePage
from .chromium_base import ChromiumBase, Timeout
from .chromium_driver import ChromiumDriver
from .chromium_page import ChromiumPage, ChromiumDownloadSetter
from .config import DriverOptions, SessionOptions, cookies_to_tuple
from .configs.session_options import SessionOptions
from .configs.driver_options import DriverOptions
from .common import cookies_to_tuple
from .session_page import SessionPage