138 lines
6.0 KiB
Python
138 lines
6.0 KiB
Python
import asyncio
|
|
import configparser
|
|
import os
|
|
|
|
from playwright.async_api import async_playwright
|
|
from xhs import XhsClient
|
|
|
|
from conf import BASE_DIR, LOCAL_CHROME_HEADLESS
|
|
from utils.base_social_media import set_init_script
|
|
from utils.log import tencent_logger, kuaishou_logger, douyin_logger
|
|
from pathlib import Path
|
|
from 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=LOCAL_CHROME_HEADLESS)
|
|
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)
|
|
# 2024.06.17 抖音创作者中心改版
|
|
# 判断
|
|
# 等待“扫码登录”元素出现，超时 5 秒（如果 5 秒没出现，说明 cookie 有效）
|
|
try:
|
|
await page.get_by_text("扫码登录").wait_for(timeout=5000)
|
|
douyin_logger.error("[+] cookie 失效，需要扫码登录")
|
|
return False
|
|
except:
|
|
douyin_logger.success("[+] cookie 有效")
|
|
return True
|
|
except:
|
|
douyin_logger.error("[+] 等待5秒 cookie 失效")
|
|
await context.close()
|
|
await browser.close()
|
|
return False
|
|
|
|
|
|
async def cookie_auth_tencent(account_file):
|
|
# 统一复用 uploader 的 cookie_auth（接口探测优先 + 页面兜底），
|
|
# 确保「账号测试」与「发布」使用同一套登录态判定口径
|
|
from uploader.tencent_uploader.main import cookie_auth as _tc_cookie_auth
|
|
return await _tc_cookie_auth(str(account_file))
|
|
|
|
|
|
async def cookie_auth_ks(account_file):
|
|
async with async_playwright() as playwright:
|
|
browser = await playwright.chromium.launch(headless=LOCAL_CHROME_HEADLESS)
|
|
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 cookie_auth_xhs(account_file):
|
|
async with async_playwright() as playwright:
|
|
browser = await playwright.chromium.launch(headless=LOCAL_CHROME_HEADLESS)
|
|
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.xiaohongshu.com/creator-micro/content/upload")
|
|
try:
|
|
await page.wait_for_url("https://creator.xiaohongshu.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 check_cookie(type, file_path):
|
|
match type:
|
|
# 小红书
|
|
case 1:
|
|
return await cookie_auth_xhs(Path(BASE_DIR / "cookiesFile" / file_path))
|
|
# 视频号
|
|
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 5:
|
|
from uploader.bilibili_uploader.runtime import run_biliup_command
|
|
account_file = Path(BASE_DIR / "cookiesFile" / file_path)
|
|
if not account_file.exists():
|
|
return False
|
|
result = run_biliup_command(["-u", str(account_file), "renew"])
|
|
return result.returncode == 0
|
|
case 6:
|
|
from uploader.baijiahao_uploader.main import cookie_auth as baijiahao_cookie_auth
|
|
return await baijiahao_cookie_auth(str(Path(BASE_DIR / "cookiesFile" / file_path)))
|
|
case 7:
|
|
from uploader.alipay_uploader.main import cookie_auth as alipay_cookie_auth
|
|
return await alipay_cookie_auth(str(Path(BASE_DIR / "cookiesFile" / file_path)))
|
|
case 8:
|
|
from uploader.weibo_uploader.main import cookie_auth as weibo_cookie_auth
|
|
return await weibo_cookie_auth(str(Path(BASE_DIR / "cookiesFile" / file_path)))
|
|
case 9:
|
|
from uploader.hupu_uploader.main import cookie_auth as hupu_cookie_auth
|
|
return await hupu_cookie_auth(str(Path(BASE_DIR / "cookiesFile" / file_path)))
|
|
case 10:
|
|
from uploader.youtube_uploader.main import cookie_auth as youtube_cookie_auth
|
|
return await youtube_cookie_auth(str(Path(BASE_DIR / "cookiesFile" / file_path)))
|
|
case 11:
|
|
from uploader.tk_uploader.main_chrome import cookie_auth as tiktok_cookie_auth
|
|
return await tiktok_cookie_auth(str(Path(BASE_DIR / "cookiesFile" / file_path)))
|
|
case _:
|
|
return False
|
|
|
|
# a = asyncio.run(check_cookie(1,"3a6cfdc0-3d51-11f0-8507-44e51723d63c.json"))
|
|
# print(a)
|