diff --git a/sau_backend/myUtils/auth.py b/sau_backend/myUtils/auth.py new file mode 100644 index 0000000..64155f9 --- /dev/null +++ b/sau_backend/myUtils/auth.py @@ -0,0 +1,103 @@ +import asyncio +import configparser +import os + +from playwright.async_api import async_playwright +from xhs import XhsClient + +from sau_backend.conf import BASE_DIR +from sau_backend.utils.base_social_media import set_init_script +from sau_backend.utils.log import tencent_logger, kuaishou_logger +from pathlib import Path +from sau_backend.uploader.xhs_uploader.main import sign_local + +async def cookie_auth_douyin(account_file): + async with async_playwright() as playwright: + browser = await playwright.chromium.launch(headless=True) + context = await browser.new_context(storage_state=account_file) + context = await set_init_script(context) + # 创建一个新的页面 + page = await context.new_page() + # 访问指定的 URL + await page.goto("https://creator.douyin.com/creator-micro/content/upload") + try: + await page.wait_for_url("https://creator.douyin.com/creator-micro/content/upload", timeout=5000) + except: + print("[+] 等待5秒 cookie 失效") + await context.close() + await browser.close() + return False + # 2024.06.17 抖音创作者中心改版 + if await page.get_by_text('手机号登录').count() or await page.get_by_text('扫码登录').count(): + print("[+] 等待5秒 cookie 失效") + return False + else: + print("[+] cookie 有效") + return True + +async def cookie_auth_tencent(account_file): + async with async_playwright() as playwright: + browser = await playwright.chromium.launch(headless=True) + context = await browser.new_context(storage_state=account_file) + context = await set_init_script(context) + # 创建一个新的页面 + page = await context.new_page() + # 访问指定的 URL + await page.goto("https://channels.weixin.qq.com/platform/post/create") + try: + await page.wait_for_selector('div.title-name:has-text("微信小店")', timeout=5000) # 等待5秒 + tencent_logger.error("[+] 等待5秒 cookie 失效") + return False + except: + tencent_logger.success("[+] cookie 有效") + return True + +async def cookie_auth_ks(account_file): + async with async_playwright() as playwright: + browser = await playwright.chromium.launch(headless=True) + context = await browser.new_context(storage_state=account_file) + context = await set_init_script(context) + # 创建一个新的页面 + page = await context.new_page() + # 访问指定的 URL + await page.goto("https://cp.kuaishou.com/article/publish/video") + try: + await page.wait_for_selector("div.names div.container div.name:text('机构服务')", timeout=5000) # 等待5秒 + + kuaishou_logger.info("[+] 等待5秒 cookie 失效") + return False + except: + kuaishou_logger.success("[+] cookie 有效") + return True + + + +async def check_cookie(type,file_path): + match type: + # 小红书 + case 1: + config = configparser.RawConfigParser() + config.read(Path(BASE_DIR / "cookiesFile" / file_path)) + cookies = config['account1']['cookies'] + xhs_client = XhsClient(cookies, sign=sign_local, timeout=60) + # auth cookie + # 注意:该校验cookie方式可能并没那么准确 + try: + xhs_client.get_video_first_frame_image_id("3214") + except: + print("cookie 失效") + return True + # 视频号 + case 2: + return await cookie_auth_tencent(Path(BASE_DIR / "cookiesFile" / file_path)) + # 抖音 + case 3: + return await cookie_auth_douyin(Path(BASE_DIR / "cookiesFile" / file_path)) + # 快手 + case 4: + return await cookie_auth_ks(Path(BASE_DIR / "cookiesFile" / file_path)) + case _: + return False + +# a = asyncio.run(check_cookie(4,"3a6cfdc0-3d51-11f0-8507-44e51723d63c.json")) +# print(a) \ No newline at end of file