import aiohttp import argparse import asyncio from api.qinglong import QlApi, QlOpenApi from api.send import SendApi from utils.ck import get_invalid_ck_ids from config import ( qinglong_data, user_datas, ) import cv2 import json from loguru import logger import os from playwright.async_api import Playwright, async_playwright from playwright._impl._errors import TimeoutError import random import re from PIL import Image # 用于图像处理 import traceback from typing import Union from utils.consts import ( jd_login_url, supported_types, supported_colors, supported_sms_func ) from utils.tools import ( get_tmp_dir, get_img_bytes, get_forbidden_users_dict, filter_forbidden_users, save_img, get_ocr, get_word, get_shape_location_by_type, get_shape_location_by_color, rgba2rgb, send_msg, new_solve_slider_captcha, ddddocr_find_files_pic, expand_coordinates, cv2_save_img, ddddocr_find_bytes_pic, solve_slider_captcha, validate_proxy_config, is_valid_verification_code, filter_cks, extract_pt_pin, desensitize_account ) """ 基于playwright做的 """ logger.add( sink="main.log", level="DEBUG" ) try: # 账号是否脱敏的开关 from config import enable_desensitize except ImportError: enable_desensitize = False async def download_image(url, filepath): async with aiohttp.ClientSession() as session: async with session.get(url) as response: if response.status == 200: with open(filepath, 'wb') as f: f.write(await response.read()) print(f"Image downloaded to {filepath}") else: print(f"Failed to download image. Status code: {response.status}") async def check_notice(page): try: logger.info("检查登录是否报错") notice = await page.wait_for_function( """ () => { const notice = document.querySelectorAll('.notice')[1]; return notice && notice.textContent.trim() !== '' ? notice.textContent.trim() : false; } """, timeout = 3000 ) raise RuntimeError(notice) except TimeoutError: logger.info("登录未发现报错") return async def auto_move_slide_v2(page, retry_times: int = 2, slider_selector: str = 'img.move-img', move_solve_type: str = ""): for i in range(retry_times): logger.info(f'第{i + 1}次开启滑块验证') # 查找小图 try: # 查找小图 await page.wait_for_selector('.captcha_drop', state='visible', timeout=3000) except Exception as e: logger.info('未找到验证码框, 退出滑块验证') return await auto_move_slide(page, retry_times=5, slider_selector = slider_selector, move_solve_type = move_solve_type) # 判断是否一次过了滑块 captcha_drop_visible = await page.is_visible('.captcha_drop') # 存在就重新滑一次 if captcha_drop_visible: if i == retry_times - 1: return logger.info('一次过滑块失败, 再次尝试滑块验证') await page.wait_for_selector('.captcha_drop', state='visible', timeout=3000) # 点外键 sign_locator = page.locator('#header').locator('.text-header') sign_locator_box = await sign_locator.bounding_box() sign_locator_left_x = sign_locator_box['x'] sign_locator_left_y = sign_locator_box['y'] await page.mouse.click(sign_locator_left_x, sign_locator_left_y) await asyncio.sleep(1) # 提交键 submit_locator = page.locator('.btn.J_ping.active') await submit_locator.click() await asyncio.sleep(1) continue return async def auto_move_slide(page, retry_times: int = 2, slider_selector: str = 'img.move-img', move_solve_type: str = ""): """ 自动识别移动滑块验证码 """ logger.info("开始滑块验证") for i in range(retry_times + 1): try: # 查找小图 await page.wait_for_selector('#slot_img', state='visible', timeout=3000) except Exception as e: # 未找到元素，认为成功，退出循环 logger.info('未找到滑块,退出滑块验证') break # 滑块验证失败了 if i + 1 == retry_times + 1: raise Exception("滑块验证失败了") logger.info(f'第{i + 1}次尝试自动移动滑块中...') # 获取 src 属性 small_src = await page.locator('#slot_img').get_attribute('src') background_src = await page.locator('#main_img').get_attribute('src') # 获取 bytes small_img_bytes = get_img_bytes(small_src) background_img_bytes = get_img_bytes(background_src) # 保存小图 small_img_path = save_img('small_img', small_img_bytes) small_img_width = await page.evaluate('() => { return document.getElementById("slot_img").clientWidth; }') # 获取网页的图片尺寸 small_img_height = await page.evaluate('() => { return document.getElementById("slot_img").clientHeight; }') # 获取网页的图片尺寸 small_image = Image.open(small_img_path) # 打开图像 resized_small_image = small_image.resize((small_img_width, small_img_height)) # 调整图像尺寸 resized_small_image.save(small_img_path) # 保存调整后的图像 # 保存大图 background_img_path = save_img('background_img', background_img_bytes) background_img_width = await page.evaluate('() => { return document.getElementById("main_img").clientWidth; }') # 获取网页的图片尺寸 background_img_height = await page.evaluate('() => { return document.getElementById("main_img").clientHeight; }') # 获取网页的图片尺寸 background_image = Image.open(background_img_path) # 打开图像 resized_background_image = background_image.resize((background_img_width, background_img_height)) # 调整图像尺寸 resized_background_image.save(background_img_path) # 保存调整后的图像 # 获取滑块 slider = page.locator(slider_selector) await asyncio.sleep(1) # 这里是一个标准算法偏差 slide_difference = 10 if move_solve_type == "old": # 用于调试 distance = ddddocr_find_bytes_pic(small_img_bytes, background_img_bytes) await asyncio.sleep(1) await solve_slider_captcha(page, slider, distance, slide_difference) await asyncio.sleep(1) continue # 获取要移动的长度 distance = ddddocr_find_files_pic(small_img_path, background_img_path) await asyncio.sleep(1) # 移动滑块 await new_solve_slider_captcha(page, slider, distance, slide_difference) await asyncio.sleep(1) async def auto_shape(page, retry_times: int = 5): logger.info("开始二次验证") # 图像识别 ocr = get_ocr(beta=True) # 文字识别 det = get_ocr(det=True) # 自己训练的ocr, 提高文字识别度 my_ocr = get_ocr(det=False, ocr=False, import_onnx_path="myocr_v1.onnx", charsets_path="charsets.json") """ 自动识别滑块验证码 """ for i in range(retry_times + 1): try: # 查找小图 await page.wait_for_selector('div.captcha_footer img', state='visible', timeout=3000) except Exception as e: # 未找到元素，认为成功，退出循环 logger.info('未找到二次验证图,退出二次验证识别') break # 二次验证失败了 if i + 1 == retry_times + 1: raise Exception("二次验证失败了") logger.info(f'第{i + 1}次自动识别形状中...') tmp_dir = get_tmp_dir() background_img_path = os.path.join(tmp_dir, f'background_img.png') # 获取大图元素 background_locator = page.locator('#cpc_img') # 获取元素的位置和尺寸 backend_bounding_box = await background_locator.bounding_box() backend_top_left_x = backend_bounding_box['x'] backend_top_left_y = backend_bounding_box['y'] # 截取元素区域 await page.screenshot(path=background_img_path, clip=backend_bounding_box) # 获取 图片的src 属性和button按键 word_img_src = await page.locator('div.captcha_footer img').get_attribute('src') button = page.locator('div.captcha_footer button#submit-btn') # 找到刷新按钮 refresh_button = page.locator('.jcap_refresh') # 获取文字图并保存 word_img_bytes = get_img_bytes(word_img_src) rgba_word_img_path = save_img('rgba_word_img', word_img_bytes) # 图像识别的解法，东哥求放过啊，写不动了 if page.locator("div.sp_msg.tip_text", has_text="请点击上图中的").is_visible(): logger.info("检测为图像, 开始图像识别......") from utils.tools import crop_center_contour, ddddocr_find_files_pic_v2 small_img_path = os.path.join(tmp_dir, f'small_img.png') # 这里是一个标准算法偏差 slide_difference = 10 try: # 将中间的图截取出来，才能更好的识别 result = crop_center_contour(rgba_word_img_path, small_img_path, min_area=100, padding=1) if result is None: raise IndexError("截图异常") # 获取要移动的长度 target_dict = ddddocr_find_files_pic_v2(small_img_path, background_img_path) # 提取坐标 x1, y1, x2, y2 = target_dict["target"] center_x = (x1 + slide_difference + x2) // 2 center_y = (y1 + y2) // 2 await asyncio.sleep(random.uniform(0, 1)) logger.info("已检测到图像，尝试点击中") x, y = backend_top_left_x + center_x, backend_top_left_y + center_y # 点击图片 await page.mouse.click(x, y) except IndexError: logger.info(f'识别图像出错,刷新中......') await refresh_button.click() await asyncio.sleep(random.uniform(2, 4)) continue # 文字图是RGBA的，有蒙板识别不了，需要转成RGB rgb_word_img_path = rgba2rgb('rgb_word_img', rgba_word_img_path) # 获取问题的文字 word = get_word(ocr, rgb_word_img_path) if word.find('色') > 0: target_color = word.split('请选出图中')[1].split('的图形')[0] if target_color in supported_colors: logger.info(f'正在点击中......') # 获取点的中心点 center_x, center_y = get_shape_location_by_color(background_img_path, target_color) if center_x is None and center_y is None: logger.info(f'识别失败,刷新中......') await refresh_button.click() await asyncio.sleep(random.uniform(2, 4)) continue # 得到网页上的中心点 x, y = backend_top_left_x + center_x, backend_top_left_y + center_y # 点击图片 await page.mouse.click(x, y) await asyncio.sleep(random.uniform(1, 4)) # 点击确定 await button.click() await asyncio.sleep(random.uniform(2, 4)) continue else: logger.info(f'不支持{target_color},刷新中......') # 刷新 await refresh_button.click() await asyncio.sleep(random.uniform(2, 4)) continue # 这里是文字验证码了 elif word.find('依次') > 0 or word.find('按照次序点选') > 0: logger.info(f'开始文字识别,点击中......') # 获取文字的顺序列表 try: if word.find('依次') > 0: target_char_list = list(re.findall(r'[\u4e00-\u9fff]+', word)[1]) if word.find('按照次序点选') > 0: target_char_list = list(word.split('请按照次序点选')[1]) except IndexError: logger.info(f'识别文字出错,刷新中......') await refresh_button.click() await asyncio.sleep(random.uniform(2, 4)) continue target_char_len = len(target_char_list) # 识别字数不对 if target_char_len < 4: logger.info(f'识别的字数小于4,刷新中......') await refresh_button.click() await asyncio.sleep(random.uniform(2, 4)) continue # 取前4个的文字 target_char_list = target_char_list[:4] # 定义【文字, 坐标】的列表 target_list = [[x, []] for x in target_char_list] # 获取大图的二进制 background_locator = page.locator('#cpc_img') background_locator_src = await background_locator.get_attribute('src') background_locator_bytes = get_img_bytes(background_locator_src) bboxes = det.detection(background_locator_bytes) count = 0 im = cv2.imread(background_img_path) for bbox in bboxes: # 左上角 x1, y1, x2, y2 = bbox # 做了一下扩大 expanded_x1, expanded_y1, expanded_x2, expanded_y2 = expand_coordinates(x1, y1, x2, y2, 10) im2 = im[expanded_y1:expanded_y2, expanded_x1:expanded_x2] img_path = cv2_save_img('word', im2) image_bytes = open(img_path, "rb").read() result = my_ocr.classification(image_bytes) if result in target_char_list: for index, target in enumerate(target_list): if result == target[0] and target[0] is not None: x = x1 + (x2 - x1) / 2 y = y1 + (y2 - y1) / 2 target_list[index][1] = [x, y] count += 1 if count != target_char_len: logger.info(f'文字识别失败,刷新中......') await refresh_button.click() await asyncio.sleep(random.uniform(2, 4)) continue await asyncio.sleep(random.uniform(0, 1)) try: for char in target_list: center_x = char[1][0] center_y = char[1][1] # 得到网页上的中心点 x, y = backend_top_left_x + center_x, backend_top_left_y + center_y # 点击图片 await page.mouse.click(x, y) await asyncio.sleep(random.uniform(1, 4)) except IndexError: logger.info(f'识别文字出错,刷新中......') await refresh_button.click() await asyncio.sleep(random.uniform(2, 4)) continue # 点击确定 await button.click() await asyncio.sleep(random.uniform(2, 4)) else: shape_type = word.split('请选出图中的')[1] if shape_type in supported_types: logger.info(f'已找到图形,点击中......') if shape_type == "圆环": shape_type = shape_type.replace('圆环', '圆形') # 获取点的中心点 center_x, center_y = get_shape_location_by_type(background_img_path, shape_type) if center_x is None and center_y is None: logger.info(f'识别失败,刷新中......') await refresh_button.click() await asyncio.sleep(random.uniform(2, 4)) continue # 得到网页上的中心点 x, y = backend_top_left_x + center_x, backend_top_left_y + center_y # 点击图片 await page.mouse.click(x, y) await asyncio.sleep(random.uniform(1, 4)) # 点击确定 await button.click() await asyncio.sleep(random.uniform(2, 4)) continue else: logger.info(f'不支持{shape_type},刷新中......') # 刷新 await refresh_button.click() await asyncio.sleep(random.uniform(2, 4)) continue async def sms_recognition(page, user, mode): try: from config import sms_func except ImportError: sms_func = "no" sms_func = user_datas[user].get("sms_func", sms_func) if sms_func not in supported_sms_func: raise Exception(f"sms_func只支持{supported_sms_func}") if mode == "cron" and sms_func == "manual_input": sms_func = "no" if sms_func == "no": raise Exception("sms_func为no关闭, 跳过短信验证码识别环节") logger.info('点击【获取验证码】中') await page.click('button.getMsg-btn') await asyncio.sleep(1) # 自动识别滑块 await auto_move_slide(page, retry_times=5) await auto_shape(page, retry_times=30) # 识别是否成功发送验证码 await page.wait_for_selector('button.getMsg-btn:has-text("重新发送")', timeout=3000) logger.info("发送短信验证码成功") # 手动输入 # 用户在60S内，手动在终端输入验证码 if sms_func == "manual_input": logger.info("启用手动输入验证码模式") from inputimeout import inputimeout, TimeoutOccurred try: verification_code = inputimeout(prompt="请输入验证码：", timeout=60) except TimeoutOccurred: return # 通过调用web_hook的方式来实现全自动输入验证码 elif sms_func == "webhook": logger.info("启用webhook获取验证码模式") from utils.tools import send_request try: from config import sms_webhook except ImportError: sms_webhook = "" sms_webhook = user_datas[user].get("sms_webhook", sms_webhook) if sms_webhook is None: raise Exception(f"sms_webhook未配置") headers = { 'Content-Type': 'application/json', } data = {"phone_number": user} response = await send_request(url=sms_webhook, method="post", headers=headers, data=data) verification_code = response['data']['code'] await asyncio.sleep(1) if not is_valid_verification_code(verification_code): logger.error(f"验证码需为6位数字, 输入的验证码为{verification_code}, 异常") raise Exception(f"验证码异常") logger.info('填写验证码中...') verification_code_input = page.locator('input.acc-input.msgCode') for v in verification_code: await verification_code_input.type(v, no_wait_after=True) await asyncio.sleep(random.random() / 10) logger.info('点击提交中...') await page.click('a.btn') async def voice_verification(page, user, mode): from utils.consts import supported_voice_func try: from config import voice_func except ImportError: voice_func = "no" voice_func = user_datas[user].get("voice_func", voice_func) if voice_func not in supported_voice_func: raise Exception(f"voice_func只支持{supported_voice_func}") if mode == "cron" and voice_func == "manual_input": voice_func = "no" if voice_func == "no": raise Exception("voice_func为no关闭, 跳过手机语音识别") logger.info('点击获取验证码中') await page.click('button.getMsg-btn:has-text("点击获取验证码")') await asyncio.sleep(1) # 自动识别滑块 await auto_move_slide(page, retry_times=5, slider_selector='#slider') await auto_shape(page, retry_times=30) # 识别是否成功发送验证码 await page.wait_for_selector('button.getMsg-btn:has-text("重新发送")', timeout=3000) logger.info("发送手机语音识别验证码成功") # 手动输入 # 用户在60S内，手动在终端输入验证码 if voice_func == "manual_input": from inputimeout import inputimeout, TimeoutOccurred try: verification_code = inputimeout(prompt="请输入验证码：", timeout=60) except TimeoutOccurred: return await asyncio.sleep(1) if not is_valid_verification_code(verification_code): logger.error(f"验证码需为6位数字, 输入的验证码为{verification_code}, 异常") raise Exception(f"验证码异常") logger.info('填写验证码中...') verification_code_input = page.locator('input.acc-input.msgCode') for v in verification_code: await verification_code_input.type(v, no_wait_after=True) await asyncio.sleep(random.random() / 10) logger.info('点击提交中...') await page.click('a.btn') async def check_dialog(page): logger.info("开始弹窗检测") try: # 等待 dialog 出现 await page.wait_for_selector(".dialog", timeout=4000) except Exception as e: logger.info('未找到弹框, 退出弹框检测') return # 获取 dialog-des 的文本内容 dialog_text = await page.locator(".dialog-des").text_content() if dialog_text == "您的账号存在风险，为了账号安全需实名认证，是否继续？": raise Exception("检测到实名认证弹窗，请前往移动端做实名认证") else: raise Exception("检测到不支持的弹窗, 更新异常") async def get_jd_pt_key(playwright: Playwright, user, mode) -> Union[str, None]: """ 获取jd的pt_key """ try: from config import headless except ImportError: headless = False args = '--no-sandbox', '--disable-setuid-sandbox', '--disable-software-rasterizer', '--disable-gpu' try: # 引入代理 from config import proxy # 检查代理的配置 is_proxy_valid, msg = validate_proxy_config(proxy) if not is_proxy_valid: logger.error(msg) proxy = None if msg == "未配置代理": logger.info(msg) proxy = None except ImportError: logger.info("未配置代理") proxy = None browser = await playwright.chromium.launch(headless=headless, args=args, proxy=proxy) try: # 引入UA from config import user_agent except ImportError: from utils.consts import user_agent context = await browser.new_context(user_agent=user_agent) try: page = await context.new_page() await page.set_viewport_size({"width": 360, "height": 640}) await page.goto(jd_login_url) if user_datas[user].get("user_type") == "qq": await page.get_by_role("checkbox").check() await asyncio.sleep(1) # 点击QQ登录 await page.locator("a.quick-qq").click() await asyncio.sleep(1) # 等待 iframe 加载完成 await page.wait_for_selector("#ptlogin_iframe") # 切换到 iframe iframe = page.frame(name="ptlogin_iframe") # 通过 id 选择 "密码登录" 链接并点击 await iframe.locator("#switcher_plogin").click() await asyncio.sleep(1) # 填写账号 username_input = iframe.locator("#u") # 替换为实际的账号 for u in user: await username_input.type(u, no_wait_after=True) await asyncio.sleep(random.random() / 10) await asyncio.sleep(1) # 填写密码 password_input = iframe.locator("#p") # 替换为实际的密码 password = user_datas[user]["password"] for p in password: await password_input.type(p, no_wait_after=True) await asyncio.sleep(random.random() / 10) await asyncio.sleep(1) # 点击登录按钮 await iframe.locator("#login_button").click() await asyncio.sleep(1) # 这里检测安全验证 new_vcode_area = iframe.locator("div#newVcodeArea") style = await new_vcode_area.get_attribute("style") if style and "display: block" in style: if await new_vcode_area.get_by_text("安全验证").text_content() == "安全验证": logger.error(f"QQ号{user}需要安全验证, 登录失败，请使用其它账号类型") raise Exception(f"QQ号{user}需要安全验证, 登录失败，请使用其它账号类型") else: await page.get_by_text("账号密码登录").click() username_input = page.locator("#username") for u in user: await username_input.type(u, no_wait_after=True) await asyncio.sleep(random.random() / 10) password_input = page.locator("#pwd") password = user_datas[user]["password"] for p in password: await password_input.type(p, no_wait_after=True) await asyncio.sleep(random.random() / 10) await asyncio.sleep(random.random()) await page.locator('.policy_tip-checkbox').click() await asyncio.sleep(random.random()) await page.locator('.btn.J_ping.active').click() if user_datas[user].get("auto_switch", True): # 自动识别移动滑块验证码 await asyncio.sleep(1) await auto_move_slide(page, retry_times=30) # 自动验证形状验证码 await asyncio.sleep(1) await auto_shape(page, retry_times=30) # 进行短信验证识别 await asyncio.sleep(1) if await page.locator('text="手机短信验证"').count() != 0: logger.info("开始短信验证码识别环节") await sms_recognition(page, user, mode) # 进行手机语音验证识别 if await page.locator('div#header .text-header:has-text("手机语音验证")').count() > 0: logger.info("检测到手机语音验证页面,开始识别") await voice_verification(page, user, mode) # 弹窗检测 await check_dialog(page) # 检查警告,如账号存在风险或账密不正确等 await check_notice(page) else: logger.info("自动过验证码开关已关, 请手动操作") # 等待验证码通过 logger.info("等待获取cookie...") await page.wait_for_selector('#msShortcutMenu', state='visible', timeout=120000) cookies = await context.cookies() for cookie in cookies: if cookie['name'] == 'pt_key': pt_key = cookie["value"] return pt_key return None except Exception as e: traceback.print_exc() return None finally: await context.close() await browser.close() async def get_ql_api(ql_data): """ 封装了QL的登录 """ logger.info("开始获取QL登录态......") # 优化client_id和client_secret client_id = ql_data.get('client_id') client_secret = ql_data.get('client_secret') if client_id and client_secret: logger.info("使用client_id和client_secret登录......") qlapi = QlOpenApi(ql_data["url"]) response = await qlapi.login(client_id=client_id, client_secret=client_secret) if response['code'] == 200: logger.info("client_id和client_secret正常可用......") return qlapi else: logger.info("client_id和client_secret异常......") qlapi = QlApi(ql_data["url"]) # 其次用token token = ql_data.get('token') if token: logger.info("已设置TOKEN,开始检测TOKEN状态......") qlapi.login_by_token(token) # 如果token失效，就用账号密码登录 response = await qlapi.get_envs() if response['code'] == 401: logger.info("Token已失效, 正使用账号密码获取QL登录态......") response = await qlapi.login_by_username(ql_data.get("username"), ql_data.get("password")) if response['code'] != 200: logger.error(f"账号密码登录失败. response: {response}") raise Exception(f"账号密码登录失败. response: {response}") else: logger.info("Token正常可用......") else: # 最后用账号密码 logger.info("正使用账号密码获取QL登录态......") response = await qlapi.login_by_username(ql_data.get("username"), ql_data.get("password")) if response['code'] != 200: logger.error(f"账号密码登录失败. response: {response}") raise Exception(f"账号密码登录失败.response: {response}") return qlapi async def main(mode: str = None): """ :param mode 运行模式, 当mode = cron时，sms_func为 manual_input时，将自动传成no """ try: qlapi = await get_ql_api(qinglong_data) send_api = SendApi("ql") # 拿到禁用的用户列表 response = await qlapi.get_envs() if response['code'] == 200: logger.info("获取环境变量成功") else: logger.error(f"获取环境变量失败， response: {response}") raise Exception(f"获取环境变量失败， response: {response}") env_data = response['data'] # 获取值为JD_COOKIE的环境变量 jd_ck_env_datas = filter_cks(env_data, name='JD_COOKIE') # 从value中过滤出pt_pin, 注意只支持单行单pt_pin jd_ck_env_datas = [ {**x, 'pt_pin': extract_pt_pin(x['value'])} for x in jd_ck_env_datas if extract_pt_pin(x['value'])] try: logger.info("检测CK任务开始") # 先获取启用中的env_data up_jd_ck_list = filter_cks(jd_ck_env_datas, status=0, name='JD_COOKIE') # 这一步会去检测这些JD_COOKIE invalid_cks_id_list = await get_invalid_ck_ids(up_jd_ck_list) if invalid_cks_id_list: # 禁用QL的失效环境变量 ck_ids_datas = bytes(json.dumps(invalid_cks_id_list), 'utf-8') await qlapi.envs_disable(data=ck_ids_datas) # 更新jd_ck_env_datas jd_ck_env_datas = [{**x, 'status': 1} if x.get('id') in invalid_cks_id_list or x.get('_id') in invalid_cks_id_list else x for x in jd_ck_env_datas] logger.info("检测CK任务完成") except Exception as e: traceback.print_exc() logger.error(f"检测CK任务失败, 跳过检测, 报错原因为{e}") # 获取需强制更新pt_pin force_update_pt_pins = [user_datas[key]["pt_pin"] for key in user_datas if user_datas[key].get("force_update") is True] # 获取禁用和需要强制更新的users forbidden_users = [x for x in jd_ck_env_datas if (x['status'] == 1 or x['pt_pin'] in force_update_pt_pins)] if not forbidden_users: logger.info("所有COOKIE环境变量正常，无需更新") return # 获取需要的字段 filter_users_list = filter_forbidden_users(forbidden_users, ['_id', 'id', 'value', 'remarks', 'name']) # 生成字典 user_dict = get_forbidden_users_dict(filter_users_list, user_datas) if not user_dict: logger.info("失效的CK信息未配置在user_datas内，无需更新") return # 登录JD获取pt_key async with async_playwright() as playwright: for user in user_dict: logger.info(f"开始更新{desensitize_account(user, enable_desensitize)}") pt_key = await get_jd_pt_key(playwright, user, mode) if pt_key is None: logger.error(f"获取pt_key失败") await send_msg(send_api, send_type=1, msg=f"{desensitize_account(user, enable_desensitize)} 更新失败") continue req_data = user_dict[user] req_data["value"] = f"pt_key={pt_key};pt_pin={user_datas[user]['pt_pin']};" logger.info(f"更新内容为{req_data}") data = json.dumps(req_data) response = await qlapi.set_envs(data=data) if response['code'] == 200: logger.info(f"{desensitize_account(user, enable_desensitize)}更新成功") else: logger.error(f"{desensitize_account(user, enable_desensitize)}更新失败, response: {response}") await send_msg(send_api, send_type=1, msg=f"{desensitize_account(user, enable_desensitize)} 更新失败") continue req_id = f"[{req_data['id']}]" if 'id' in req_data.keys() else f'[\"{req_data["_id"]}\"]' data = bytes(req_id, 'utf-8') response = await qlapi.envs_enable(data=data) if response['code'] == 200: logger.info(f"{desensitize_account(user, enable_desensitize)}启用成功") await send_msg(send_api, send_type=0, msg=f"{desensitize_account(user, enable_desensitize)} 更新成功") else: logger.error(f"{desensitize_account(user, enable_desensitize)}启用失败, response: {response}") except Exception as e: traceback.print_exc() def parse_args(): """ 解析参数 """ parser = argparse.ArgumentParser() parser.add_argument('-m', '--mode', choices=['cron'], help="运行的main的模式(例如: 'cron')") return parser.parse_args() if __name__ == '__main__': # 使用解析参数的函数 args = parse_args() asyncio.run(main(mode=args.mode))