1 Star 0 Fork 0

Sync.Nep / djc_helper

加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
该仓库未声明开源许可证文件(LICENSE),使用请关注具体项目描述及其代码上游依赖。
克隆/下载
qq_login.py 96.11 KB
一键复制 编辑 原始数据 按行查看 历史
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058
from __future__ import annotations
# Generated by Selenium IDE
import datetime
import json
import logging
import os
import shutil
import threading
import time
from collections import Counter
from urllib.parse import quote_plus, unquote_plus
from selenium import webdriver
from selenium.common.exceptions import (
InvalidArgumentException,
NoSuchWindowException,
StaleElementReferenceException,
TimeoutException,
)
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.webdriver import WebDriver
from selenium.webdriver.common.action_chains import ActionChains
from selenium.webdriver.common.by import By
from selenium.webdriver.common.desired_capabilities import DesiredCapabilities
from selenium.webdriver.support import expected_conditions
from selenium.webdriver.support.ui import WebDriverWait
from alist import download_from_alist
from compress import decompress_dir_with_bandizip
from config import AccountConfig, CommonConfig
from config import config as get_config
from config import load_config
from config_cloud import config_cloud
from dao import GuanJiaUserInfo
from data_struct import ConfigInterface
from db import CaptchaDB, LoginRetryDB
from download import download_latest_github_release
from exceptions_def import (
GithubActionLoginException,
RequireVerifyMessageButInHeadlessMode,
SameAccountTryLoginAtMultipleThreadsException,
)
from first_run import is_first_run_in
from log import color, logger
from urls import get_act_url
from util import (
MiB,
async_message_box,
count_down,
download_chrome_driver,
get_file_or_directory_size,
get_screen_size,
is_run_in_github_action,
is_windows,
pause_and_exit,
range_from_one,
show_head_line,
truncate,
try_except,
use_by_myself,
)
from version import now_version, ver_time
if is_windows():
import win32con
class LoginResult(ConfigInterface):
def __init__(
self,
uin="",
skey="",
openid="",
p_skey="",
vuserid="",
qc_openid="",
qc_k="",
apps_p_skey="",
xinyue_openid="",
xinyue_access_token="",
qc_access_token="",
qc_nickname="",
iwan_openid="",
iwan_access_token="",
common_openid="",
common_access_token="",
):
super().__init__()
# 使用炎炎夏日活动界面得到
self.uin = uin
self.skey = skey
# 登录QQ空间得到
self.p_skey = p_skey
# 使用心悦活动界面得到
self.openid = openid
# 使用腾讯视频相关页面得到
self.vuserid = vuserid
# 登录电脑管家页面得到
self.qc_openid = qc_openid
self.qc_k = qc_k
self.qc_access_token = qc_access_token
self.qc_nickname = qc_nickname
# 分享用p_skey
self.apps_p_skey = apps_p_skey
# 心悦相关信息
self.xinyue_openid = xinyue_openid
self.xinyue_access_token = xinyue_access_token
# 爱玩相关
self.iwan_openid = iwan_openid
self.iwan_access_token = iwan_access_token
# 后续需要登录特定网页获取的openid和access-token都放在这里
self.common_openid = common_openid
self.common_access_token = common_access_token
self.guanjia_skey_version = 0
class QQLogin:
login_type_auto_login = "账密自动登录"
login_type_qr_login = "扫码登录"
login_mode_normal = "normal"
login_mode_xinyue = "xinyue"
login_mode_qzone = "qzone"
login_mode_guanjia = "guanjia"
login_mode_wegame = "wegame"
login_mode_club_vip = "club_vip"
login_mode_iwan = "iwan"
login_mode_supercore = "supercore"
login_mode_djc = "djc"
login_mode_to_description = {
login_mode_normal: "普通",
login_mode_xinyue: "心悦",
login_mode_qzone: "QQ空间",
login_mode_guanjia: "电脑管家",
login_mode_wegame: "Wegame",
login_mode_club_vip: "club.vip",
login_mode_iwan: "爱玩",
login_mode_supercore: "超享玩",
login_mode_djc: "道聚城",
}
bandizip_executable_path = os.path.realpath("./utils/bandizip_portable/bz.exe")
# re: chrome版本一键升级流程
# 0. 使用 _update_chrome.py 脚本,按照提示操作即可获取最新稳定版本chrome的便携版、driver、安装包等
# .
# note: chrome版本手动升级流程
# 1. 下载新版本chrome driver => chromedriver_{ver}.exe
# 1.1 https://sites.google.com/chromium.org/driver/downloads
# 2. 制作新版本便携版压缩包 => chrome_portable_{ver}.7z
# 2.1 获取安装包
# 2.1.1 找到系统安装的chrome的安装包
# 2.1.1.1 %PROGRAMFILES%\Google\Chrome\Application
# 2.1.1.2 %PROGRAMFILES(X86)%\Google\Chrome\Application
# 2.1.1.3 在这个目录中找到 90.0.4430.93\Installer\chrome.7z
# 2.1.1.4 90.0.4430.93可替换为最新版本的版本号
# 2.1.2 也可以从网上下载离线版安装包
# 2.1.2.1 下载地址
# 2.1.2.1.1 https://www.iplaysoft.com/tools/chrome/
# 2.1.2.2 下载内容形如90.0.4430.93_chrome_installer.exe,使用bandizip打开然后解压得到chrome.7z,即可进行下一步
# 2.2 将chrome.7z解压然后重新压缩,得到 chrome_portable_90.7z
# 2.2.1 确保chrome_portable_90.7z压缩包的首层目录形如(89.0.4389.72、chrome.exe、chrome_proxy.exe)
# 3. 替换chromedriver_{ver}.exe和chrome_portable_{ver}.7z到小助手 utils 目录下
# 3.1 修改版本号为 {ver} 后,测试下登录流程
# todo:
# 4. 下载新版本安装包 => Chrome_92.0.4515.131_普通安装包_非便携版.exe
# 4.1 https://www.iplaysoft.com/tools/chrome/
# 5. 上传以下内容到网盘的工具目录
# 5.1 chromedriver_{ver}.exe
# 5.1 chrome_portable_{ver}.7z
# 5.1 Chrome_92.0.4515.131_普通安装包_非便携版.exe
# undone:
# 6. 更新linux版的路径
# 6.1 _ubuntu_download_chrome_and_driver.sh
# 6.2 _centos_download_and_install_chrome_and_driver.sh
# re:
# 7. 入库以下文件
# qq_login.py
# chromedriver_{ver}.exe
# chrome_portable_{ver}.7z
# _centos_download_and_install_chrome_and_driver.sh
# _ubuntu_download_chrome_and_driver.sh
chrome_major_version = 107
chrome_driver_version = "107.0.5304.62"
default_window_width = 390
default_window_height = 360
mobile_emulation_qq = {
"deviceMetrics": {"width": 420, "height": 780},
"userAgent": "Mozilla/5.0 (Linux; U; Android 5.0.2; zh-cn; X900 Build/CBXCNOP5500912251S) AppleWebKit/533.1 (KHTML, like Gecko)Version/4.0 MQQBrowser/5.4 TBS/025489 Mobile Safari/533.1 V1_AND_SQ_6.0.0_300_YYB_D QQ/6.0.0.2605 NetType/WIFI WebP/0.3.0 Pixel/1440",
}
def __init__(self, common_config, window_index=1):
self.cfg: CommonConfig = common_config
self.driver: WebDriver | None = None
self.window_title = ""
self.time_start_login = datetime.datetime.now()
self.screen_width, self.screen_height = get_screen_size()
col_size, row_size = round(self.screen_width / self.default_window_width), round(
self.screen_height / self.default_window_height
)
self.window_position_x = self.default_window_width * ((window_index - 1) % col_size)
self.window_position_y = self.default_window_height * (((window_index - 1) // col_size) % row_size)
def prepare_chrome(self, ctx: str, login_type: str, login_url: str):
logger.info(
color("fg_bold_cyan")
+ f"{self.name} 正在初始化chrome driver(版本为{self.get_chrome_major_version()}),用以进行【{ctx}】相关操作。"
f"浏览器坐标:({self.window_position_x}, {self.window_position_y})@{self.default_window_width}*{self.default_window_height}({self.screen_width}*{self.screen_height})"
)
self.login_url = login_url
logger.info(color("bold_green") + f"{self.name} {ctx} 登录链接为: {login_url}")
if is_windows():
self.prepare_chrome_windows(login_type, login_url)
else:
self.prepare_chrome_linux(login_type, login_url)
self.cookies = self.driver.get_cookies()
def prepare_chrome_windows(self, login_type: str, login_url: str):
inited = False
try:
if not self.cfg.force_use_portable_chrome:
# 如果未强制使用便携版chrome,则首先尝试使用系统安装的chrome
options = self.new_options()
self.append_common_options(options, login_type, login_url)
self.driver = webdriver.Chrome(
service=Service(executable_path=self.chrome_driver_executable_path()),
options=options,
)
logger.info(color("bold_yellow") + f"{self.name} 使用自带chrome")
inited = True
except Exception:
pass
if not inited:
# 如果找不到,则尝试使用打包的便携版chrome
zip_name = os.path.basename(self.chrome_binary_7z())
# 判定本地是否有便携版压缩包,若无则说明自动下载失败,提示去网盘手动下载
if not os.path.isfile(self.chrome_binary_7z()):
zip_name = zip_name
installer_name = self.chrome_installer_name()
version = self.get_chrome_major_version()
chrome_root_directory = self.chrome_root_directory()
msg = (
"================ 这一段是问题描述 ================\n"
f"当前电脑未发现{version}版本的chrome浏览器,且{chrome_root_directory}目录下无便携版chrome浏览器的压缩包({zip_name})\n"
"\n"
"================ 这一段是解决方法 ================\n"
f"如果不想影响系统浏览器,请在稍后打开的网盘页面中下载[{zip_name}],并放到{chrome_root_directory}目录里(注意:是把这个压缩包原原本本地放到这个目录里,而不是解压后再放过来!!!),然后重新打开程序~\n"
"\n"
f"如果愿意装一个浏览器,请在稍后打开的网盘页面中下载{installer_name},下载完成后双击安装即可\n"
"\n"
"(一定要看清版本,如果发现网盘里的便携版和安装版版本都比提示里的高(比如这里提示87,网盘里显示89),建议直接下个最新的小助手压缩包,解压后把配置文件复制过去~)\n"
"\n"
"================ 这一段是补充说明 ================\n"
"1. 如果之前版本已经下载过这个文件,可以直接去之前版本复制过来~不需要再下载一次~\n"
"2. 如果之前一直都运行的好好的,今天突然不行了,可能是以下原因\n"
"2.1 系统安装的chrome自动升级到新版本了,当前小助手使用的驱动不支持该版本。解决办法:下载当前版本小助手对应版本的便携版chrome\n"
"2.2 新版小助手升级了驱动,当前系统安装的chrome或便携版chrome的版本太低了。解决办法:升级新版本chrome或下载新版本的便携版chrome\n"
"\n"
"------- 已经说得如此明白,如果还有人进群问,将直接踢出群聊 -------\n"
"------- 已经说得如此明白,如果还有人进群问,将直接踢出群聊 -------\n"
"------- 已经说得如此明白,如果还有人进群问,将直接踢出群聊 -------\n"
)
async_message_box(
msg,
f"你没有{self.get_chrome_major_version()}版本的chrome浏览器,需要安装完整版或下载便携版",
icon=win32con.MB_ICONERROR,
open_url="http://101.43.54.94:5244/%E6%96%87%E6%9C%AC%E7%BC%96%E8%BE%91%E5%99%A8%E3%80%81chrome%E6%B5%8F%E8%A7%88%E5%99%A8%E3%80%81autojs%E3%80%81HttpCanary%E7%AD%89%E5%B0%8F%E5%B7%A5%E5%85%B7",
)
pause_and_exit(-1)
# 然后使用本地的chrome来初始化driver对象
options = self.new_options()
options.binary_location = self.chrome_binary_location()
options.add_argument("--no-sandbox")
options.add_argument("--no-default-browser-check")
options.add_argument("--no-first-run")
self.append_common_options(options, login_type, login_url)
self.driver = webdriver.Chrome(
service=Service(executable_path=self.chrome_driver_executable_path()),
options=options,
)
logger.info(color("bold_yellow") + f"{self.name} 使用便携版chrome")
def prepare_chrome_linux(self, login_type: str, login_url: str):
# linux下只尝试使用系统安装的chrome
options = self.new_options()
options.binary_location = self.chrome_binary_location_linux()
options.add_argument("--no-sandbox")
options.add_argument("--disable-dev-shm-usage")
options.add_argument("--no-default-browser-check")
options.add_argument("--no-first-run")
self.append_common_options(options, login_type, login_url)
self.driver = webdriver.Chrome(
service=Service(executable_path=self.chrome_driver_executable_path_linux()),
options=options,
)
logger.info(color("bold_yellow") + f"{self.name} Linux环境下使用自带chrome")
def new_options(self) -> Options:
options = Options()
caps = DesiredCapabilities().CHROME
# caps["pageLoadStrategy"] = "normal" # Waits for full page load
caps["pageLoadStrategy"] = "none" # Do not wait for full page load
for k, v in caps.items():
options.set_capability(k, v)
return options
def append_common_options(self, options: Options, login_type: str, login_url: str):
options.add_argument(f"window-position={self.window_position_x},{self.window_position_y}")
options.add_argument(f"window-size={self.default_window_width},{self.default_window_height}")
options.add_argument(f"app={login_url}")
# 设置静音
options.add_argument("--mute-audio")
exclude_switches = []
if not self.cfg._debug_show_chrome_logs:
exclude_switches.append("enable-logging")
selenium_logger = logging.getLogger("selenium.webdriver.remote.remote_connection")
selenium_logger.setLevel(logging.WARNING)
# 使用Selenium期间将urllib的日志关闭
urllib_logger = logging.getLogger("urllib3.connectionpool")
urllib_logger.setLevel(logging.WARNING)
if self.cfg.run_in_headless_mode:
if login_type == self.login_type_auto_login:
logger.warning(f"{self.name} 已配置在自动登录模式时使用headless模式运行chrome")
options.add_argument("--headless")
else:
logger.warning(f"{self.name} 扫码登录模式不使用headless模式")
# 特殊处理linux环境
if not is_windows():
options.add_argument("--headless")
logger.warning(f"{self.name} 在linux环境下强制使用headless模式运行chrome")
# 隐藏提示:Chrome 正收到自动测试软件的控制。
exclude_switches.append("enable-automation")
if len(exclude_switches) != 0:
options.add_experimental_option("excludeSwitches", exclude_switches)
# 隐藏保存密码的提示窗
options.add_experimental_option(
"prefs", {"credentials_enable_service": False, "profile": {"password_manager_enabled": False}}
)
if self.login_mode == self.login_mode_supercore:
logger.info("当前是超享玩登录,将设置设备信息为手机qq,否则不能正常登录")
options.add_experimental_option("mobileEmulation", self.mobile_emulation_qq)
def destroy_chrome(self):
logger.info(f"{self.name} 释放chrome实例")
if self.driver is not None:
# 最小化网页
if is_windows():
self.driver.minimize_window()
threading.Thread(target=self.driver.quit, daemon=True).start()
# 使用Selenium结束将日志级别改回去
urllib_logger = logging.getLogger("urllib3.connectionpool")
urllib_logger.setLevel(logger.level)
@try_except(extra_msg="自动下载缺失的dlc失败,请根据上面打印的提示日志去操作~")
def check_and_download_chrome_ahead(self):
"""
尝试预先下载解压缩chrome的driver和便携版
主要用于处理多进程模式下,可能多个进程同时尝试该操作导致的问题
:return:
"""
logger.info("检查chrome相关内容是否ok")
if is_windows():
self.check_and_download_chrome_ahead_windows()
else:
self.check_and_download_chrome_ahead_linux()
def check_and_download_chrome_ahead_windows(self):
logger.info(
color("bold_yellow")
+ "如果自动下载失败,可能是网络问题,请根据提示下载的内容,自行去备用网盘下载该内容到utils目录下 https://docs.qq.com/doc/DYmdpaUthQnp4Rnpy"
)
chrome_driver_exe_name = os.path.basename(self.chrome_driver_executable_path())
zip_name = os.path.basename(self.chrome_binary_7z())
chrome_root_directory = self.chrome_root_directory()
logger.info("检查driver是否存在")
if not self.is_valid_chrome_file(self.chrome_driver_executable_path()):
logger.info(color("bold_yellow") + f"未在小助手utils目录里发现 {chrome_driver_exe_name} ,将尝试从网盘下载")
logger.info(
color("bold_cyan") + f"如果速度实在太慢,可以去QQ群文件里面下载 {chrome_driver_exe_name},然后原样放到小助手的 utils 目录中,再重新启动即可"
)
self.download_chrome_driver(chrome_driver_exe_name)
options = self.new_options()
options.add_argument("--headless")
options.add_experimental_option("excludeSwitches", ["enable-logging"])
if not self.cfg.force_use_portable_chrome:
try:
logger.info(
color("bold_green")
+ "检查系统自带的chrome是否可用,如果一直卡在这里,请试试打开【配置工具/公共配置/登录/强制使用便携版chrome】开关后,再次运行~。如果安装了【360浏览器/qq浏览器】,可以先试试把chrome修改为默认浏览器,或者卸载掉他们,然后重启电脑后再运行小助手试试。"
)
# note: 调用chrome_driver创建新session时,chrome_driver会尝试添加en-us的键盘布局-。-这个目前没法修,因为必须依赖这个
self.driver = webdriver.Chrome(
service=Service(executable_path=self.chrome_driver_executable_path()), options=options
)
self.driver.quit()
return
except Exception:
logger.info("走到这里说明系统自带的chrome不可用")
pass
else:
logger.info("当前配置为强制使用便携版chrome")
# 尝试从网盘下载合适版本的便携版chrome
if not self.is_valid_chrome_file(self.chrome_binary_7z()):
logger.info(
color("bold_yellow") + f"未在小助手utils目录里发现 便携版chrome 的压缩包,尝试自动从网盘下载 {zip_name},需要下载大概80MB的压缩包,请耐心等候"
)
logger.info(color("bold_cyan") + f"如果速度实在太慢,可以去QQ群文件里面下载 {zip_name},然后原样放到小助手的 utils 目录中,再重新启动即可")
self.download_chrome_file(zip_name)
# 尝试解压
if not os.path.isdir(self.chrome_binary_directory()):
logger.info("自动解压便携版chrome到当前目录")
decompress_dir_with_bandizip(self.chrome_binary_7z(), dst_parent_folder=chrome_root_directory)
logger.info("检查便携版chrome是否有效")
try:
options.binary_location = self.chrome_binary_location()
# you may need some other options
options.add_argument("--no-sandbox")
options.add_argument("--no-default-browser-check")
options.add_argument("--no-first-run")
self.driver = webdriver.Chrome(
service=Service(executable_path=self.chrome_driver_executable_path()), options=options
)
self.driver.quit()
return
except Exception:
pass
# 走到这里,大概率是多线程并行下载导致文件出错了,尝试重新下载
logger.info(color("bold_yellow") + "似乎chrome相关文件损坏了,尝试重新下载并解压")
logger.info(
color("bold_cyan")
+ f"如果速度实在太慢,可以去QQ群文件里面下载 {zip_name}{chrome_driver_exe_name},然后原样放到小助手的 utils 目录中,再重新启动即可"
)
self.download_chrome_driver(chrome_driver_exe_name)
self.download_chrome_file(zip_name)
shutil.rmtree(self.chrome_binary_directory(), ignore_errors=True)
decompress_dir_with_bandizip(self.chrome_binary_7z(), dst_parent_folder=chrome_root_directory)
def check_and_download_chrome_ahead_linux(self):
ok = True
if not os.path.exists(self.chrome_binary_location_linux()):
ok = False
logger.info(
color("bold_red")
+ (
f"未在发现chrome,请按照下面这个推文的步骤去下载安装最新稳定版本chrome到该位置\n"
f"预期位置: {self.chrome_binary_location_linux()}\n"
f"安装教程: https://blog.csdn.net/weixin_42649856/article/details/103275162 \n"
)
)
if not os.path.exists(self.chrome_driver_executable_path_linux()):
ok = False
logger.info(
color("bold_red")
+ (
f"未在发现chromedriver,请按照下面这个推文的步骤去下载安装最新稳定版本chromedriver到该位置(需要确保与安装的chrome版本匹配)\n"
f"预期位置: {self.chrome_driver_executable_path_linux()}\n"
f"安装教程: https://blog.csdn.net/weixin_42649856/article/details/103275162 \n"
)
)
if not ok:
logger.info(
color("bold_yellow")
+ (
"当前运行在非windows的环境,检测到chrome和对应driver未全部安装,请按照上述提示完成安装后重新运行~\n"
"或者根据你的系统直接使用或参考以下脚本之一来完成一键下载安装\n"
"1. _ubuntu_download_and_install_chrome_and_driver.sh \n"
"2. _centos_download_and_install_chrome_and_driver.sh \n"
)
)
pause_and_exit(-1)
def download_chrome_file(self, filename: str) -> str:
def download_by_github() -> str:
logger.warning("尝试通过github下载")
return download_latest_github_release(
"utils",
filename,
"fzls",
"djc_helper_chrome",
)
def download_by_alist() -> str:
logger.warning("尝试通过alist下载")
return download_from_alist(self.get_path_in_netdisk(filename), self.chrome_root_directory())
download_functions = [
download_by_github,
download_by_alist,
]
for download_function in download_functions:
try:
return download_function()
except Exception:
logger.info("下载失败了,尝试下一个下载方式")
raise Exception("所有下载方式都失败了")
def download_chrome_driver(self, chrome_driver_exe_name: str) -> str:
try:
return download_chrome_driver(self.chrome_driver_version, "utils", ".")
except Exception as e:
logger.error("从chrome官网下载driver失败,尝试从网盘下载", exc_info=e)
return self.download_chrome_file(chrome_driver_exe_name)
def download_from_github(self, filename: str) -> str:
return download_latest_github_release(
"utils",
filename,
"fzls",
"djc_helper_chrome",
)
def get_path_in_netdisk(self, filename: str) -> str:
return f"/文本编辑器、chrome浏览器、autojs、HttpCanary等小工具/{filename}"
def chrome_driver_executable_path(self):
# re: 这里chromedriver可以使用另一个库来自动为维护,后面可以看看有没有必要调整
# https://pypi.org/project/webdriver-manager/
return os.path.realpath(f"{self.chrome_root_directory()}/chromedriver_{self.get_chrome_major_version()}.exe")
def chrome_binary_7z(self):
return os.path.realpath(f"{self.chrome_root_directory()}/chrome_portable_{self.get_chrome_major_version()}.7z")
def chrome_binary_directory(self):
return os.path.realpath(f"{self.chrome_root_directory()}/chrome_portable_{self.get_chrome_major_version()}")
def chrome_binary_location(self):
return os.path.realpath(
f"{self.chrome_root_directory()}/chrome_portable_{self.get_chrome_major_version()}/chrome.exe"
)
def chrome_installer_name(self):
return f"Chrome_{self.get_chrome_major_version()}.(小版本号)_普通安装包_非便携版.exe"
def chrome_driver_executable_path_linux(self):
return "/usr/local/bin/chromedriver"
def chrome_binary_location_linux(self):
return "/usr/bin/google-chrome"
def chrome_root_directory(self):
return os.path.realpath("./utils")
def is_valid_chrome_file(self, chrome_filepath) -> bool:
if not os.path.isfile(chrome_filepath):
# 文件不存在
return False
invalid_filesize = 1 * MiB
if get_file_or_directory_size(chrome_filepath) <= invalid_filesize:
# 文件大小太小了,很有可能是没下载完全,或者下载报错了
return False
return True
def get_chrome_major_version(self) -> int:
version = self.chrome_major_version
if self.cfg is None or self.cfg.force_use_chrome_major_version == 0:
rule = config_cloud().chrome_version_replace_rule
if self.chrome_major_version in rule.troublesome_major_version_list and rule.valid_chrome_version != 0:
# 当前chrome版本在部分系统可能无法正常使用,替换为远程配置的可用版本
version = rule.valid_chrome_version
else:
# 使用默认的版本
version = self.chrome_major_version
else:
# 使用本地配置的版本
version = self.cfg.force_use_chrome_major_version
return version
def login(self, account, password, login_mode: str, name=""):
"""
自动登录指定账号,并返回登陆后的cookie中包含的uin、skey数据
:param account: 账号
:param password: 密码
:rtype: LoginResult
"""
self.name = name
self.account = account
self.password = password
self.window_title = f"将登录 {name}({account}) - {login_mode}"
logger.info(f"{name} 即将开始自动登录,无需任何手动操作,等待其完成即可")
logger.info(f"{name} 如果出现报错,可以尝试调高相关超时时间然后重新执行脚本")
def login_with_account_and_password():
logger.info(color("bold_green") + f"{name} 当前为自动登录模式,请不要手动操作网页,否则可能会导致登录流程失败")
# 等待页面加载
self.wait_for_login_page_loaded()
logger.info("由于账号密码登录有可能会触发短信验证,因此优先尝试点击头像来登录~")
login_by_click_avatar_success = self.try_auto_click_avatar(account, name, self.login_type_auto_login)
if login_by_click_avatar_success:
logger.info("使用头像点击登录成功")
else:
logger.warning("点击头像登录失败,尝试输入账号密码来进行登录")
# 选择密码登录
self.driver.find_element(By.ID, "switcher_plogin").click()
# 输入账号
self.driver.find_element(By.ID, "u").clear()
self.driver.find_element(By.ID, "u").send_keys(account)
# 输入密码
self.driver.find_element(By.ID, "p").clear()
self.driver.find_element(By.ID, "p").send_keys(password)
logger.info(f"{name} 等待一会,确保登录键可以点击")
time.sleep(3)
# 发送登录请求
self.driver.find_element(By.ID, "login_button").click()
# 尝试自动处理验证码
self.try_auto_resolve_captcha()
return self._login(
self.login_type_auto_login, login_action_fn=login_with_account_and_password, login_mode=login_mode
)
def qr_login(self, login_mode: str, name="", account=""):
"""
二维码登录,并返回登陆后的cookie中包含的uin、skey数据
:rtype: LoginResult
"""
logger.info("即将开始扫码登录,请在弹出的网页中扫码登录~")
self.name = name
self.account = ""
self.password = ""
self.window_title = f"请扫码 {name} - {login_mode}"
def replace_qr_code_tip():
qr_js_wait_time = 1
try:
# 扫码登录
tip_class_name = "qr_safe_tips"
tip = f"请扫码 {name}"
logger.info(color("bold_green") + f"准备修改二维码上方 扫码提示文字 为 {tip}")
WebDriverWait(self.driver, qr_js_wait_time).until(
expected_conditions.visibility_of_element_located((By.CLASS_NAME, tip_class_name))
)
self.driver.execute_script(
f"document.getElementsByClassName('{tip_class_name}')[0].innerText = '{tip}'; "
)
except Exception as e:
logger.warning("替换扫码提示文字出错了(不影响登录流程)")
logger.debug("", exc_info=e)
try:
# 提示点击头像登录
tip_id = "qlogin_tips"
tip = f"请点击头像授权登录 {name} - 多于两个账号可以点击两侧箭头切换"
logger.info(color("bold_green") + f"准备修改二维码上方 点击头像提示文字 为 {tip}")
WebDriverWait(self.driver, qr_js_wait_time).until(
expected_conditions.visibility_of_element_located((By.ID, tip_id))
)
self.driver.execute_script(f"document.getElementById('{tip_id}').innerText = '{tip}'; ")
except Exception as e:
logger.warning("替换扫码提示文字出错了(不影响登录流程)")
logger.debug("", exc_info=e)
try:
# 调整箭头
logger.info(color("bold_green") + "准备修改两侧箭头为可见的 ⬅️和 ➡️")
self.driver.execute_script(
"""
function setArrow(elementId = "", arrow="") {
let target = document.getElementById(elementId)
// 需要修改 display 为 block 才能获取高度
let oldDisplay = target.style.display
target.style.display = "block"
let desiredHeight = parseInt((target.clientHeight || 120) * 1.5)
target.style.display = oldDisplay
// 修改为位于头像左右的箭头
target.innerText = arrow
target.style.lineHeight = desiredHeight + "px"
}
setArrow("prePage", "⬅️")
setArrow("nextPage", "➡️")
"""
)
except Exception as e:
logger.warning("修改箭头失败了(不影响登录流程)")
logger.debug("", exc_info=e)
def login_with_qr_code():
logger.info(color("bold_yellow") + f"{name} 当前为扫码登录模式,请在{self.get_login_timeout(True)}s内完成扫码登录操作或快捷登录操作")
self.wait_for_login_page_loaded()
replace_qr_code_tip()
logger.info(color("bold_green") + f"{name} 尝试自动点击头像进行登录")
self.try_auto_click_avatar(account, name, self.login_type_qr_login)
return self._login(self.login_type_qr_login, login_action_fn=login_with_qr_code, login_mode=login_mode)
def wait_for_login_page_loaded(self):
logger.info(f"{self.name} 等待页面加载")
time.sleep(self.cfg.login.open_url_wait_time)
if self.login_type == self.login_type_auto_login:
# 仅自动登录模式需要检测窗口是否已经弹出来
logger.info(f"{self.name} 等待 登录框#switcher_plogin 加载完毕")
WebDriverWait(self.driver, self.cfg.login.load_login_iframe_timeout).until(
expected_conditions.visibility_of_element_located((By.ID, "switcher_plogin"))
)
def try_auto_click_avatar(self, account: str, name: str, login_type: str) -> bool:
# 检测功能开关
if login_type == self.login_type_auto_login:
enable = self.cfg.login.enable_auto_click_avatar_in_auto_login
else:
enable = self.cfg.login.enable_auto_click_avatar_in_qr_login
if not enable:
logger.warning(f"当前未开启【{login_type} 模式下尝试点击头像来登录】,请自行操作~。若需要该功能,可在配置工具【公共配置/登录】中开启本功能")
return False
logger.info(f"当前已开启【{login_type} 模式下尝试点击头像来登录】。如该功能有异常,导致登录流程无法正常进行,可在配置工具【公共配置/登录】中关闭本功能")
# 实际登录流程
login_success = False
ctx = f"【{name}({account})】"
try:
# 尝试自动点击头像登录
if account != "":
selector = f"#qlogin_list > a[uin='{account}']"
logger.info(color("bold_green") + f"{ctx} 尝试点击头像来登录")
logger.info("检查对应头像是否存在")
time.sleep(1)
self.driver.find_element(By.CSS_SELECTOR, selector)
logger.info("开始点击对应头像")
self.driver.execute_script(
f"""
document.querySelector("{selector}").click()
"""
)
# 由于有时候这个登录框可能会迟一点消失,这里改为多次尝试,减小误判的情况
logger.info(f"判断点击头像登录是否成功,最大等待 {self.cfg.login.login_by_click_avatar_finished_timeout} 秒")
login_success = False
try:
WebDriverWait(self.driver, self.cfg.login.login_by_click_avatar_finished_timeout).until(
expected_conditions.invisibility_of_element((By.ID, "switcher_plogin"))
)
login_success = True
except InvalidArgumentException as e:
# 如果已经刷新到新页面,则会报这个异常,说明也是成功了
logger.info(color("bold_yellow") + f"{ctx} 跳转到新的页面了,导致无法定位到登录按钮,这说明登录也成功了")
login_success = True
logger.debug("保存下异常信息", exc_info=e)
except Exception as e:
login_success = False
logger.debug("头像登录出错了", exc_info=e)
logger.info(color("bold_cyan") + f"{ctx} 点击头像登录的结果为: {'成功' if login_success else '失败'}")
elif login_type == self.login_type_qr_login:
async_message_box("现已支持扫码模式下自动点击头像进行登录,不过需要填写QQ号码,可使用配置工具填写QQ号码即可体验本功能", "扫码自动点击头像功能提示", show_once=True)
except Exception as e:
logger.warning(f"{ctx} 尝试自动点击头像登录失败了,请自行操作~")
logger.debug("", exc_info=e)
return login_success
def _login(self, login_type, login_action_fn=None, login_mode="normal"):
if not is_first_run_in(f"login_locker_{login_mode}_{self.name}", duration=datetime.timedelta(seconds=10)):
raise SameAccountTryLoginAtMultipleThreadsException
login_retry_key = "login_retry_key"
login_retry_data, retry_timeouts = self.get_retry_data(
login_retry_key, self.cfg.login.max_retry_count - 1, self.cfg.login.retry_wait_time
)
self.login_slow_retry_max_count = self.cfg.login.max_retry_count
for idx in range_from_one(self.login_slow_retry_max_count):
self.login_slow_retry_index = idx
logger.info(
color("bold_green") + f"[慢速重试阶段] [{idx}/{self.login_slow_retry_max_count}] {self.name} 开始本轮登录流程"
)
self.login_type = login_type
self.login_mode = login_mode
# note: 如果get_login_url的surl变更,代码中确认登录完成的地方也要一起改
login_fn, suffix, login_url = {
self.login_mode_normal: (
self._login_real,
"",
self.get_login_url(21000127, 8, "https://dnf.qq.com/"),
),
self.login_mode_xinyue: (
self._login_xinyue,
"心悦",
get_act_url("DNF地下城与勇士心悦特权专区"),
),
self.login_mode_qzone: (
self._login_qzone,
"QQ空间业务(如抽卡等需要用到)(不启用QQ空间系活动就不会触发本类型的登录,完整列表参见示例配置)",
self.get_login_url(15000103, 5, "https://act.qzone.qq.com/"),
),
self.login_mode_guanjia: (
self._login_guanjia,
"电脑管家(如电脑管家蚊子腿需要用到,完整列表参见示例配置)",
get_act_url("管家蚊子腿"),
),
self.login_mode_wegame: (
self._login_wegame,
"wegame(获取wegame相关api需要用到)",
self.get_login_url(1600001063, 733, "https://www.wegame.com.cn/"),
),
self.login_mode_club_vip: (
self._login_club_vip,
"club.vip.qq.com",
self.get_login_url(8000212, 18, "https://club.vip.qq.com/qqvip/acts2021/dnf"),
),
self.login_mode_iwan: (
self._login_iwan,
"爱玩",
"https://iwan.qq.com/g/gift",
),
self.login_mode_supercore: (
self._login_supercore,
"超享玩",
"https://act.supercore.qq.com/supercore/act/ac2cb66d798da4d71bd33c7a2ec1a7efb/index.html",
),
self.login_mode_djc: (
self._login_djc,
"道聚城",
get_act_url("道聚城"),
),
}[login_mode]
ctx = f"{login_type}-{suffix}"
login_exception = None
try:
if idx > 1:
logger.info(color("bold_cyan") + f"已经是第{idx}次登陆,说明可能出现某些问题,将关闭隐藏浏览器选项,方便观察出现什么问题")
self.cfg.run_in_headless_mode = False
self.prepare_chrome(ctx, login_type, login_url)
lr = login_fn(ctx, login_action_fn=login_action_fn)
logger.debug(f"{self.name} 登录结果为 {lr}")
return lr
except Exception as e:
login_exception = e
finally:
login_result = color("bold_green") + "登录成功"
if login_exception is not None:
login_result = color("bold_cyan") + "登录失败"
current_url = ""
if self.driver is not None:
current_url = self.driver.current_url
used_time = datetime.datetime.now() - self.time_start_login
logger.info("")
logger.info(
f"[{login_result}] "
+ color("bold_yellow")
+ f"{self.name} [慢速重试阶段] 第{idx}/{self.login_slow_retry_max_count}{ctx} 共耗时为 {used_time}"
)
logger.info("")
self.destroy_chrome()
if login_exception is not None:
# 登陆失败
msg = (
f"[慢速重试阶段] {self.name}{self.login_slow_retry_index}/{self.login_slow_retry_max_count}次尝试登录出错"
)
if idx < self.login_slow_retry_max_count:
# 每次等待时长线性递增
wait_time = retry_timeouts[idx - 1]
msg += f"将等待较长一段时间后再重试,也就是 {wait_time:.2f}秒后重试(v{now_version} {ver_time})"
msg += f"\n\t当前登录重试等待时间序列:{retry_timeouts}"
msg += f"\n\t根据历史数据得出的推荐重试等待时间:{login_retry_data.recommended_first_retry_timeout}"
if use_by_myself():
msg += f"\n\t(仅我可见)历史重试成功等待时间列表:{login_retry_data.history_success_timeouts}"
msg += f"\n\t当前网址为 {current_url}"
logger.exception(msg, exc_info=login_exception)
if type(login_exception) is RequireVerifyMessageButInHeadlessMode:
logger.info(color("bold_yellow") + f"检测到需要手动验证流程({login_exception}),将立即开始第二次慢速重试,且显示浏览器界面")
wait_time = 1
count_down(f"{truncate(self.name, 20):20s} 重试", wait_time)
else:
logger.exception(msg, exc_info=login_exception)
else:
# 登陆成功
if idx > 1:
# 第idx-1次的重试成功了,尝试更新历史数据
self.update_retry_data(
login_retry_key,
retry_timeouts[idx - 2],
self.cfg.login.recommended_retry_wait_time_change_rate,
self.name,
)
# 能走到这里说明登录失败了,大概率是网络不行
logger.warning(
color("bold_yellow")
+ (
f"已经尝试登录 {self.name} {self.cfg.login.max_retry_count}次,均已失败,大概率是网络有问题(v{now_version})\n"
"建议依次尝试下列措施\n"
"1. 重新打开程序\n"
"2. 重启电脑\n"
"3. 切换旧版本chrome(如果之前都是正常的) - 配置工具/公共配置/登录/chrome版本,修改为94或者更早的版本,并开启 强制使用便携版 开关\n"
"4. 更换dns,如谷歌、阿里、腾讯、百度的dns,具体更换方法请百度\n"
"5. 重装网卡驱动\n"
"6. 换个网络环境\n"
"7. 换台电脑\n"
)
)
if login_mode == self.login_mode_guanjia:
logger.warning(color("bold_cyan") + "如果一直卡在管家登录流程,可能是你网不行,建议多试几次,真不行就去配置工具关闭管家活动的开关(不是关闭这个登录页面)~")
if is_run_in_github_action():
# github action 环境下特殊处理
raise GithubActionLoginException()
raise Exception(
"网络很有可能有问题(备注:访问其他网页没问题不代表访问这个网页也没问题-。-)\n" "如果是chrome版本更新后才这样,可以尝试在配置工具中设置强制使用便携版chrome,并指定chrome的版本号,如89"
)
def _login_real(self, login_type, login_action_fn=None):
"""
通用登录逻辑,并返回登陆后的cookie中包含的uin、skey数据
:rtype: LoginResult
"""
s_url = "https://dnf.qq.com/"
def switch_to_login_frame_fn():
if self.need_reopen_url(login_type):
self.get_switch_to_login_frame_fn(21000127, 8, s_url)
def assert_login_finished_fn():
logger.info(f"{self.name} 请等待网页切换为目标网页,则说明已经登录完成了,最大等待时长为{self.cfg.login.login_finished_timeout}")
WebDriverWait(self.driver, self.cfg.login.login_finished_timeout).until(
expected_conditions.url_contains(s_url)
)
self._login_common(login_type, switch_to_login_frame_fn, assert_login_finished_fn, login_action_fn)
# 从cookie中获取uin和skey
return LoginResult(
uin=self.get_cookie("uin"),
skey=self.get_cookie("skey"),
p_skey=self.get_cookie("p_skey"),
vuserid=self.get_cookie("vuserid"),
apps_p_skey=self.get_cookie("apps_p_skey"),
)
def _login_qzone(self, login_type, login_action_fn=None):
"""
通用登录逻辑,并返回登陆后的cookie中包含的uin、skey数据
:rtype: LoginResult
"""
s_url = "https://act.qzone.qq.com/"
def switch_to_login_frame_fn():
if self.need_reopen_url(login_type):
self.get_switch_to_login_frame_fn(15000103, 5, s_url)
def assert_login_finished_fn():
logger.info(f"{self.name} 请等待网页切换为目标网页,则说明已经登录完成了,最大等待时长为{self.cfg.login.login_finished_timeout}")
WebDriverWait(self.driver, self.cfg.login.login_finished_timeout).until(
expected_conditions.url_contains(s_url)
)
self._login_common(login_type, switch_to_login_frame_fn, assert_login_finished_fn, login_action_fn)
# 从cookie中获取uin和skey
return LoginResult(
p_skey=self.get_cookie("p_skey"),
uin=self.get_cookie("uin"),
skey=self.get_cookie("skey"),
vuserid=self.get_cookie("vuserid"),
apps_p_skey=self.get_cookie("apps_p_skey"),
)
def _login_club_vip(self, login_type, login_action_fn=None):
"""
通用登录逻辑,并返回登陆后的cookie中包含的uin、skey数据
:rtype: LoginResult
"""
s_url = "https://club.vip.qq.com/qqvip/acts2021/dnf"
def switch_to_login_frame_fn():
if self.need_reopen_url(login_type):
self.get_switch_to_login_frame_fn(8000212, 18, s_url)
def assert_login_finished_fn():
logger.info(f"{self.name} 请等待网页切换为目标网页,则说明已经登录完成了,最大等待时长为{self.cfg.login.login_finished_timeout}")
WebDriverWait(self.driver, self.cfg.login.login_finished_timeout).until(
expected_conditions.url_contains(s_url)
)
self._login_common(login_type, switch_to_login_frame_fn, assert_login_finished_fn, login_action_fn)
# 从cookie中获取uin和skey
return LoginResult(
uin=self.get_cookie("uin"),
skey=self.get_cookie("skey"),
p_skey=self.get_cookie("p_skey"),
)
def _login_iwan(self, login_type, login_action_fn=None):
"""
通用登录逻辑,并返回登陆后的cookie中包含的uin、skey数据
:rtype: LoginResult
"""
def switch_to_login_frame_fn():
if self.need_reopen_url(login_type):
logger.info("打开活动界面")
self.open_url_on_start("https://iwan.qq.com/g/gift")
self.set_window_size()
try:
logger.info("尝试处理可能有的每日签到弹窗")
xpath_close_daily_sign = "//i[contains(@class, 'athena-dialog-close-icon')]"
WebDriverWait(self.driver, self.cfg.login.open_url_wait_time).until(
expected_conditions.element_to_be_clickable((By.XPATH, xpath_close_daily_sign))
)
logger.info("等待3秒,确保加载完成")
time.sleep(3)
logger.info("点击关闭每日签到弹窗")
self.driver.find_element(By.XPATH, xpath_close_daily_sign).click()
time.sleep(3)
except Exception:
logger.warning("爱玩处理 关闭签到弹窗 流程失败了,可能是本次没有弹窗")
try:
logger.info("等待登录按钮出来,确保加载完成")
xpath_login = "//div[contains(text(), '未登录')]"
WebDriverWait(self.driver, self.cfg.login.load_page_timeout).until(
expected_conditions.element_to_be_clickable((By.XPATH, xpath_login))
)
logger.info("等待3秒,确保加载完成")
time.sleep(3)
logger.info("点击登录按钮")
self.driver.find_element(By.XPATH, xpath_login).click()
time.sleep(3)
except Exception:
logger.warning("爱玩处理 点击登录按钮 流程失败了,可能是已经处理成功了")
try:
logger.info("勾选同意协议")
self.driver.find_element(By.CLASS_NAME, "js-check-input").click()
time.sleep(2)
logger.info("点击使用QQ登录")
self.driver.find_element(By.CLASS_NAME, "iwanLogin_qq").click()
except Exception:
logger.warning("爱玩处理 中间同意协议 流程失败了,可能是已经处理成功了")
try:
logger.info("等待#ptlogin_iframe加载完毕并切换")
WebDriverWait(self.driver, self.cfg.login.load_login_iframe_timeout).until(
expected_conditions.visibility_of_element_located((By.ID, "ptlogin_iframe"))
)
ptlogin_iframe = self.driver.find_element(By.ID, "ptlogin_iframe")
self.driver.switch_to.frame(ptlogin_iframe)
except Exception:
logger.warning("爱玩处理 切换登录iframe 流程失败了,可能是已经处理成功了")
def assert_login_finished_fn():
logger.info("等待 切换账号 按钮出来,确保加载完成")
xpath_switch_login = "//span[contains(text(), '切换账号')]"
WebDriverWait(self.driver, self.cfg.login.load_page_timeout).until(
expected_conditions.visibility_of_element_located((By.XPATH, xpath_switch_login))
)
self._login_common(login_type, switch_to_login_frame_fn, assert_login_finished_fn, login_action_fn)
# 从cookie中获取openid
return LoginResult(
iwan_openid=self.get_cookie("vqq_openid"),
iwan_access_token=self.get_cookie("vqq_access_token"),
)
def _login_supercore(self, login_type, login_action_fn=None):
"""
登录超享玩活动页面,获取需要的一些参数
:rtype: LoginResult
"""
s_url = "https://act.supercore.qq.com/supercore/act/ac2cb66d798da4d71bd33c7a2ec1a7efb/index.html"
def switch_to_login_frame_fn():
if self.need_reopen_url(login_type):
logger.info("打开活动界面")
self.open_url_on_start(
"https://act.supercore.qq.com/supercore/act/ac2cb66d798da4d71bd33c7a2ec1a7efb/index.html"
)
# 这里大小与设备信息中的一致
size = self.mobile_emulation_qq["deviceMetrics"]
self.set_window_size(size["width"], size["height"])
logger.info("等待登录按钮出来,确保加载完成")
id_login = "ptLoginBtn"
WebDriverWait(self.driver, self.cfg.login.load_page_timeout).until(
expected_conditions.element_to_be_clickable((By.ID, id_login))
)
logger.info("等待3秒,确保加载完成")
time.sleep(3)
logger.info("点击登录按钮")
self.driver.find_element(By.ID, id_login).click()
time.sleep(3)
def assert_login_finished_fn():
logger.info(f"{self.name} 请等待网页切换为目标网页,则说明已经登录完成了,最大等待时长为{self.cfg.login.login_finished_timeout}")
WebDriverWait(self.driver, self.cfg.login.login_finished_timeout).until(
expected_conditions.url_contains(s_url)
)
# 超享玩这个页面必须使用手机QQ的user-agent才能登录,而这种情况下出现的登录框与以往的样式不同,且不能调整,因此这里需要自定义下
def login_with_account_and_password():
if self.account != "":
logger.info(color("bold_green") + f"{self.name} 开始账号密码登录")
# 输入账号
self.driver.find_element(By.ID, "u").clear()
self.driver.find_element(By.ID, "u").send_keys(self.account)
# 输入密码
self.driver.find_element(By.ID, "p").clear()
self.driver.find_element(By.ID, "p").send_keys(self.password)
logger.info(f"{self.name} 等待一会,确保登录键可以点击")
time.sleep(3)
# 发送登录请求
self.driver.find_element(By.ID, "go").click()
# 尝试自动处理验证码
self.try_auto_resolve_captcha()
else:
logger.info(color("bold_yellow") + f"{self.name} 当前设置的是扫码登录,但是这个 超享玩 活动只能输入账号密码登录,请手动完成登录")
@try_except()
def try_click_login_after_filled_in():
start_time = time.time()
max_wait_seconds = self.get_login_timeout(True)
last_qq = ""
last_password = ""
while True:
qq = self.driver.find_element(By.ID, "u").get_attribute("value")
password = self.driver.find_element(By.ID, "p").get_attribute("value")
if qq != "" and password != "" and qq == last_qq and password == last_password:
logger.info("检测到qq没有发生变化,尝试帮点一下登录")
self.driver.find_element(By.ID, "go").click()
break
# 如果超时了,也直接跳出
if time.time() - start_time >= max_wait_seconds:
logger.warning(f"已超过最大等待时长 {max_wait_seconds}秒,将不再尝试帮忙点击登录按钮")
break
last_qq = qq
last_password = password
time.sleep(2)
# 因为有时候这个不能点登录按钮,这里尝试帮忙点一下
try_click_login_after_filled_in()
self._login_common(
login_type, switch_to_login_frame_fn, assert_login_finished_fn, login_with_account_and_password
)
# 从cookie中获取openid
return LoginResult(
common_openid=self.get_cookie("openid"),
common_access_token=self.get_cookie("access_token"),
)
def _login_guanjia(self, login_type, login_action_fn=None):
"""
通用登录逻辑,并返回登陆后的cookie中包含的uin、skey数据
:rtype: LoginResult
"""
def switch_to_login_frame_fn():
if self.need_reopen_url(login_type):
logger.info("打开活动界面")
self.open_url_on_start(get_act_url("管家蚊子腿"))
self.set_window_size()
logger.info("等待登录按钮#dologin出来,确保加载完成")
WebDriverWait(self.driver, self.cfg.login.load_page_timeout).until(
expected_conditions.text_to_be_present_in_element((By.ID, "layer72"), "【登录】")
)
logger.info("等待5秒,确保加载完成")
time.sleep(5)
logger.info("点击登录按钮")
self.driver.find_element(By.ID, "layer72").click()
logger.info("等待2秒,确保#login_ifr显示出来并切换")
time.sleep(2)
loginIframe = list(
iframe
for iframe in self.driver.find_elements(by=By.TAG_NAME, value="iframe")
if "https://graph.qq.com/oauth2.0/authorize" in iframe.get_property("src")
)[0]
self.driver.switch_to.frame(loginIframe)
logger.info("等待#login_ifr#ptlogin_iframe加载完毕并切换")
WebDriverWait(self.driver, self.cfg.login.load_login_iframe_timeout).until(
expected_conditions.visibility_of_element_located((By.ID, "ptlogin_iframe"))
)
ptlogin_iframe = self.driver.find_element(By.ID, "ptlogin_iframe")
self.driver.switch_to.frame(ptlogin_iframe)
def assert_login_finished_fn():
logger.info(f"{self.name} 请等待#logined的div可见,则说明已经登录完成了,最大等待时长为{self.cfg.login.login_finished_timeout}")
WebDriverWait(self.driver, self.cfg.login.login_finished_timeout).until(
expected_conditions.text_to_be_present_in_element((By.ID, "layer72"), "【注销】")
)
self._login_common(login_type, switch_to_login_frame_fn, assert_login_finished_fn, login_action_fn)
# {"province":"","city":"","year":"19XX","openid":"XXXXXX","sex":1,"nickname":"XXXXXX","headimgurl":"XXXXXX","key":"XXXXXX"}
cookie_userinfo = unquote_plus(self.get_cookie("uInfo101478239"))
raw_userinfo = json.loads(cookie_userinfo)
user_info = GuanJiaUserInfo().auto_update_config(raw_userinfo)
# 从cookie中获取uin和skey
return LoginResult(
qc_openid=user_info.openid,
qc_k=user_info.key,
qc_access_token=user_info.key,
qc_nickname=user_info.nickname,
uin=self.get_cookie("uin"),
skey=self.get_cookie("skey"),
p_skey=self.get_cookie("p_skey"),
vuserid=self.get_cookie("vuserid"),
)
def _login_wegame(self, login_type, login_action_fn=None):
"""
通用登录逻辑,并返回登陆后的cookie中包含的uin、skey数据
:rtype: LoginResult
"""
s_url = "https://www.wegame.com.cn/"
def switch_to_login_frame_fn():
if self.need_reopen_url(login_type):
self.get_switch_to_login_frame_fn(1600001063, 733, s_url)
def assert_login_finished_fn():
logger.info(f"{self.name} 请等待网页切换为目标网页,则说明已经登录完成了,最大等待时长为{self.cfg.login.login_finished_timeout}")
WebDriverWait(self.driver, self.cfg.login.login_finished_timeout).until(
expected_conditions.url_contains(s_url)
)
self._login_common(login_type, switch_to_login_frame_fn, assert_login_finished_fn, login_action_fn)
# 从cookie中获取uin和skey
return LoginResult(uin=self.get_cookie("uin"), skey=self.get_cookie("skey"), p_skey=self.get_cookie("p_skey"))
def _login_xinyue(self, login_type, login_action_fn=None):
"""
通用登录逻辑,并返回登陆后的cookie中包含的uin、skey数据
:rtype: LoginResult
"""
def switch_to_login_frame_fn():
if self.need_reopen_url(login_type):
logger.info("打开活动界面")
self.open_url_on_start(get_act_url("DNF地下城与勇士心悦特权专区"))
self.set_window_size()
logger.info("等待#loginframe加载完毕")
WebDriverWait(self.driver, self.cfg.login.load_login_iframe_timeout).until(
expected_conditions.visibility_of_element_located((By.CLASS_NAME, "loginframe"))
)
login_frame = self.driver.find_element(By.CLASS_NAME, "loginframe")
self.driver.switch_to.frame(login_frame)
logger.info("等待#loginframe#ptlogin_iframe加载完毕并切换")
WebDriverWait(self.driver, self.cfg.login.load_login_iframe_timeout).until(
expected_conditions.visibility_of_element_located((By.ID, "ptlogin_iframe"))
)
ptlogin_iframe = self.driver.find_element(By.ID, "ptlogin_iframe")
self.driver.switch_to.frame(ptlogin_iframe)
def assert_login_finished_fn():
logger.info(f"{self.name} 请等待#login-box不可见,则说明已经登录完成了,最大等待时长为{self.cfg.login.login_finished_timeout}")
WebDriverWait(self.driver, self.cfg.login.login_finished_timeout).until(
expected_conditions.invisibility_of_element_located((By.ID, "login-box"))
)
logger.info("等待1s,确认获取openid的请求完成")
time.sleep(1)
# 确保openid已设置
for t in range(1, 3 + 1):
if self.driver.get_cookie("openid") is None:
logger.info(f"第{t}/3未在心悦的cookie中找到openid,等一秒再试")
time.sleep(1)
continue
break
self._login_common(login_type, switch_to_login_frame_fn, assert_login_finished_fn, login_action_fn)
# 从cookie中获取openid
return LoginResult(
openid=self.get_cookie("openid"),
xinyue_openid=self.get_cookie("xinyue_openid") or self.get_cookie("openid"),
xinyue_access_token=self.get_cookie("xinyue_access_token") or self.get_cookie("access_token"),
)
def _login_djc(self, login_type, login_action_fn=None):
"""
通用登录逻辑,并返回登陆后的cookie中包含的uin、skey数据
:rtype: LoginResult
"""
def switch_to_login_frame_fn():
if self.need_reopen_url(login_type):
logger.info("打开活动界面")
self.open_url_on_start(get_act_url("道聚城"))
self.set_window_size()
logger.info("等待登录按钮#unlogin出来,确保加载完成")
WebDriverWait(self.driver, self.cfg.login.load_page_timeout).until(
expected_conditions.visibility_of_element_located((By.CSS_SELECTOR, "#unlogin"))
)
logger.info("等待5秒,确保加载完成")
time.sleep(5)
logger.info("点击登录按钮")
self.driver.find_element(By.CSS_SELECTOR, "#unlogin > a").click()
logger.info("等待2秒,确保#loginframe显示出来并切换")
time.sleep(2)
WebDriverWait(self.driver, self.cfg.login.load_login_iframe_timeout).until(
expected_conditions.visibility_of_element_located((By.CLASS_NAME, "loginframe"))
)
login_frame = self.driver.find_element(By.CLASS_NAME, "loginframe")
self.driver.switch_to.frame(login_frame)
logger.info("等待#loginframe#ptlogin_iframe加载完毕并切换")
WebDriverWait(self.driver, self.cfg.login.load_login_iframe_timeout).until(
expected_conditions.visibility_of_element_located((By.ID, "ptlogin_iframe"))
)
ptlogin_iframe = self.driver.find_element(By.ID, "ptlogin_iframe")
self.driver.switch_to.frame(ptlogin_iframe)
def assert_login_finished_fn():
logger.info(f"{self.name} 请等待#logined可见,则说明已经登录完成了,最大等待时长为{self.cfg.login.login_finished_timeout}")
WebDriverWait(self.driver, self.cfg.login.login_finished_timeout).until(
expected_conditions.visibility_of_element_located((By.ID, "logined"))
)
logger.info("等待1s,确认获取openid的请求完成")
time.sleep(1)
# 确保openid已设置
for t in range(1, 3 + 1):
if self.driver.get_cookie("openid") is None:
logger.info(f"第{t}/3未在道聚城的cookie中找到openid,等一秒再试")
time.sleep(1)
continue
break
self._login_common(login_type, switch_to_login_frame_fn, assert_login_finished_fn, login_action_fn)
# 从cookie中获取openid
return LoginResult(
common_openid=self.get_cookie("openid"),
common_access_token=self.get_cookie("access_token"),
)
default_login_style = 20
def get_switch_to_login_frame_fn(self, appid, daid, s_url, style=default_login_style, theme=2):
# 参数:appid daid
# 21000127 8 普通游戏活动 https://dnf.qq.com/
# 15000103 5 qq空间 https://act.qzone.qq.com/
# 716027609 383 安全管家 https://guanjia.qq.com/
# 1600001063 733 wegame https://www.wegame.com.cn/
# 716027609 383 心悦战场 https://xinyue.qq.com/
# 21000115 8 腾讯游戏/移动游戏 https://dnf.qq.com/
# 532001604 ? 腾讯视频 https://film.qq.com/
# 8000212 18 club.vip https://club.vip.qq.com/qqvip/acts2021/dnf
# 参数:s_url
# 登陆完毕后要跳转的网页
# 参数:style
# 仅二维码 样式一(QQ邮箱设备锁):30
# 二维码/快捷/密码 样式一(整个页面-与之前的兼容(其实就是原来点登录的弹窗)):0/11-15/17/19-23/32-33/40
# 二维码/快捷/密码 样式二(限定大小):25
# 二维码/快捷/密码 样式三(限定大小-格式美化):34 re: 选用
# 二维码/快捷/密码 样式四(居中-移动端风格-需要在手机上,且安装手机QQ后才可以):35/42
# 授权登录(需要在手机上使用):39
# 参数:theme
# 绿色风格:1
# 蓝色风格:2 re: 选用
logger.info("打开登录界面")
login_url = self.get_login_url(appid, daid, s_url, style, theme)
self.open_url_on_start(login_url)
def get_login_url(self, appid, daid, s_url, style=default_login_style, theme=2):
return f"https://xui.ptlogin2.qq.com/cgi-bin/xlogin?appid={appid}&daid={daid}&s_url={quote_plus(s_url)}&style={style}&theme={theme}&target=self"
def _login_common(self, login_type, switch_to_login_frame_fn, assert_login_finished_fn, login_action_fn=None):
"""
通用登录逻辑,并返回登陆后的cookie中包含的uin、skey数据
:rtype: LoginResult
"""
# 实际登录的逻辑,不同方式的处理不同,这里调用外部传入的函数
logger.info(f"{self.name} 开始{login_type}流程")
# 当登录出错时,默认快速重试该次数,避免网络状况等造成的偶然登录失败
quick_retry_max_count = 3
quick_retry_wait_seconds = 2
is_qr_login = self.login_type_qr_login in login_type
short_login_retry_key = "short_login_retry_key"
login_retry_data, retry_timeouts = self.get_retry_data(
short_login_retry_key, quick_retry_max_count, self.get_login_timeout(is_qr_login)
)
if is_qr_login:
# 如果是扫码登录,则每次都等待固定时长
retry_timeouts = [self.get_login_timeout(True) for v in retry_timeouts]
for idx in range_from_one(quick_retry_max_count):
try:
logger.info(color("bold_green") + f"设置标题框为 {self.window_title}")
self.driver.execute_script(f"document.title = '{self.window_title}';")
switch_to_login_frame_fn()
logger.info(color("bold_green") + f"[快速重试阶段] [{idx}/{quick_retry_max_count}] {self.name} 尝试进行登陆")
if login_action_fn is not None:
login_action_fn()
wait_time = retry_timeouts[idx - 1]
logger.info(
color("bold_green")
+ f"[快速重试阶段] [{idx}/{quick_retry_max_count}] {self.name} 尝试等待登录按钮消失~ 最大等待 {wait_time} 秒, retry_timeouts={retry_timeouts}"
)
try:
login_button_id = "login"
if self.login_mode == self.login_mode_supercore:
login_button_id = "go"
WebDriverWait(self.driver, wait_time).until(
expected_conditions.invisibility_of_element_located((By.ID, login_button_id))
)
except NoSuchWindowException:
logger.debug("这种情况好像不影响登录,可以无视")
except BaseException as e:
def _check_secure_verify(ctx: str, css_selector: str):
try:
self.driver.find_element(By.CSS_SELECTOR, css_selector)
verify_max_wait_time = 600
logger.warning(
color("bold_yellow") + f"{self.name} 需要进行 {ctx},将最多等待 {verify_max_wait_time} 秒"
)
if self.cfg.run_in_headless_mode and self.login_slow_retry_index == 1:
raise RequireVerifyMessageButInHeadlessMode(ctx)
WebDriverWait(self.driver, verify_max_wait_time).until(
expected_conditions.invisibility_of_element_located((By.CSS_SELECTOR, css_selector))
)
except RequireVerifyMessageButInHeadlessMode:
raise
except Exception as exc:
logger.debug("other exception", exc_info=exc)
_check_secure_verify("手机号码验证", "#verify_iframe_mask")
_check_secure_verify("安全验证", "#qlogin > #title_1[style='display: block;']")
# 如果没有安全验证验证,则按原样抛出异常
raise e
if idx > 1:
# 第idx-1次的重试成功了,尝试更新历史数据
self.update_retry_data(
short_login_retry_key,
retry_timeouts[idx - 1 - 1],
self.cfg.login.recommended_retry_wait_time_change_rate,
self.name,
)
break
except RequireVerifyMessageButInHeadlessMode:
raise
except Exception as e:
login_mode_name = self.login_mode_to_description[self.login_mode]
logger.error(
f"[快速重试阶段] [{idx}/{quick_retry_max_count}] {self.name} {login_type} 出错了。(此时慢速重试阶段为[{self.login_slow_retry_index}/{self.login_slow_retry_max_count}])\n"
f"为避免偶然因素,前{quick_retry_max_count}次出错将采用快速重试策略,也就是等待{quick_retry_wait_seconds}秒后立刻重试登陆。\n"
+ color("bold_yellow")
+ (
"也许是短期内登陆太多账号显示登录环境异常/网络有问题/出现短信验证码/账号密码不匹配导致。\n"
"请确保 配置工具/公共配置/登录/不显示浏览器 配置为 未勾选 状态,从而确认是否是上述几个问题之一\n"
)
+ "\n"
+ color("bold_green")
+ (
"如果提示扫码登录,应该就是你的qq被判定登录环境异常(类比安全模式),需要扫码一段时间让其自然消失才可以\n"
"如果上面的方法都试了,还是不行,试试关闭 网络连接的ipv6功能,也许会有作用(具体流程请百度)\n"
)
+ "\n"
+ color("bold_cyan")
+ "如果之前版本都是正常的,可以试试切换旧版本chrome - 配置工具/公共配置/登录/强制使用特定大版本chrome,修改为94或者更早的版本,并开启 强制使用便携版 开关\n"
+ "\n"
+ color("bold_green")
+ f"如果一直卡在 {login_mode_name} 登录流程,可能是你网络没法登录这个,或者是该登陆类型的服务器抽风了(经常出现),建议多试几次。\n"
+ f"真不行就去配置工具打开 【当前账号配置/活动开关/登陆类型开关/禁用 {login_mode_name} 登录】 开关,从而跳过 {login_mode_name} 类型的登录\n"
+ "",
exc_info=e,
)
if self.login_mode == self.login_mode_guanjia:
logger.warning(
color("bold_green") + "如果一直卡在管家登录流程,可能是你网络没法登录这个,建议多试几次,真不行就去配置工具关闭 管家 活动 的开关(不是关闭这个登录页面)~"
)
logger.info("电脑管家 模式不尝试短时间重试,直接等待下次重试")
break
if self.login_mode == self.login_mode_iwan:
logger.warning(
color("bold_green")
+ "如果一直卡在 iwan 登录流程,可能是你网络没法登录这个,建议多试几次,真不行就去配置工具关闭 qq视频 活动 的开关(不是关闭这个登录页面)~"
)
logger.info("iwan 模式不尝试短时间重试,直接等待下次重试")
break
logger.info(f"为避免本次异常是在登录完成后发生的,也就是此时页面已经不是登录页面了,导致后续登录尝试一直失败,这里主动重新打开登录页面: {self.login_url}")
self.driver.get(self.login_url)
if idx < quick_retry_max_count:
time.sleep(quick_retry_wait_seconds)
else:
raise Exception("快速重试最大上限后仍失败了")
logger.info(f"{self.name} 回到主iframe")
self.driver.switch_to.default_content()
logger.info(f"{self.name} 当前网址为 {self.driver.current_url}")
assert_login_finished_fn()
logger.info(f"{self.name} 登录完成")
self.cookies = self.driver.get_cookies()
if self.login_mode in [self.login_mode_normal, self.login_mode_qzone]:
self.fetch_qq_video_vuserid()
if self.login_mode in [self.login_mode_normal]:
self.fetch_apps_p_skey()
if self.login_mode in [self.login_mode_xinyue]:
self.fetch_xinyue_openid_access_token()
if self.login_mode in [self.login_mode_guanjia]:
self.wait_for_IED_LOG_INFO2_QC()
if self.login_mode in [self.login_mode_iwan]:
self.fetch_iwan_openid_access_token()
if self.login_mode in [self.login_mode_supercore]:
self.wait_for_openid_access_token()
if self.login_mode in [self.login_mode_djc]:
self.fetch_djc_openid_access_token()
self.print_cookie()
return
def get_retry_data(self, retry_key: str, max_retry_count: int, max_retry_wait_time: int):
# 结合历史数据和配置,计算各轮重试等待的时间
login_retry_data = LoginRetryDB().with_context(retry_key).load()
retry_timeouts = []
if max_retry_count == 1:
retry_timeouts = [max_retry_wait_time]
elif max_retry_count > 1:
# 默认重试时间为按时长等分递增
retry_timeouts = list(
idx / max_retry_count * max_retry_wait_time for idx in range_from_one(max_retry_count)
)
if login_retry_data.recommended_first_retry_timeout != 0:
# 如果有历史成功数据,则以推荐首次重试时间为第一次重试的时间,后续重试次数则等分递增
retry_timeouts = [login_retry_data.recommended_first_retry_timeout]
remaining_retry_count = max_retry_count - 1
retry_timeouts.extend(
list(
idx / remaining_retry_count * max_retry_wait_time
for idx in range_from_one(remaining_retry_count)
)
)
return login_retry_data, retry_timeouts
def update_retry_data(
self, retry_key: str, success_timeout: int, recommended_retry_wait_time_change_rate=0.125, debug_ctx=""
):
def cb(login_retry_data: LoginRetryDB):
# 第idx-1次的重试成功了,尝试更新历史数据
cr = recommended_retry_wait_time_change_rate
login_retry_data.recommended_first_retry_timeout = (
1 - cr
) * login_retry_data.recommended_first_retry_timeout + cr * success_timeout
login_retry_data.history_success_timeouts.append(success_timeout)
if use_by_myself():
logger.info(
color("bold_cyan") + f"(仅我可见){debug_ctx} 本次重试等待时间为{success_timeout},当前历史重试数据为{login_retry_data}"
)
LoginRetryDB().with_context(retry_key).update(cb)
def get_login_timeout(self, is_qr_mode=False):
if not is_qr_mode:
return self.cfg.login.login_timeout
else:
# 扫码模式保底需要等待600秒
return max(self.cfg.login.login_timeout, 600)
def fetch_qq_video_vuserid(self):
logger.info(f"{self.name} 转到qq视频界面,从而可以获取vuserid,用于腾讯视频的蚊子腿")
self.driver.get("https://m.film.qq.com/magic-act/110254/index.html")
(vuserid,) = self._wait_for_cookies("vuserid")
self.add_cookie("vuserid", vuserid)
def fetch_apps_p_skey(self):
logger.info(f"{self.name} 跳转到apps.game.qq.com,用于获取该域名下的p_skey,用于部分分享功能")
self.driver.get("https://apps.game.qq.com/")
time.sleep(1)
(p_skey,) = self._wait_for_cookies("p_skey")
self.add_cookie("apps_p_skey", p_skey)
def fetch_xinyue_openid_access_token(self):
# 先尝试使用活动页面的cookie
logger.info(f"{self.name} 先尝试使用act活动页面 {self.driver.current_url},用于获取该域名下的openid和access_token,用于心悦相关操作")
openid, access_token = self._wait_for_cookies("openid", "access_token")
self.add_cookie("xinyue_openid", openid)
self.add_cookie("xinyue_access_token", access_token)
# 然后跳转到心悦域名,若该页面也能获取到相关cookie,则优先使用该cookie
logger.info(f"{self.name} 再等待1秒,然后跳转到xinyue.qq.com,用于获取该域名下的openid和access_token,用于心悦相关操作")
time.sleep(1)
self.driver.get("https://xinyue.qq.com/")
time.sleep(1)
openid, access_token = self._wait_for_cookies("openid", "access_token")
self.add_cookie("xinyue_openid", openid)
self.add_cookie("xinyue_access_token", access_token)
def wait_for_IED_LOG_INFO2_QC(self):
(userinfo,) = self._wait_for_cookies("uInfo101478239")
def fetch_iwan_openid_access_token(self):
logger.info(f"{self.name} 获取爱玩的openid和access_token,用于腾讯视频蚊子腿相关操作")
openid, access_token = self._wait_for_cookies("vqq_openid", "vqq_access_token")
def fetch_djc_openid_access_token(self):
logger.info(f"{self.name} 获取道聚城的openid和access_token,用于道聚城相关操作")
openid, access_token = self._wait_for_cookies("openid", "access_token")
def wait_for_openid_access_token(self):
logger.info(f"{self.name} 等待openid和access_token出现")
self._wait_for_cookies("openid", "access_token")
def _wait_for_cookies(self, *cookie_names: str, max_try: int = 5) -> list[dict | None]:
values: list[dict | None] = []
for _i in range_from_one(max_try):
values = [None for i in range(len(cookie_names))]
try:
for idx, name in enumerate(cookie_names):
values[idx] = self.driver.get_cookie(name)
except Exception as e:
logger.debug(f"第 {_i}/{max_try} 次尝试获取cookie {cookie_names} 失败了", exc_info=e)
if self._all_is_not_none(values):
break
time.sleep(1)
return values
def _all_is_not_none(self, iterable) -> bool:
for element in iterable:
if element is None:
return False
return True
def try_auto_resolve_captcha(self):
try:
self._try_auto_resolve_captcha()
except RequireVerifyMessageButInHeadlessMode:
raise
except Exception as e:
msg = f"ver {now_version} {ver_time} {self.name} 自动处理验证失败了,出现未捕获的异常,请加群{get_config().common.qq_group}反馈或自行解决。请手动进行处理验证码"
logger.exception(color("fg_bold_red") + msg, exc_info=e)
logger.warning(color("fg_bold_cyan") + "如果稳定报错,不妨打开网盘,看看是否有新版本修复了这个问题~")
logger.warning(color("fg_bold_cyan") + f"链接:{get_config().common.netdisk_link}")
def _try_auto_resolve_captcha(self):
if not self.cfg.login.auto_resolve_captcha:
logger.info(f"{self.name} 未启用自动处理拖拽验证码的功能")
return
if self.cfg.login.move_captcha_delta_width_rate_v2 <= 0:
logger.info(f"{self.name} 未设置每次尝试的偏移值,跳过自动拖拽验证码")
return
captcha_try_count = 0
success_xoffset = 0
account_db = CaptchaDB().with_context(self.name).load()
try:
iframe_id = "tcaptcha_iframe_dy"
WebDriverWait(self.driver, self.cfg.login.open_url_wait_time).until(
expected_conditions.visibility_of_element_located((By.ID, iframe_id))
)
tcaptcha_iframe = self.driver.find_element(By.ID, iframe_id)
self.driver.switch_to.frame(tcaptcha_iframe)
logger.info(
color("bold_green") + f"{self.name} 检测到了滑动验证码,将开始自动处理。(若验证码完毕会出现短信验证,请去配置文件关闭本功能,目前暂不支持带短信验证的情况)"
)
logger.warning(color("bold_yellow") + "新版滑动验证码限制最大滑动次数为3次,之前的暴力尝试策略不再可用,请先手动操作。待日后有空时,会改用图像识别的方式来进行处理")
if self.cfg.run_in_headless_mode and self.login_slow_retry_index == 1:
raise RequireVerifyMessageButInHeadlessMode("新版滑动验证码")
return
# 新版中,三个组件的位置随机的,需要根据其样式去判断是哪个
selectors = [
"#tcOperation > div:nth-child(6)",
"#tcOperation > div:nth-child(7)",
"#tcOperation > div:nth-child(8)",
]
try:
WebDriverWait(self.driver, self.cfg.login.open_url_wait_time).until(
expected_conditions.visibility_of_element_located((By.CSS_SELECTOR, selectors[0]))
)
WebDriverWait(self.driver, self.cfg.login.open_url_wait_time).until(
expected_conditions.visibility_of_element_located((By.CSS_SELECTOR, selectors[1]))
)
WebDriverWait(self.driver, self.cfg.login.open_url_wait_time).until(
expected_conditions.visibility_of_element_located((By.CSS_SELECTOR, selectors[2]))
)
except Exception as e:
logger.warning(f"{self.name} 等待验证码相关元素出现失败了,将按照默认宽度进行操作", exc_info=e)
# 先获取每个组件
items = [self.driver.find_element(By.CSS_SELECTOR, selector) for selector in selectors]
items.sort(key=lambda item: item.size["width"], reverse=True)
# 按照页面里的效果,理论上,宽度从大到小依次为 滑轨、滑块、上方缺失方块,比如测试时的数值分别为 280/54/50
drag_tarck_width = items[0].size["width"] or 280 # 进度条轨道宽度
drag_block_width = items[1].size["width"] or 54 # 滑块宽度
missing_block_width = items[2].size["width"] or 50 # 上方缺失方块宽度
delta_width = int(missing_block_width * self.cfg.login.move_captcha_delta_width_rate_v2) or 5 # 每次尝试多移动该宽度
# 获取滑块,方便后面滚动
drag_button = items[1] # 滑块
# 根据经验,缺失验证码大部分时候出现在右侧,所以从右侧开始尝试
xoffsets = []
init_offset = drag_tarck_width - drag_block_width - delta_width
if len(account_db.offset_to_history_succes_count) != 0:
# 若有则取其中最频繁的前几个作为优先尝试项
mostCommon = Counter(account_db.offset_to_history_succes_count).most_common()
logger.info(f"{self.name} 根据本地记录数据,过去运行中成功解锁次数最多的偏移值为:{mostCommon},将首先尝试他们")
for xoffset, _success_count in mostCommon:
xoffsets.append(int(xoffset))
else:
# 没有历史数据,只能取默认经验值了
# 有几个位置经常出现,如2/4和3/4个滑块处,优先尝试
xoffsets.append(init_offset - 2 * (missing_block_width // 4))
xoffsets.append(init_offset - 3 * (missing_block_width // 4))
logger.info(
color("bold_green")
+ f"{self.name} 验证码相关信息:轨道宽度为{drag_tarck_width},滑块宽度为{drag_block_width},上方方块宽度为{missing_block_width},偏移递增量为{delta_width}({self.cfg.login.move_captcha_delta_width_rate_v2:.2f}倍滑块宽度),初始偏差值为{init_offset}"
)
# 将普通序列放入其中
xoffset = init_offset
while xoffset > 0:
xoffsets.append(xoffset)
xoffset -= delta_width
wait_time = 1
logger.info(f"{self.name} 先release滑块一次,以避免首次必定失败的问题")
ActionChains(self.driver).release(on_element=drag_button).perform()
time.sleep(wait_time)
logger.info(color("bold_green") + f"{self.name} 开始拖拽验证码,将依次尝试下列偏移量:\n{xoffsets}")
for xoffset in xoffsets:
ActionChains(self.driver).click_and_hold(on_element=drag_button).perform() # 左键按下
time.sleep(0.5)
ActionChains(self.driver).move_by_offset(xoffset=xoffset, yoffset=0).perform() # 将滑块向右滑动指定距离
time.sleep(0.5)
ActionChains(self.driver).release(on_element=drag_button).perform() # 左键放下,完成一次验证尝试
time.sleep(0.5)
captcha_try_count += 1
success_xoffset = xoffset
distance_rate = (init_offset - xoffset) / missing_block_width
logger.info(
f"{self.name} 尝试第{captcha_try_count}次拖拽验证码,本次尝试偏移量为{xoffset},距离右侧初始尝试位置({init_offset})距离相当于{distance_rate:.2f}个滑块宽度(若失败将等待{wait_time}秒)"
)
time.sleep(wait_time)
self.driver.switch_to.parent_frame()
except StaleElementReferenceException:
logger.info(f"{self.name} 成功完成了拖拽验证码操作,总计尝试次数为{captcha_try_count}")
# 更新历史数据
account_db.increse_success_count(success_xoffset)
account_db.save()
except (TimeoutException, NoSuchWindowException):
logger.info(f"{self.name} 看上去没有出现验证码")
def set_window_size(self, width: int = 1936, height: int = 1056):
logger.info(f"浏览器设为 {width} * {height}")
self.driver.set_window_size(width, height)
def add_cookies(self, cookies):
to_add = []
for cookie in cookies:
if self.get_cookie(cookie["name"]) == "":
to_add.append(cookie)
self.cookies.extend(to_add)
def add_cookie(self, new_name, cookie):
if cookie is None:
return
cookie["name"] = new_name
self.cookies.append(cookie)
logger.warning(f"{self.name} add_cookie {cookie['domain']} {cookie['name']} {cookie['value']}")
def get_cookie(self, name):
# 这里倒着遍历,从而优先获取后添加进去的
for cookie in reversed(self.cookies):
if cookie["name"] == name and cookie["value"] != "":
return cookie["value"]
return ""
def print_cookie(self):
for cookie in self.cookies:
domain, name, value = cookie["domain"], cookie["name"], cookie["value"]
logger.debug(f"{domain:20s} {name:20s} {value:20s} {cookie}")
def open_url_on_start(self, url):
chrome_default_url = "data:,"
while True:
self.driver.get(url)
# 等待一会,确保地址栏url变量已变更
time.sleep(0.1)
if self.driver.current_url != chrome_default_url:
break
logger.info(f"尝试打开网页({url}),但似乎指令未生效,当前地址栏仍为{chrome_default_url},等待{self.cfg.retry.retry_wait_time}秒后重试")
time.sleep(self.cfg.retry.retry_wait_time)
def need_reopen_url(self, login_type):
return self.login_type_auto_login in login_type and self.cfg.run_in_headless_mode
def test():
# 读取配置信息
load_config("config.toml", "config.toml.local")
cfg = get_config()
cfg.common.force_use_portable_chrome = True
cfg.common.run_in_headless_mode = False
cfg.common.login.enable_auto_click_avatar_in_auto_login = False
cfg.common.login.enable_auto_click_avatar_in_qr_login = False
cfg.common.login.login_timeout = 5
cfg.common.login.retry_wait_time = 10
# 获取登录相关信息
all_login_types = [getattr(QQLogin, attr) for attr in dir(QQLogin) if attr.startswith("login_type_")]
all_login_modes = [getattr(QQLogin, attr) for attr in dir(QQLogin) if attr.startswith("login_mode_")]
# 测试开关
TEST_SWITCH_RUN_COUNT = 1
TEST_SWITCH_RUN_PARALLEL = False
TEST_SWITCH_TEST_ALL_ACCOUNTS = False
TEST_SWITCH_TEST_ALL_LOGIN_TYPES = False
TEST_SWITCH_TEST_ALL_LOGIN_MODES = False
# 需要运行的测试维度:账号、登录类别、登录模式
login_accounts = [cfg.account_configs[idx - 1] for idx in [1]]
login_types = [QQLogin.login_type_auto_login]
login_modes = [QQLogin.login_mode_qzone]
if TEST_SWITCH_TEST_ALL_ACCOUNTS:
login_accounts = [account for account in cfg.account_configs]
if TEST_SWITCH_TEST_ALL_LOGIN_TYPES:
login_types = all_login_types
if TEST_SWITCH_TEST_ALL_LOGIN_MODES:
login_modes = all_login_modes
# 根据设定的维度组合出所有需要测试的用例
test_cases = []
total = len(login_types) * len(login_modes) * len(login_accounts) * TEST_SWITCH_RUN_COUNT
window_index = 0
for _loop_index in range(TEST_SWITCH_RUN_COUNT):
for login_type in login_types:
for login_mode in login_modes:
for login_account in login_accounts:
window_index += 1
test_cases.append((login_mode, login_type, window_index, total, cfg.common, login_account))
# 预先尝试下载解压缩
QQLogin(cfg.common).check_and_download_chrome_ahead()
# 展示测试概览
show_head_line(f"将开始测试以下{total}个用例,并行模式={TEST_SWITCH_RUN_PARALLEL}", color("bold_green"))
logger.info(color("bold_green") + f"循环次数={TEST_SWITCH_RUN_COUNT}")
logger.info(color("bold_green") + f"账号列表={[account.name for account in login_accounts]}")
logger.info(color("bold_green") + f"登陆类型={login_types}")
logger.info(color("bold_green") + f"登录模式={login_modes}")
# 实际测试内容
if not TEST_SWITCH_RUN_PARALLEL:
# 串行测试
for test_case in test_cases:
do_login(*test_case)
else:
# 并行测试
from multiprocessing import Pool, cpu_count, freeze_support
freeze_support()
pool_size = min(cpu_count(), len(test_cases))
with Pool(pool_size) as pool:
logger.info(color("bold_cyan") + f"进程池已初始化完毕,大小为 {pool_size}")
pool.starmap(do_login, test_cases)
show_head_line(f"全部{total}个测试运行完毕", color("bold_green"))
def do_login(
login_mode: str, login_type: str, window_index: int, total, common_cfg: CommonConfig, account: AccountConfig
):
show_head_line(
f"用例({window_index}/{total}): 测试 {account.name} 使用 {login_type} 来登录 {login_mode}", color("bold_green")
)
acc = account.account_info
ql = QQLogin(common_cfg, window_index=window_index)
if login_type == ql.login_type_auto_login:
lr = ql.login(acc.account, acc.password, login_mode, name=account.name)
else:
lr = ql.qr_login(login_mode, name=account.name, account=acc.account)
logger.info(
color("bold_green")
+ f"用例({window_index}/{total}): 测试 {account.name} 使用 {login_type} 来登录 {login_mode} 的结果为: {lr}"
)
def demo_download_chrome():
load_config("config.toml", "config.toml.local")
cfg = get_config()
cfg.common.force_use_portable_chrome = True
QQLogin(cfg.common).check_and_download_chrome_ahead()
if __name__ == "__main__":
test()
# demo_download_chrome()
1
https://gitee.com/a915646637/djc_helper.git
git@gitee.com:a915646637/djc_helper.git
a915646637
djc_helper
djc_helper
master

搜索帮助