300 lines
9.7 KiB
Python
300 lines
9.7 KiB
Python
|
"""
|
|||
|
|
抖音认证实现
|
|||
|
|
"""
|
|||
|
|
import asyncio
|
|||
|
|
from typing import Optional
|
|||
|
|
|
|||
|
|
from playwright.async_api import Page
|
|||
|
|
|
|||
|
|
from .base_auth import BaseAuth
|
|||
|
|
from ..core.models import PlatformType, AccountInfo
|
|||
|
|
from ..config.platform_config import get_platform_url, get_selectors, get_wait_times
|
|||
|
|
from ..utils.logger import get_platform_logger
|
|||
|
|
from ..utils.exceptions import LoginFailedError, AuthenticationError
|
|||
|
|
|
|||
|
|
|
|||
|
|
class DouyinAuth(BaseAuth):
|
|||
|
|
"""抖音认证实现"""
|
|||
|
|
|
|||
|
|
def __init__(self):
|
|||
|
|
super().__init__(PlatformType.DOUYIN)
|
|||
|
|
self.login_url = get_platform_url(PlatformType.DOUYIN, "login")
|
|||
|
|
self.logger = get_platform_logger("douyin")
|
|||
|
|
|
|||
|
|
async def login(self, account_info: AccountInfo, headless: bool = False) -> bool:
|
|||
|
|
"""
|
|||
|
|
执行抖音登录流程
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
account_info: 账号信息
|
|||
|
|
headless: 是否使用无头模式
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
登录是否成功
|
|||
|
|
"""
|
|||
|
|
try:
|
|||
|
|
self.logger.info(f"开始抖音登录: {account_info.username}")
|
|||
|
|
|
|||
|
|
# 创建页面
|
|||
|
|
page = await self.create_page(account_info, headless)
|
|||
|
|
|
|||
|
|
# 尝试加载现有Cookie
|
|||
|
|
cookies_loaded = await self.load_cookies(page, account_info)
|
|||
|
|
|
|||
|
|
if cookies_loaded:
|
|||
|
|
# 验证Cookie是否有效
|
|||
|
|
if await self._validate_existing_session(page):
|
|||
|
|
self.logger.info("使用现有Cookie登录成功")
|
|||
|
|
return True
|
|||
|
|
else:
|
|||
|
|
self.logger.info("Cookie已失效，需要重新登录")
|
|||
|
|
|
|||
|
|
# 执行扫码登录
|
|||
|
|
success = await self._perform_qr_login(page, account_info)
|
|||
|
|
|
|||
|
|
if success:
|
|||
|
|
# 保存新的Cookie
|
|||
|
|
await self.save_cookies(page, account_info)
|
|||
|
|
self.logger.success("抖音登录完成")
|
|||
|
|
else:
|
|||
|
|
self.logger.failure("抖音登录失败")
|
|||
|
|
|
|||
|
|
# 关闭页面
|
|||
|
|
await page.close()
|
|||
|
|
|
|||
|
|
return success
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
self.logger.error(f"抖音登录异常: {e}")
|
|||
|
|
raise LoginFailedError(f"抖音登录失败: {e}", platform="douyin")
|
|||
|
|
|
|||
|
|
async def check_login_status(self, page: Page) -> bool:
|
|||
|
|
"""
|
|||
|
|
检查抖音登录状态
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
page: Playwright页面对象
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
是否已登录
|
|||
|
|
"""
|
|||
|
|
try:
|
|||
|
|
selectors = get_selectors(PlatformType.DOUYIN, "login", "login_success_indicators")
|
|||
|
|
|
|||
|
|
for selector in selectors:
|
|||
|
|
try:
|
|||
|
|
element = await page.wait_for_selector(selector, timeout=3000)
|
|||
|
|
if element and await element.is_visible():
|
|||
|
|
self.logger.debug(f"检测到登录状态指示器: {selector}")
|
|||
|
|
return True
|
|||
|
|
except:
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
# 检查URL是否包含登录相关页面
|
|||
|
|
current_url = page.url
|
|||
|
|
if "login" not in current_url.lower() and "auth" not in current_url.lower():
|
|||
|
|
# 尝试检测用户信息元素
|
|||
|
|
user_selectors = [
|
|||
|
|
"[data-testid='user-avatar']",
|
|||
|
|
".user-avatar img",
|
|||
|
|
".creator-info",
|
|||
|
|
"[class*='user'] [class*='avatar']",
|
|||
|
|
".header-user-info"
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
for selector in user_selectors:
|
|||
|
|
try:
|
|||
|
|
element = await page.wait_for_selector(selector, timeout=2000)
|
|||
|
|
if element and await element.is_visible():
|
|||
|
|
self.logger.debug(f"检测到用户信息元素: {selector}")
|
|||
|
|
return True
|
|||
|
|
except:
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
self.logger.warning(f"登录状态检查失败: {e}")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
async def _validate_existing_session(self, page: Page) -> bool:
|
|||
|
|
"""
|
|||
|
|
验证现有会话是否有效
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
page: Playwright页面对象
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
会话是否有效
|
|||
|
|
"""
|
|||
|
|
try:
|
|||
|
|
# 导航到创作者中心首页
|
|||
|
|
await self.safe_goto(page, self.login_url)
|
|||
|
|
await asyncio.sleep(3)
|
|||
|
|
|
|||
|
|
# 检查是否需要登录
|
|||
|
|
if await self.check_login_status(page):
|
|||
|
|
self.logger.debug("现有会话有效")
|
|||
|
|
return True
|
|||
|
|
else:
|
|||
|
|
self.logger.debug("现有会话无效，需要重新登录")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
self.logger.warning(f"会话验证失败: {e}")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
async def _perform_qr_login(self, page: Page, account_info: AccountInfo) -> bool:
|
|||
|
|
"""
|
|||
|
|
执行扫码登录流程
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
page: Playwright页面对象
|
|||
|
|
account_info: 账号信息
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
登录是否成功
|
|||
|
|
"""
|
|||
|
|
try:
|
|||
|
|
self.logger.info("开始扫码登录流程")
|
|||
|
|
|
|||
|
|
# 导航到登录页面
|
|||
|
|
if not await self.safe_goto(page, self.login_url):
|
|||
|
|
raise LoginFailedError("无法访问登录页面")
|
|||
|
|
|
|||
|
|
# 等待页面加载并检查是否已经有登录按钮
|
|||
|
|
await asyncio.sleep(3)
|
|||
|
|
|
|||
|
|
# 检查是否已经在登录页面，如果没有，寻找登录入口
|
|||
|
|
login_button_selectors = [
|
|||
|
|
"[data-testid='login-btn']",
|
|||
|
|
".login-btn",
|
|||
|
|
"button[class*='login']",
|
|||
|
|
"a[href*='login']"
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
login_button_found = False
|
|||
|
|
for selector in login_button_selectors:
|
|||
|
|
if await self.wait_for_element(page, selector, timeout=5000):
|
|||
|
|
try:
|
|||
|
|
await page.click(selector)
|
|||
|
|
await asyncio.sleep(2)
|
|||
|
|
login_button_found = True
|
|||
|
|
break
|
|||
|
|
except:
|
|||
|
|
continue
|
|||
|
|
|
|||
|
|
# 等待二维码加载
|
|||
|
|
qr_selectors = get_selectors(PlatformType.DOUYIN, "login", "qr_code")
|
|||
|
|
qr_code_loaded = False
|
|||
|
|
|
|||
|
|
for selector in qr_selectors:
|
|||
|
|
if await self.wait_for_element(page, selector, timeout=15000):
|
|||
|
|
qr_code_loaded = True
|
|||
|
|
self.logger.info("二维码已加载，请使用抖音App扫码")
|
|||
|
|
break
|
|||
|
|
|
|||
|
|
if not qr_code_loaded:
|
|||
|
|
# 尝试其他可能的二维码选择器
|
|||
|
|
alternative_selectors = [
|
|||
|
|
"img[src*='qr']",
|
|||
|
|
"canvas",
|
|||
|
|
".qrcode",
|
|||
|
|
"[class*='qr']",
|
|||
|
|
"[id*='qr']",
|
|||
|
|
".login-qr img"
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
for selector in alternative_selectors:
|
|||
|
|
if await self.wait_for_element(page, selector, timeout=5000):
|
|||
|
|
qr_code_loaded = True
|
|||
|
|
self.logger.info(f"检测到二维码元素: {selector}")
|
|||
|
|
break
|
|||
|
|
|
|||
|
|
if not qr_code_loaded:
|
|||
|
|
self.logger.warning("未检测到二维码，可能已经登录")
|
|||
|
|
# 检查是否已经登录
|
|||
|
|
if await self.check_login_status(page):
|
|||
|
|
return True
|
|||
|
|
else:
|
|||
|
|
raise LoginFailedError("二维码加载失败")
|
|||
|
|
|
|||
|
|
# 等待用户扫码登录
|
|||
|
|
wait_times = get_wait_times(PlatformType.DOUYIN, "login")
|
|||
|
|
login_timeout = wait_times.get("login_timeout", 300)
|
|||
|
|
|
|||
|
|
success = await self.wait_for_login_success(page, timeout=login_timeout)
|
|||
|
|
|
|||
|
|
if success:
|
|||
|
|
self.logger.info("扫码登录成功")
|
|||
|
|
return True
|
|||
|
|
else:
|
|||
|
|
raise LoginFailedError("扫码登录超时")
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
self.logger.error(f"扫码登录失败: {e}")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
async def force_relogin(self, account_info: AccountInfo, headless: bool = False) -> bool:
|
|||
|
|
"""
|
|||
|
|
强制重新登录
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
account_info: 账号信息
|
|||
|
|
headless: 是否使用无头模式
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
登录是否成功
|
|||
|
|
"""
|
|||
|
|
try:
|
|||
|
|
self.logger.info("强制重新登录")
|
|||
|
|
|
|||
|
|
# 删除现有Cookie
|
|||
|
|
self.delete_cookies(account_info)
|
|||
|
|
|
|||
|
|
# 执行登录
|
|||
|
|
return await self.login(account_info, headless)
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
self.logger.error(f"强制重新登录失败: {e}")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
async def test_login(self, account_info: AccountInfo) -> bool:
|
|||
|
|
"""
|
|||
|
|
测试登录状态
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
account_info: 账号信息
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
登录是否有效
|
|||
|
|
"""
|
|||
|
|
try:
|
|||
|
|
self.logger.info("测试登录状态")
|
|||
|
|
|
|||
|
|
# 创建页面
|
|||
|
|
page = await self.create_page(account_info, headless=True)
|
|||
|
|
|
|||
|
|
# 加载Cookie
|
|||
|
|
cookies_loaded = await self.load_cookies(page, account_info)
|
|||
|
|
|
|||
|
|
if not cookies_loaded:
|
|||
|
|
self.logger.info("没有找到Cookie文件")
|
|||
|
|
await page.close()
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
# 验证登录状态
|
|||
|
|
login_valid = await self._validate_existing_session(page)
|
|||
|
|
|
|||
|
|
await page.close()
|
|||
|
|
|
|||
|
|
if login_valid:
|
|||
|
|
self.logger.info("登录状态有效")
|
|||
|
|
return True
|
|||
|
|
else:
|
|||
|
|
self.logger.info("登录状态无效")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
self.logger.error(f"登录状态测试失败: {e}")
|
|||
|
|
return False
|