from abc import ABC, abstractmethod
|
from pathlib import Path
|
import os
|
import asyncio
|
from playwright.async_api import async_playwright, TimeoutError as PlaywrightTimeoutError
|
|
from auto_media_publisher.config.conf_base import COOKIE_DIR,PLAYWRIGHT_STEALTH_JS_PATH,DEFAULT_USER_AGENT
|
from auto_media_publisher.config.conf_platform_login_url import PLATFORM_LOGIN_MAP
|
from auto_media_publisher.utils.logger import get_logger
|
|
|
class LoginChrome(ABC):
|
def __init__(self, user,platform,proxy):
|
self.user = user
|
self.platform = platform
|
self.proxy = proxy
|
self.cookie_path = os.path.join(COOKIE_DIR, user, f"{platform}.json")
|
os.makedirs(os.path.dirname(self.cookie_path), exist_ok=True)
|
self.logger = get_logger(platform,user)
|
for k, v in PLATFORM_LOGIN_MAP.get(self.platform).items():
|
setattr(self, k, v)
|
|
@classmethod
|
def create_sub_class(cls, user,platform,proxy): # cls 相当于 self
|
return cls(user, platform, proxy)
|
|
|
async def _init_browser_context(self,playwright,headless=True):
|
"""
|
初始化 self.browser and self.context
|
如果有 self.proxy,设置
|
如果有 self.cookie_path ,设置
|
"""
|
browser_args = {"headless": headless,"channel": "chrome"}
|
|
if getattr(self, 'browser_addon_options', None):
|
browser_args.update(self.browser_addon_options)
|
print(f"start to init browser Using proxy: {self.proxy}")
|
if isinstance(self.proxy, str) and self.proxy.strip().lower().startswith(('http://', 'https://')):
|
browser_args["proxy"] = {"server": self.proxy}
|
self.browser = await playwright.chromium.launch(**browser_args)
|
if os.path.exists(self.cookie_path):
|
self.context = await self.browser.new_context(storage_state=f"{self.cookie_path}", user_agent=DEFAULT_USER_AGENT,viewport={"width": 1920, "height": 1080})
|
else:
|
self.context = await self.browser.new_context(user_agent=DEFAULT_USER_AGENT,viewport={"width": 1920, "height": 1080})
|
if os.path.exists(PLAYWRIGHT_STEALTH_JS_PATH):
|
await self.context.add_init_script(path=Path(PLAYWRIGHT_STEALTH_JS_PATH))
|
await self.context.grant_permissions(['geolocation'])
|
|
async def check_cookie_valid(self,if_hold_page:bool = False,headless:bool = True,login_url_time_out_ = 100000,login_element_time_out = 300000) -> bool:
|
"""
|
检查当前 cookie 是否有效（如是否已登录）
|
默认 if_hold_page = False，就是登录后即退出，不等待
|
默认 headless = False ，就是 不显示浏览器窗口
|
"""
|
async with async_playwright() as playwright:
|
await self._init_browser_context(playwright,headless)
|
# 创建一个新的页面
|
page = await self.context.new_page()
|
# 访问指定的 URL
|
try:
|
self.logger.info(f"{self.user}，正在打开{self.login_url}")
|
if if_hold_page:
|
login_url_time_out_ = 300000
|
await page.goto(self.login_url, timeout=login_url_time_out_)
|
except Exception as e:
|
self.logger.error(f"[login_chrome] 发生错误：{e}")
|
try:
|
locator = page.locator(self.login_element)
|
await locator.first.wait_for(timeout=login_element_time_out) # 等待至少一个元素出现
|
await self.context.storage_state(path=self.cookie_path)
|
self.logger.info(f"check_cookie_valid：{self.user}登录{self.platform}成功，cookie 已保存到 {self.cookie_path}")
|
if if_hold_page:
|
while True:
|
await asyncio.sleep(1)
|
|
return True
|
except PlaywrightTimeoutError:
|
self.logger.error("check_cookie_valid：{self.user}登录{self.platform}失败，等待跳转到目标页面超时，cookie 未保存")
|
return False
|
|
finally:
|
await self.browser.close()
|
|
|
async def run(self):
|
try:
|
self.logger.info(f'[+] 开始检测{self.user}在{self.platform}的登录信息cookie文件是否存在或过期！')
|
if not os.path.exists(self.cookie_path) or not await self.check_cookie_valid():
|
self.logger.info('[+] cookie文件不存在或已失效，即将自动打开浏览器，请扫码登录，登陆后会自动生成cookie文件')
|
await self.check_cookie_valid(if_hold_page=False,headless=False)
|
|
return True
|
except Exception as e:
|
self.logger.error(f"{self.user} login fails :{e}")
|
return False
|
|
|
async def open(self):
|
try:
|
await self.check_cookie_valid(if_hold_page=True,headless=False)
|
return True
|
except Exception as e:
|
self.logger.error(f"{self.user} login fails :{e}")
|
return False
|
