DerekEXS
dff5a3d7ea
Two related changes for kuaishou uploader: - Default browser channel is switched from 'chrome' to 'chromium'. The launcher binary that ships with patchright no longer reliably resolves the system Chrome channel on every platform, so falling back to bundled chromium avoids launch-time channel-resolution errors. The user can still force a real Chrome via conf.py:LOCAL_CHROME_PATH. - New cdp_url parameter on ks_setup / get_ks_cookie. When set, the login flow uses connect_over_cdp() to drive the user's already-running real Chrome instead of spawning a separate browser. This is the supported way around strict anti-bot detection: real Chrome has a fingerprint that passes verification, while the bundled chromium does not. should_close_context tracks ownership so we don't close a context we didn't create.
735 lines
30 KiB
Python
735 lines
30 KiB
Python
# -*- coding: utf-8 -*-
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import inspect
|
|
import os
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from patchright.async_api import Page
|
|
from patchright.async_api import Playwright
|
|
from patchright.async_api import async_playwright
|
|
|
|
from conf import DEBUG_MODE, LOCAL_CHROME_HEADLESS, LOCAL_CHROME_PATH
|
|
from uploader.base_video import BaseVideoUploader
|
|
from utils.base_social_media import set_init_script
|
|
from utils.files_times import get_absolute_path
|
|
from utils.login_qrcode import build_login_qrcode_path
|
|
from utils.login_qrcode import decode_qrcode_from_path
|
|
from utils.login_qrcode import print_terminal_qrcode
|
|
from utils.login_qrcode import remove_qrcode_file
|
|
from utils.login_qrcode import save_data_url_image
|
|
from utils.log import kuaishou_logger
|
|
|
|
KUAISHOU_UPLOAD_URL = "https://cp.kuaishou.com/article/publish/video"
|
|
KUAISHOU_MANAGE_URL = "https://cp.kuaishou.com/article/manage/video?status=2&from=publish"
|
|
KUAISHOU_LOGIN_URL = "https://passport.kuaishou.com/pc/account/login/?sid=kuaishou.web.cp.api&callback=https%3A%2F%2Fcp.kuaishou.com%2Frest%2Finfra%2Fsts%3FfollowUrl%3Dhttps%253A%252F%252Fcp.kuaishou.com%252Farticle%252Fpublish%252Fvideo%26setRootDomain%3Dtrue"
|
|
KUAISHOU_UPLOAD_URL_PATTERN = "**/article/publish/video**"
|
|
KUAISHOU_MANAGE_URL_PATTERN = "**/article/manage/video?status=2&from=publish**"
|
|
KUAISHOU_COOKIE_INVALID_SELECTOR = "div.names div.container div.name:text('机构服务')"
|
|
KUAISHOU_PUBLISH_STRATEGY_IMMEDIATE = "immediate"
|
|
KUAISHOU_PUBLISH_STRATEGY_SCHEDULED = "scheduled"
|
|
|
|
|
|
def _msg(emoji: str, text: str) -> str:
|
|
return f"{emoji} {text}"
|
|
|
|
|
|
def _print_ks_qrcode(qrcode_content: str, qrcode_path: Path) -> None:
|
|
try:
|
|
print_terminal_qrcode(qrcode_content, qrcode_path, "快手APP", compact=False, border=2)
|
|
except TypeError as exc:
|
|
if "unexpected keyword argument 'compact'" not in str(exc):
|
|
raise
|
|
kuaishou_logger.warning(_msg("😵", "检测到旧版二维码打印函数，小人切回兼容模式继续登录"))
|
|
print_terminal_qrcode(qrcode_content, qrcode_path, "快手APP")
|
|
|
|
|
|
async def _emit_qrcode_callback(qrcode_callback, payload: dict):
|
|
if not qrcode_callback:
|
|
return
|
|
|
|
callback_result = qrcode_callback(payload)
|
|
if inspect.isawaitable(callback_result):
|
|
await callback_result
|
|
|
|
|
|
def _build_login_result(
|
|
success: bool,
|
|
status: str,
|
|
message: str,
|
|
account_file: str,
|
|
qrcode: dict | None = None,
|
|
current_url: str = "",
|
|
) -> dict:
|
|
return {
|
|
"success": success,
|
|
"status": status,
|
|
"message": message,
|
|
"account_file": str(account_file),
|
|
"qrcode": qrcode,
|
|
"current_url": current_url,
|
|
}
|
|
|
|
|
|
async def _is_ks_cookie_invalid(page: Page, timeout: int = 5000) -> bool:
|
|
try:
|
|
await page.wait_for_selector(KUAISHOU_COOKIE_INVALID_SELECTOR, timeout=timeout)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
async def _extract_ks_qrcode_src(page: Page) -> str:
|
|
login_form = page.locator("main#login-form").first
|
|
await login_form.wait_for(state="visible", timeout=30000)
|
|
|
|
qrcode_img = login_form.locator('div.qr-login img[alt="qrcode"]').first
|
|
try:
|
|
if not await qrcode_img.count() or not await qrcode_img.is_visible():
|
|
platform_switch = login_form.locator("div.platform-switch").first
|
|
await platform_switch.wait_for(state="visible", timeout=10000)
|
|
await platform_switch.click()
|
|
await asyncio.sleep(1)
|
|
except Exception:
|
|
platform_switch = login_form.locator("div.platform-switch").first
|
|
await platform_switch.wait_for(state="visible", timeout=10000)
|
|
await platform_switch.click()
|
|
await asyncio.sleep(1)
|
|
|
|
await qrcode_img.wait_for(state="visible", timeout=15000)
|
|
|
|
qrcode_src = await qrcode_img.get_attribute("src")
|
|
if not qrcode_src:
|
|
raise RuntimeError("未获取到快手登录二维码地址")
|
|
|
|
return qrcode_src
|
|
|
|
|
|
async def _save_ks_qrcode(page: Page, account_file: str, previous_qrcode_path: Path | None = None, qrcode_callback=None) -> dict:
|
|
qrcode_src = await _extract_ks_qrcode_src(page)
|
|
qrcode_path = save_data_url_image(qrcode_src, build_login_qrcode_path(account_file, suffix="ks_login_qrcode"))
|
|
|
|
if previous_qrcode_path and previous_qrcode_path != qrcode_path:
|
|
if remove_qrcode_file(previous_qrcode_path):
|
|
kuaishou_logger.info(_msg("🧹", f"临时二维码文件已清理: {previous_qrcode_path}"))
|
|
|
|
kuaishou_logger.info(_msg("🖼️", f"二维码已经准备好啦，已保存到: {qrcode_path}"))
|
|
qrcode_content = decode_qrcode_from_path(qrcode_path)
|
|
if qrcode_content:
|
|
_print_ks_qrcode(qrcode_content, qrcode_path)
|
|
else:
|
|
kuaishou_logger.warning(_msg("😵", f"终端没法完整显示二维码，请打开 {qrcode_path} 扫码"))
|
|
|
|
qrcode_info = {
|
|
"image_path": str(qrcode_path),
|
|
"image_data_url": qrcode_src,
|
|
}
|
|
await _emit_qrcode_callback(qrcode_callback, qrcode_info)
|
|
return qrcode_info
|
|
|
|
|
|
async def _is_ks_qrcode_expired(page: Page) -> bool:
|
|
expired_box = page.locator("div.qrcode-status.qrcode-status-timeout").first
|
|
try:
|
|
if not await expired_box.count():
|
|
return False
|
|
return await expired_box.is_visible()
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
async def _is_ks_login_page_gone(page: Page) -> bool:
|
|
try:
|
|
login_form = page.locator("main#login-form").first
|
|
if not await login_form.count():
|
|
return True
|
|
return not await login_form.is_visible()
|
|
except Exception:
|
|
return True
|
|
|
|
|
|
async def cookie_auth(account_file):
|
|
async with async_playwright() as playwright:
|
|
if LOCAL_CHROME_PATH:
|
|
browser = await playwright.chromium.launch(headless=True, executable_path=LOCAL_CHROME_PATH)
|
|
else:
|
|
browser = await playwright.chromium.launch(headless=True, channel="chromium")
|
|
try:
|
|
context = await browser.new_context(storage_state=account_file)
|
|
context = await set_init_script(context)
|
|
page = await context.new_page()
|
|
await page.goto(KUAISHOU_UPLOAD_URL)
|
|
if await _is_ks_cookie_invalid(page):
|
|
kuaishou_logger.info(_msg("🥹", "cookie 已失效，得重新登录一下"))
|
|
return False
|
|
|
|
kuaishou_logger.success(_msg("🥳", "cookie 有效"))
|
|
return True
|
|
except Exception as exc:
|
|
kuaishou_logger.warning(_msg("😵", f"cookie 校验时出错，按失效处理: {exc}"))
|
|
return False
|
|
finally:
|
|
await browser.close()
|
|
|
|
|
|
async def ks_setup(account_file, handle=False, return_detail=False, qrcode_callback=None, headless: bool = LOCAL_CHROME_HEADLESS, cdp_url: str | None = None):
|
|
account_file = get_absolute_path(account_file, "ks_uploader")
|
|
if not os.path.exists(account_file) or not await cookie_auth(account_file):
|
|
if not handle:
|
|
result = _build_login_result(False, "cookie_invalid", "cookie文件不存在或已失效", account_file)
|
|
return result if return_detail else False
|
|
kuaishou_logger.info(_msg("🥹", "cookie 失效了，准备重新登录快手创作者平台"))
|
|
result = await get_ks_cookie(account_file, qrcode_callback=qrcode_callback, headless=headless, cdp_url=cdp_url)
|
|
return result if return_detail else result["success"]
|
|
|
|
result = _build_login_result(True, "cookie_valid", "cookie有效", account_file)
|
|
return result if return_detail else True
|
|
|
|
|
|
async def get_ks_cookie(
|
|
account_file,
|
|
qrcode_callback=None,
|
|
headless: bool = LOCAL_CHROME_HEADLESS,
|
|
poll_interval: int = 3,
|
|
max_checks: int = 100,
|
|
cdp_url: str | None = None,
|
|
):
|
|
if headless:
|
|
kuaishou_logger.info(_msg("🖼️", "快手登录将以无头模式运行，小人会输出终端二维码并保存本地二维码图片"))
|
|
|
|
async with async_playwright() as playwright:
|
|
if cdp_url:
|
|
browser = await playwright.chromium.connect_over_cdp(cdp_url)
|
|
context = browser.contexts[0] if browser.contexts else await browser.new_context()
|
|
should_close_context = False
|
|
else:
|
|
if LOCAL_CHROME_PATH:
|
|
browser = await playwright.chromium.launch(headless=headless, executable_path=LOCAL_CHROME_PATH)
|
|
else:
|
|
browser = await playwright.chromium.launch(headless=headless, channel="chromium")
|
|
context = await browser.new_context()
|
|
should_close_context = True
|
|
context = await set_init_script(context)
|
|
qrcode_path = None
|
|
qrcode_info = None
|
|
result = _build_login_result(False, "failed", "快手登录失败", account_file)
|
|
try:
|
|
page = await context.new_page()
|
|
await page.goto(KUAISHOU_LOGIN_URL)
|
|
kuaishou_logger.info(_msg("🧍", "请在浏览器里扫码登录快手，小人正在耐心等待"))
|
|
|
|
qrcode_info = await _save_ks_qrcode(page, account_file, qrcode_callback=qrcode_callback)
|
|
qrcode_path = Path(qrcode_info["image_path"])
|
|
|
|
for _ in range(max_checks):
|
|
if page.url.startswith(KUAISHOU_UPLOAD_URL) or await _is_ks_login_page_gone(page):
|
|
await context.storage_state(path=account_file)
|
|
if await cookie_auth(account_file):
|
|
kuaishou_logger.success(_msg("🥳", "快手扫码登录成功，小人开心收工"))
|
|
result = _build_login_result(True, "success", "快手扫码登录成功", account_file, qrcode_info, page.url)
|
|
else:
|
|
kuaishou_logger.error(_msg("😢", "快手扫码完成了，但 cookie 校验失败"))
|
|
result = _build_login_result(
|
|
False,
|
|
"cookie_invalid",
|
|
"快手扫码流程结束，但 cookie 校验失败",
|
|
account_file,
|
|
qrcode_info,
|
|
page.url,
|
|
)
|
|
return result
|
|
|
|
if qrcode_info and await _is_ks_qrcode_expired(page):
|
|
kuaishou_logger.warning(_msg("😵", "二维码失效了，小人马上去刷新"))
|
|
refresh_button = page.locator("p.qrcode-refresh").first
|
|
if await refresh_button.count():
|
|
await refresh_button.click()
|
|
await asyncio.sleep(1)
|
|
qrcode_info = await _save_ks_qrcode(
|
|
page,
|
|
account_file,
|
|
qrcode_path,
|
|
qrcode_callback=qrcode_callback,
|
|
)
|
|
qrcode_path = Path(qrcode_info["image_path"])
|
|
|
|
await asyncio.sleep(poll_interval)
|
|
|
|
result = _build_login_result(
|
|
False,
|
|
"timeout",
|
|
"等待快手扫码登录超时",
|
|
account_file,
|
|
qrcode_info,
|
|
page.url,
|
|
)
|
|
except Exception as exc:
|
|
result = _build_login_result(False, "failed", str(exc), account_file, current_url=page.url if "page" in locals() else "")
|
|
finally:
|
|
if remove_qrcode_file(qrcode_path):
|
|
kuaishou_logger.info(_msg("🧹", f"临时二维码文件已清理: {qrcode_path}"))
|
|
if not result["success"]:
|
|
kuaishou_logger.error(_msg("😢", f"登录失败: {result['message']}"))
|
|
if should_close_context:
|
|
await context.close()
|
|
await browser.close()
|
|
|
|
return result
|
|
|
|
|
|
class KSBaseUploader(BaseVideoUploader):
|
|
def __init__(
|
|
self,
|
|
publish_date: datetime | int,
|
|
account_file,
|
|
publish_strategy: str | None = None,
|
|
debug: bool = DEBUG_MODE,
|
|
headless: bool = LOCAL_CHROME_HEADLESS,
|
|
):
|
|
self.publish_date = publish_date
|
|
self.account_file = str(account_file)
|
|
self.publish_strategy = publish_strategy
|
|
self.debug = debug
|
|
self.headless = headless
|
|
self.local_executable_path = LOCAL_CHROME_PATH
|
|
self.date_format = "%Y-%m-%d %H:%M"
|
|
|
|
async def validate_base_args(self):
|
|
if not os.path.exists(self.account_file):
|
|
raise RuntimeError(f"cookie文件不存在，请先完成快手登录: {self.account_file}")
|
|
if not await cookie_auth(self.account_file):
|
|
raise RuntimeError(f"cookie文件已失效，请先完成快手登录: {self.account_file}")
|
|
|
|
if self.publish_strategy is None:
|
|
self.publish_strategy = (
|
|
KUAISHOU_PUBLISH_STRATEGY_SCHEDULED
|
|
if self.publish_date != 0
|
|
else KUAISHOU_PUBLISH_STRATEGY_IMMEDIATE
|
|
)
|
|
|
|
if self.publish_strategy not in {
|
|
KUAISHOU_PUBLISH_STRATEGY_IMMEDIATE,
|
|
KUAISHOU_PUBLISH_STRATEGY_SCHEDULED,
|
|
}:
|
|
raise ValueError(f"不支持的发布策略: {self.publish_strategy}")
|
|
|
|
if self.publish_strategy == KUAISHOU_PUBLISH_STRATEGY_SCHEDULED:
|
|
self.publish_date = self.validate_publish_date(self.publish_date)
|
|
else:
|
|
self.publish_date = 0
|
|
|
|
async def set_schedule_time(self, page: Page, publish_date: datetime):
|
|
kuaishou_logger.info(_msg("🕒", "小人准备设置定时发布时间"))
|
|
publish_date_str = publish_date.strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
# 1. 切换到"定时发布"radio (用文本匹配更稳)
|
|
await page.locator('label.ant-radio-wrapper').filter(has_text="定时发布").click()
|
|
await asyncio.sleep(2)
|
|
|
|
# 2. 点击 picker 打开下拉面板
|
|
await page.locator('input[placeholder="选择日期时间"]').click()
|
|
await asyncio.sleep(1)
|
|
|
|
# 3. 用 React 兼容的方式直接设置 input 的 value
|
|
# (ant-design DatePicker 是 controlled component, 必须用 native setter + bubbling event)
|
|
js_code = """
|
|
(newValue) => {
|
|
const input = document.querySelector('input[placeholder="选择日期时间"]');
|
|
if (!input) return false;
|
|
const nativeSetter = Object.getOwnPropertyDescriptor(
|
|
window.HTMLInputElement.prototype, 'value'
|
|
).set;
|
|
nativeSetter.call(input, newValue);
|
|
input.dispatchEvent(new Event('input', { bubbles: true }));
|
|
input.dispatchEvent(new Event('change', { bubbles: true }));
|
|
return true;
|
|
}
|
|
"""
|
|
ok = await page.evaluate(js_code, publish_date_str)
|
|
if not ok:
|
|
kuaishou_logger.error("❌ 找不到时间选择器输入框")
|
|
return
|
|
|
|
await asyncio.sleep(1)
|
|
# 4. 按 Enter 确认
|
|
await page.keyboard.press("Enter")
|
|
await asyncio.sleep(2)
|
|
kuaishou_logger.info(f"✅ 定时发布时间已设置为 {publish_date_str}")
|
|
|
|
async def close_guide_overlay(self, page: Page) -> bool:
|
|
joyride_tooltip = page.locator('div[id^="react-joyride-step"] div[role="alertdialog"]')
|
|
|
|
# 判断是否显示
|
|
if await joyride_tooltip.count() > 0 and await joyride_tooltip.first.is_visible():
|
|
print("检测到 Joyride 引导遮罩，正在关闭...")
|
|
|
|
# 点击关闭按钮（X），使用多个可靠特征
|
|
close_button = page.locator('div[role="alertdialog"]').locator(
|
|
'[aria-label="Skip"], [data-action="skip"], button[title="Skip"]'
|
|
)
|
|
|
|
await close_button.click(force=True)
|
|
|
|
# 等待遮罩消失
|
|
await joyride_tooltip.wait_for(state="hidden", timeout=5000)
|
|
|
|
print("✅ 已关闭 Joyride 遮罩")
|
|
else:
|
|
print("未检测到 Joyride 遮罩，继续执行")
|
|
|
|
|
|
class KSVideo(KSBaseUploader):
|
|
def __init__(
|
|
self,
|
|
title,
|
|
file_path,
|
|
tags,
|
|
publish_date: datetime | int,
|
|
account_file,
|
|
publish_strategy: str | None = None,
|
|
debug: bool = DEBUG_MODE,
|
|
headless: bool = LOCAL_CHROME_HEADLESS,
|
|
thumbnail_path=None,
|
|
desc: str | None = None,
|
|
):
|
|
super().__init__(
|
|
publish_date=publish_date,
|
|
account_file=account_file,
|
|
publish_strategy=publish_strategy,
|
|
debug=debug,
|
|
headless=headless,
|
|
)
|
|
self.title = title
|
|
self.file_path = file_path
|
|
self.tags = tags or []
|
|
self.thumbnail_path = thumbnail_path
|
|
self.desc = desc or ""
|
|
|
|
async def validate_upload_args(self):
|
|
await self.validate_base_args()
|
|
if not self.title or not str(self.title).strip():
|
|
raise ValueError("快手视频上传时，title 是必须的")
|
|
self.file_path = str(self.validate_video_file(self.file_path))
|
|
if self.thumbnail_path:
|
|
self.thumbnail_path = str(self.validate_image_file(self.thumbnail_path))
|
|
|
|
async def handle_upload_error(self, page: Page):
|
|
kuaishou_logger.warning(_msg("😵", "视频上传摔了一跤，小人马上重新上传"))
|
|
await page.locator('div.progress-div [class^="upload-btn-input"]').set_input_files(self.file_path)
|
|
|
|
async def set_thumbnail(self, page: Page):
|
|
if not self.thumbnail_path:
|
|
return
|
|
|
|
kuaishou_logger.info(_msg("🖼️", "小人准备设置封面"))
|
|
|
|
cover_label = page.locator("span").filter(has_text="封面设置")
|
|
await cover_label.wait_for(state="visible", timeout=30000)
|
|
await cover_label.locator("xpath=../following-sibling::div[1]").locator('div').nth(0).click()
|
|
|
|
modal = page.locator('div[role="document"].ant-modal')
|
|
await modal.wait_for(state="visible", timeout=30000)
|
|
|
|
upload_cover_tab = modal.get_by_text("上传封面", exact=True)
|
|
await upload_cover_tab.wait_for(state="visible", timeout=10000)
|
|
await upload_cover_tab.click()
|
|
|
|
file_input = modal.locator('input[type="file"]')
|
|
await file_input.wait_for(state="attached", timeout=30000)
|
|
await file_input.set_input_files(self.thumbnail_path)
|
|
await asyncio.sleep(1)
|
|
|
|
confirm_button = modal.get_by_role("button", name="确认", exact=True)
|
|
await confirm_button.wait_for(state="visible", timeout=10000)
|
|
await confirm_button.click()
|
|
|
|
await modal.wait_for(state="hidden", timeout=30000)
|
|
kuaishou_logger.success(_msg("🥳", "封面已经设置完成"))
|
|
|
|
async def upload(self, playwright: Playwright) -> None:
|
|
kuaishou_logger.info(_msg("🧍", "小人先检查 cookie、视频文件、封面和发布时间"))
|
|
await self.validate_upload_args()
|
|
kuaishou_logger.info(_msg("🥳", "上传前检查通过"))
|
|
|
|
if self.local_executable_path:
|
|
browser = await playwright.chromium.launch(
|
|
headless=self.headless,
|
|
executable_path=self.local_executable_path,
|
|
)
|
|
else:
|
|
browser = await playwright.chromium.launch(
|
|
headless=self.headless,
|
|
channel="chromium",
|
|
)
|
|
context = await browser.new_context(storage_state=self.account_file)
|
|
context = await set_init_script(context)
|
|
|
|
upload_success = False
|
|
try:
|
|
page = await context.new_page()
|
|
await page.goto(KUAISHOU_UPLOAD_URL)
|
|
kuaishou_logger.info(_msg("🏃", f"小人开始搬运视频: {self.title}.mp4"))
|
|
kuaishou_logger.info(_msg("🧭", "小人正在赶往快手上传主页"))
|
|
await page.wait_for_url(KUAISHOU_UPLOAD_URL_PATTERN)
|
|
|
|
upload_button = page.locator("button[class^='_upload-btn']")
|
|
await upload_button.wait_for(state="visible", timeout=10000)
|
|
|
|
async with page.expect_file_chooser() as fc_info:
|
|
await upload_button.click()
|
|
file_chooser = await fc_info.value
|
|
await file_chooser.set_files(self.file_path)
|
|
|
|
await asyncio.sleep(2)
|
|
|
|
know_button = page.locator('button[type="button"] span:text("我知道了")').first
|
|
try:
|
|
if await know_button.count() and await know_button.is_visible():
|
|
await know_button.click()
|
|
except Exception:
|
|
pass
|
|
|
|
await self.close_guide_overlay(page)
|
|
|
|
kuaishou_logger.info(_msg("✍️", "小人开始填描述和话题"))
|
|
await page.get_by_text("描述").locator("xpath=following-sibling::div").click()
|
|
await page.keyboard.press("Backspace")
|
|
await page.keyboard.press("Control+KeyA")
|
|
await page.keyboard.press("Delete")
|
|
await page.keyboard.type(self.desc or self.title)
|
|
await page.keyboard.press("Enter")
|
|
|
|
for index, tag in enumerate(self.tags[:3], start=1):
|
|
kuaishou_logger.info(_msg("🏷️", f"小人正在添加第 {index} 个话题: #{tag}"))
|
|
await page.keyboard.type(f"#{tag} ")
|
|
await asyncio.sleep(2)
|
|
|
|
max_retries = 60
|
|
retry_count = 0
|
|
while retry_count < max_retries:
|
|
try:
|
|
number = await page.locator("text=上传中").count()
|
|
if number == 0:
|
|
kuaishou_logger.success(_msg("🥳", "视频已经传完啦"))
|
|
break
|
|
|
|
if retry_count % 5 == 0:
|
|
kuaishou_logger.info(_msg("🏃", "小人正在努力上传视频"))
|
|
|
|
if await page.locator("text=上传失败").count():
|
|
await self.handle_upload_error(page)
|
|
|
|
await asyncio.sleep(2)
|
|
except Exception as exc:
|
|
kuaishou_logger.warning(_msg("😵", f"检查上传状态时出错，小人继续重试: {exc}"))
|
|
await asyncio.sleep(2)
|
|
retry_count += 1
|
|
|
|
if retry_count == max_retries:
|
|
kuaishou_logger.warning(_msg("😵", "超过最大重试次数，视频上传可能未完成"))
|
|
|
|
await self.set_thumbnail(page)
|
|
|
|
if self.publish_strategy == KUAISHOU_PUBLISH_STRATEGY_SCHEDULED and self.publish_date != 0:
|
|
await self.set_schedule_time(page, self.publish_date)
|
|
|
|
while True:
|
|
try:
|
|
publish_button = page.get_by_text("发布", exact=True)
|
|
if await publish_button.count() > 0:
|
|
await publish_button.click()
|
|
|
|
await asyncio.sleep(1)
|
|
confirm_button = page.get_by_text("确认发布")
|
|
if await confirm_button.count() > 0:
|
|
await confirm_button.click()
|
|
|
|
await page.wait_for_url(KUAISHOU_MANAGE_URL_PATTERN, timeout=5000)
|
|
kuaishou_logger.success(_msg("🥳", "视频发布成功，小人开心收工"))
|
|
break
|
|
except Exception as exc:
|
|
kuaishou_logger.info(_msg("🏃", f"小人正在冲刺发布视频: {exc}"))
|
|
if self.debug:
|
|
await page.screenshot(full_page=True)
|
|
await asyncio.sleep(1)
|
|
|
|
upload_success = True
|
|
finally:
|
|
if upload_success:
|
|
await context.storage_state(path=self.account_file)
|
|
kuaishou_logger.success(_msg("🥳", "cookie 更新完毕"))
|
|
await asyncio.sleep(2)
|
|
await context.close()
|
|
await browser.close()
|
|
|
|
async def main(self):
|
|
async with async_playwright() as playwright:
|
|
await self.upload(playwright)
|
|
|
|
|
|
class KSNote(KSBaseUploader):
|
|
def __init__(
|
|
self,
|
|
image_paths,
|
|
note,
|
|
tags,
|
|
publish_date: datetime | int,
|
|
account_file,
|
|
title: str | None = None,
|
|
publish_strategy: str | None = None,
|
|
debug: bool = DEBUG_MODE,
|
|
headless: bool = LOCAL_CHROME_HEADLESS,
|
|
):
|
|
super().__init__(
|
|
publish_date=publish_date,
|
|
account_file=account_file,
|
|
publish_strategy=publish_strategy,
|
|
debug=debug,
|
|
headless=headless,
|
|
)
|
|
self.image_paths = image_paths
|
|
self.note = note or ""
|
|
self.title = title or (self.note[:20] if self.note else "")
|
|
self.tags = tags or []
|
|
|
|
async def validate_upload_args(self):
|
|
await self.validate_base_args()
|
|
if not self.title or not str(self.title).strip():
|
|
raise ValueError("快手图文上传时，title 是必须的")
|
|
if not self.image_paths:
|
|
raise ValueError("快手图文上传时，图片是必须的")
|
|
|
|
if isinstance(self.image_paths, (str, Path)):
|
|
self.image_paths = [self.image_paths]
|
|
|
|
normalized_image_paths = []
|
|
for image_path in self.image_paths:
|
|
normalized_image_paths.append(str(self.validate_image_file(image_path)))
|
|
self.image_paths = normalized_image_paths
|
|
|
|
async def upload_note_content(self, page: Page) -> None:
|
|
kuaishou_logger.info(_msg("🏃", f"小人开始搬运图文，共 {len(self.image_paths)} 张图片"))
|
|
kuaishou_logger.info(_msg("🔀", "小人正在切换到图文发布"))
|
|
await page.locator('div[role="tablist"] div[role="tab"]:has-text("图文")').click()
|
|
await page.wait_for_timeout(1000)
|
|
|
|
kuaishou_logger.info(_msg("📤", "小人正在上传图片"))
|
|
upload_button = page.locator("button[class^='_upload-btn']").filter(has_text="上传图片")
|
|
await upload_button.wait_for(state="visible", timeout=10000)
|
|
|
|
async with page.expect_file_chooser() as fc_info:
|
|
await upload_button.click()
|
|
file_chooser = await fc_info.value
|
|
await file_chooser.set_files(self.image_paths)
|
|
|
|
know_button = page.locator('button[type="button"] span:text("我知道了")').first
|
|
try:
|
|
if await know_button.count() and await know_button.is_visible():
|
|
await know_button.click()
|
|
except Exception:
|
|
pass
|
|
|
|
await self.close_guide_overlay(page)
|
|
|
|
kuaishou_logger.info(_msg("✍️", "小人开始填写图文内容和话题"))
|
|
await page.get_by_text("描述").locator("xpath=following-sibling::div").click()
|
|
await page.keyboard.press("Backspace")
|
|
await page.keyboard.press("Control+KeyA")
|
|
await page.keyboard.press("Delete")
|
|
await page.keyboard.type(self.note)
|
|
await page.keyboard.press("Enter")
|
|
|
|
for index, tag in enumerate(self.tags[:3], start=1):
|
|
kuaishou_logger.info(_msg("🏷️", f"小人正在添加第 {index} 个话题: #{tag}"))
|
|
await page.keyboard.type(f"#{tag} ")
|
|
await asyncio.sleep(2)
|
|
|
|
max_retries = 60
|
|
retry_count = 0
|
|
while retry_count < max_retries:
|
|
try:
|
|
number = await page.locator("text=上传中").count()
|
|
if number == 0:
|
|
kuaishou_logger.success(_msg("🥳", "图文素材已经传完啦"))
|
|
break
|
|
|
|
if retry_count % 5 == 0:
|
|
kuaishou_logger.info(_msg("🏃", "小人正在努力上传图文素材"))
|
|
|
|
if await page.locator("text=上传失败").count():
|
|
kuaishou_logger.warning(_msg("😵", "图文素材上传摔了一跤，小人马上重新上传"))
|
|
await page.locator('div.progress-div [class^="upload-btn-input"]').set_input_files(self.image_paths)
|
|
|
|
await asyncio.sleep(2)
|
|
except Exception as exc:
|
|
kuaishou_logger.warning(_msg("😵", f"检查图文上传状态时出错，小人继续重试: {exc}"))
|
|
await asyncio.sleep(2)
|
|
retry_count += 1
|
|
|
|
if retry_count == max_retries:
|
|
kuaishou_logger.warning(_msg("😵", "超过最大重试次数，图文上传可能未完成"))
|
|
|
|
if self.publish_strategy == KUAISHOU_PUBLISH_STRATEGY_SCHEDULED and self.publish_date != 0:
|
|
await self.set_schedule_time(page, self.publish_date)
|
|
|
|
while True:
|
|
try:
|
|
publish_button = page.get_by_text("发布", exact=True)
|
|
if await publish_button.count() > 0:
|
|
await publish_button.click()
|
|
|
|
await asyncio.sleep(1)
|
|
confirm_button = page.get_by_text("确认发布")
|
|
if await confirm_button.count() > 0:
|
|
await confirm_button.click()
|
|
|
|
await page.wait_for_url(KUAISHOU_MANAGE_URL_PATTERN, timeout=5000)
|
|
kuaishou_logger.success(_msg("🥳", "图文发布成功，小人开心收工"))
|
|
break
|
|
except Exception as exc:
|
|
kuaishou_logger.info(_msg("🏃", f"小人正在冲刺发布图文: {exc}"))
|
|
if self.debug:
|
|
await page.screenshot(full_page=True)
|
|
await asyncio.sleep(1)
|
|
|
|
async def upload(self, playwright: Playwright) -> None:
|
|
kuaishou_logger.info(_msg("🧍", "小人先检查 cookie、图片和发布时间"))
|
|
await self.validate_upload_args()
|
|
kuaishou_logger.info(_msg("🥳", "图文上传前检查通过"))
|
|
|
|
if self.local_executable_path:
|
|
browser = await playwright.chromium.launch(
|
|
headless=self.headless,
|
|
executable_path=self.local_executable_path,
|
|
)
|
|
else:
|
|
browser = await playwright.chromium.launch(
|
|
headless=self.headless,
|
|
channel="chromium",
|
|
)
|
|
context = await browser.new_context(storage_state=self.account_file)
|
|
context = await set_init_script(context)
|
|
|
|
upload_success = False
|
|
try:
|
|
page = await context.new_page()
|
|
await page.goto(KUAISHOU_UPLOAD_URL)
|
|
kuaishou_logger.info(_msg("🧭", "小人正在赶往快手图文发布页"))
|
|
await page.wait_for_url(KUAISHOU_UPLOAD_URL_PATTERN)
|
|
|
|
await self.upload_note_content(page)
|
|
upload_success = True
|
|
finally:
|
|
if upload_success:
|
|
await context.storage_state(path=self.account_file)
|
|
kuaishou_logger.success(_msg("🥳", "cookie 更新完毕"))
|
|
await asyncio.sleep(2)
|
|
await context.close()
|
|
await browser.close()
|
|
|
|
async def main(self):
|
|
async with async_playwright() as playwright:
|
|
await self.upload(playwright)
|