钟停为止
27bb20b206
## 背景与复现问题 当前架构中，许多用户即使在 `conf.py` 中配置了 `LOCAL_CHROME_HEADLESS=False` 以及自己的 `LOCAL_CHROME_PATH`，扫码登录时依然会调用自带的不隐身内核，这诱发了两个严重阻断： 1. 环境变量逃逸：因为散落在 `login.py` 等文件中大多只传了 `headless` 参数，并未传递 `executable_path`，导致外部浏览器配置彻底失效。 2. 反爬风控假死：系统直接抛出的 `webdriver=true` 会遭来抖音/小红书级别极其严厉的滑块甚至二维码阻断，最终卡死，后台报错 `监听页面跳转超时`。 ## 本次修复 1. **全局接管配置**：构建统一的浏览器环境变量拦截钩子 `utils/browser_hook.py`。 2. **防风控注入**：加入 `--disable-blink-features=AutomationControlled` 隐藏驱动器指纹，免受大数据识别拦截，极大提升扫码登入与发布的通过率。 3. **修复登录扫码挂载**：改写 `login.py` 对接，已验证能够丝滑拉起本地指纹浏览器。
321 lines
12 KiB
Python
321 lines
12 KiB
Python
import asyncio
|
|
import sqlite3
|
|
|
|
from playwright.async_api import async_playwright
|
|
|
|
from myUtils.auth import check_cookie
|
|
from utils.base_social_media import set_init_script
|
|
import uuid
|
|
from pathlib import Path
|
|
from conf import BASE_DIR, LOCAL_CHROME_HEADLESS, LOCAL_CHROME_PATH
|
|
|
|
# 统一获取浏览器启动配置（防风控+引入本地浏览器）
|
|
def get_browser_options():
|
|
options = {
|
|
'headless': LOCAL_CHROME_HEADLESS,
|
|
'args': [
|
|
'--disable-blink-features=AutomationControlled', # 核心防爬屏蔽：去掉 window.navigator.webdriver 标签
|
|
'--lang=zh-CN',
|
|
'--disable-infobars',
|
|
'--start-maximized'
|
|
]
|
|
}
|
|
# 如果用户在 conf.py 里配置了本地 Chrome，就用本地的，这样成功率极高
|
|
if LOCAL_CHROME_PATH:
|
|
options['executable_path'] = LOCAL_CHROME_PATH
|
|
|
|
return options
|
|
|
|
# 抖音登录
|
|
async def douyin_cookie_gen(id,status_queue):
|
|
url_changed_event = asyncio.Event()
|
|
async def on_url_change():
|
|
# 检查是否是主框架的变化
|
|
if page.url != original_url:
|
|
url_changed_event.set()
|
|
async with async_playwright() as playwright:
|
|
options = get_browser_options()
|
|
# Make sure to run headed.
|
|
browser = await playwright.chromium.launch(**options)
|
|
# Setup context however you like.
|
|
context = await browser.new_context() # Pass any options
|
|
context = await set_init_script(context)
|
|
# Pause the page, and start recording manually.
|
|
page = await context.new_page()
|
|
await page.goto("https://creator.douyin.com/")
|
|
original_url = page.url
|
|
img_locator = page.get_by_role("img", name="二维码")
|
|
# 获取 src 属性值
|
|
src = await img_locator.get_attribute("src")
|
|
print("✅ 图片地址:", src)
|
|
status_queue.put(src)
|
|
# 监听页面的 'framenavigated' 事件，只关注主框架的变化
|
|
page.on('framenavigated',
|
|
lambda frame: asyncio.create_task(on_url_change()) if frame == page.main_frame else None)
|
|
try:
|
|
# 等待 URL 变化或超时
|
|
await asyncio.wait_for(url_changed_event.wait(), timeout=200) # 最多等待 200 秒
|
|
print("监听页面跳转成功")
|
|
except asyncio.TimeoutError:
|
|
print("监听页面跳转超时")
|
|
await page.close()
|
|
await context.close()
|
|
await browser.close()
|
|
status_queue.put("500")
|
|
return None
|
|
uuid_v1 = uuid.uuid1()
|
|
print(f"UUID v1: {uuid_v1}")
|
|
# 确保cookiesFile目录存在
|
|
cookies_dir = Path(BASE_DIR / "cookiesFile")
|
|
cookies_dir.mkdir(exist_ok=True)
|
|
await context.storage_state(path=cookies_dir / f"{uuid_v1}.json")
|
|
result = await check_cookie(3, f"{uuid_v1}.json")
|
|
if not result:
|
|
status_queue.put("500")
|
|
await page.close()
|
|
await context.close()
|
|
await browser.close()
|
|
return None
|
|
await page.close()
|
|
await context.close()
|
|
await browser.close()
|
|
with sqlite3.connect(Path(BASE_DIR / "db" / "database.db")) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute('''
|
|
INSERT INTO user_info (type, filePath, userName, status)
|
|
VALUES (?, ?, ?, ?)
|
|
''', (3, f"{uuid_v1}.json", id, 1))
|
|
conn.commit()
|
|
print("✅ 用户状态已记录")
|
|
status_queue.put("200")
|
|
|
|
|
|
# 视频号登录
|
|
async def get_tencent_cookie(id,status_queue):
|
|
url_changed_event = asyncio.Event()
|
|
async def on_url_change():
|
|
# 检查是否是主框架的变化
|
|
if page.url != original_url:
|
|
url_changed_event.set()
|
|
|
|
async with async_playwright() as playwright:
|
|
options = {
|
|
'args': [
|
|
'--lang en-GB'
|
|
],
|
|
'headless': LOCAL_CHROME_HEADLESS, # Set headless option here
|
|
}
|
|
# Make sure to run headed.
|
|
browser = await playwright.chromium.launch(**options)
|
|
# Setup context however you like.
|
|
context = await browser.new_context() # Pass any options
|
|
# Pause the page, and start recording manually.
|
|
context = await set_init_script(context)
|
|
page = await context.new_page()
|
|
await page.goto("https://channels.weixin.qq.com")
|
|
original_url = page.url
|
|
|
|
# 监听页面的 'framenavigated' 事件，只关注主框架的变化
|
|
page.on('framenavigated',
|
|
lambda frame: asyncio.create_task(on_url_change()) if frame == page.main_frame else None)
|
|
|
|
# 等待 iframe 出现（最多等 60 秒）
|
|
iframe_locator = page.frame_locator("iframe").first
|
|
|
|
# 获取 iframe 中的第一个 img 元素
|
|
img_locator = iframe_locator.get_by_role("img").first
|
|
|
|
# 获取 src 属性值
|
|
src = await img_locator.get_attribute("src")
|
|
print("✅ 图片地址:", src)
|
|
status_queue.put(src)
|
|
|
|
try:
|
|
# 等待 URL 变化或超时
|
|
await asyncio.wait_for(url_changed_event.wait(), timeout=200) # 最多等待 200 秒
|
|
print("监听页面跳转成功")
|
|
except asyncio.TimeoutError:
|
|
status_queue.put("500")
|
|
print("监听页面跳转超时")
|
|
await page.close()
|
|
await context.close()
|
|
await browser.close()
|
|
return None
|
|
uuid_v1 = uuid.uuid1()
|
|
print(f"UUID v1: {uuid_v1}")
|
|
# 确保cookiesFile目录存在
|
|
cookies_dir = Path(BASE_DIR / "cookiesFile")
|
|
cookies_dir.mkdir(exist_ok=True)
|
|
await context.storage_state(path=cookies_dir / f"{uuid_v1}.json")
|
|
result = await check_cookie(2,f"{uuid_v1}.json")
|
|
if not result:
|
|
status_queue.put("500")
|
|
await page.close()
|
|
await context.close()
|
|
await browser.close()
|
|
return None
|
|
await page.close()
|
|
await context.close()
|
|
await browser.close()
|
|
|
|
with sqlite3.connect(Path(BASE_DIR / "db" / "database.db")) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute('''
|
|
INSERT INTO user_info (type, filePath, userName, status)
|
|
VALUES (?, ?, ?, ?)
|
|
''', (2, f"{uuid_v1}.json", id, 1))
|
|
conn.commit()
|
|
print("✅ 用户状态已记录")
|
|
status_queue.put("200")
|
|
|
|
# 快手登录
|
|
async def get_ks_cookie(id,status_queue):
|
|
url_changed_event = asyncio.Event()
|
|
async def on_url_change():
|
|
# 检查是否是主框架的变化
|
|
if page.url != original_url:
|
|
url_changed_event.set()
|
|
async with async_playwright() as playwright:
|
|
options = {
|
|
'args': [
|
|
'--lang en-GB'
|
|
],
|
|
'headless': LOCAL_CHROME_HEADLESS, # Set headless option here
|
|
}
|
|
# Make sure to run headed.
|
|
browser = await playwright.chromium.launch(**options)
|
|
# Setup context however you like.
|
|
context = await browser.new_context() # Pass any options
|
|
context = await set_init_script(context)
|
|
# Pause the page, and start recording manually.
|
|
page = await context.new_page()
|
|
await page.goto("https://cp.kuaishou.com")
|
|
|
|
# 定位并点击“立即登录”按钮（类型为 link）
|
|
await page.get_by_role("link", name="立即登录").click()
|
|
await page.get_by_text("扫码登录").click()
|
|
img_locator = page.get_by_role("img", name="qrcode")
|
|
# 获取 src 属性值
|
|
src = await img_locator.get_attribute("src")
|
|
original_url = page.url
|
|
print("✅ 图片地址:", src)
|
|
status_queue.put(src)
|
|
# 监听页面的 'framenavigated' 事件，只关注主框架的变化
|
|
page.on('framenavigated',
|
|
lambda frame: asyncio.create_task(on_url_change()) if frame == page.main_frame else None)
|
|
|
|
try:
|
|
# 等待 URL 变化或超时
|
|
await asyncio.wait_for(url_changed_event.wait(), timeout=200) # 最多等待 200 秒
|
|
print("监听页面跳转成功")
|
|
except asyncio.TimeoutError:
|
|
status_queue.put("500")
|
|
print("监听页面跳转超时")
|
|
await page.close()
|
|
await context.close()
|
|
await browser.close()
|
|
return None
|
|
uuid_v1 = uuid.uuid1()
|
|
print(f"UUID v1: {uuid_v1}")
|
|
# 确保cookiesFile目录存在
|
|
cookies_dir = Path(BASE_DIR / "cookiesFile")
|
|
cookies_dir.mkdir(exist_ok=True)
|
|
await context.storage_state(path=cookies_dir / f"{uuid_v1}.json")
|
|
result = await check_cookie(4, f"{uuid_v1}.json")
|
|
if not result:
|
|
status_queue.put("500")
|
|
await page.close()
|
|
await context.close()
|
|
await browser.close()
|
|
return None
|
|
await page.close()
|
|
await context.close()
|
|
await browser.close()
|
|
|
|
with sqlite3.connect(Path(BASE_DIR / "db" / "database.db")) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute('''
|
|
INSERT INTO user_info (type, filePath, userName, status)
|
|
VALUES (?, ?, ?, ?)
|
|
''', (4, f"{uuid_v1}.json", id, 1))
|
|
conn.commit()
|
|
print("✅ 用户状态已记录")
|
|
status_queue.put("200")
|
|
|
|
# 小红书登录
|
|
async def xiaohongshu_cookie_gen(id,status_queue):
|
|
url_changed_event = asyncio.Event()
|
|
|
|
async def on_url_change():
|
|
# 检查是否是主框架的变化
|
|
if page.url != original_url:
|
|
url_changed_event.set()
|
|
|
|
async with async_playwright() as playwright:
|
|
options = {
|
|
'args': [
|
|
'--lang en-GB'
|
|
],
|
|
'headless': LOCAL_CHROME_HEADLESS, # Set headless option here
|
|
}
|
|
# Make sure to run headed.
|
|
browser = await playwright.chromium.launch(**options)
|
|
# Setup context however you like.
|
|
context = await browser.new_context() # Pass any options
|
|
context = await set_init_script(context)
|
|
# Pause the page, and start recording manually.
|
|
page = await context.new_page()
|
|
await page.goto("https://creator.xiaohongshu.com/")
|
|
await page.locator('img.css-wemwzq').click()
|
|
|
|
img_locator = page.get_by_role("img").nth(2)
|
|
# 获取 src 属性值
|
|
src = await img_locator.get_attribute("src")
|
|
original_url = page.url
|
|
print("✅ 图片地址:", src)
|
|
status_queue.put(src)
|
|
# 监听页面的 'framenavigated' 事件，只关注主框架的变化
|
|
page.on('framenavigated',
|
|
lambda frame: asyncio.create_task(on_url_change()) if frame == page.main_frame else None)
|
|
|
|
try:
|
|
# 等待 URL 变化或超时
|
|
await asyncio.wait_for(url_changed_event.wait(), timeout=200) # 最多等待 200 秒
|
|
print("监听页面跳转成功")
|
|
except asyncio.TimeoutError:
|
|
status_queue.put("500")
|
|
print("监听页面跳转超时")
|
|
await page.close()
|
|
await context.close()
|
|
await browser.close()
|
|
return None
|
|
uuid_v1 = uuid.uuid1()
|
|
print(f"UUID v1: {uuid_v1}")
|
|
# 确保cookiesFile目录存在
|
|
cookies_dir = Path(BASE_DIR / "cookiesFile")
|
|
cookies_dir.mkdir(exist_ok=True)
|
|
await context.storage_state(path=cookies_dir / f"{uuid_v1}.json")
|
|
result = await check_cookie(1, f"{uuid_v1}.json")
|
|
if not result:
|
|
status_queue.put("500")
|
|
await page.close()
|
|
await context.close()
|
|
await browser.close()
|
|
return None
|
|
await page.close()
|
|
await context.close()
|
|
await browser.close()
|
|
|
|
with sqlite3.connect(Path(BASE_DIR / "db" / "database.db")) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute('''
|
|
INSERT INTO user_info (type, filePath, userName, status)
|
|
VALUES (?, ?, ?, ?)
|
|
''', (1, f"{uuid_v1}.json", id, 1))
|
|
conn.commit()
|
|
print("✅ 用户状态已记录")
|
|
status_queue.put("200")
|
|
|
|
# a = asyncio.run(xiaohongshu_cookie_gen(4,None))
|
|
# print(a)
|