awaioi/EveryPublish
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(kuaishou): robust desc editor focus, cookie auth and Joyride cleanup

Browse Source 
- _focus_desc_editor: locate the description editor with multiple fallback
  strategies instead of a single brittle selector; dump page screenshot/HTML
  on failure for easier selector updates
- cookie_auth: positive verification via the upload button presence, with
  login-page / intro-page / legacy checks as fallbacks
- close_guide_overlay: also remove the Joyride portal and spotlight layers
  that intercept clicks
- add collection_name support to KSVideo
This commit is contained in:
suihf
2026-08-13 09:50:21 +08:00
parent 008e4ff66a
commit 233cb990a6
2 changed files with 178 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
uploader/ks_uploader/main.py
+174 -18
View File
@@ -36,6 +36,62 @@ def _msg(emoji: str, text: str) -> str:
return f"{emoji} {text}"
async def _dump_page_debug(page, tag: str) -> str:
"""出错时保存整页截图 + 当前 HTML，返回保存目录，便于对照新 DOM 修选择器。"""
import time
base = Path("ks_debug")
base.mkdir(parents=True, exist_ok=True)
ts = time.strftime("%Y%m%d_%H%M%S")
try:
await page.screenshot(path=str(base / f"{tag}_{ts}.png"), full_page=True)
except Exception:
pass
try:
(base / f"{tag}_{ts}.html").write_text(await page.content(), encoding="utf-8")
except Exception:
pass
return str(base.resolve())
async def _focus_desc_editor(page) -> None:
"""定位并聚焦快手发布页的『描述』编辑区。
快手创作者中心 DOM 时有改版，旧的
get_by_text("描述").locator("xpath=following-sibling::div")
一旦结构变化就会干等 30s 超时。这里按多种策略依次尝试（都锚定在「描述」
标签附近，避免误点到标题框），每种短超时快速失败；全部失败则保存截图/HTML
供排查后抛出明确错误，而不是无脑超时。
"""
label = page.get_by_text("描述") # 默认子串匹配，"作品描述" 等也能命中
strategies = [
# 旧结构：『描述』相邻 div
lambda: label.locator("xpath=following-sibling::div"),
# 新版描述区通常是紧随其后的富文本可编辑区
lambda: label.locator("xpath=following::div[@contenteditable='true'][1]"),
# 同容器内的可编辑区
lambda: label.locator("xpath=ancestor::*[1]//div[@contenteditable='true'][1]"),
# 兜底：其后第一个任意可编辑元素
lambda: label.locator("xpath=following::*[@contenteditable='true'][1]"),
]
last_err = None
for i, make in enumerate(strategies):
try:
loc = make().first
await loc.wait_for(state="visible", timeout=8000)
await loc.click(force=True)
if i > 0:
kuaishou_logger.warning(_msg(
"⚠️", f"描述区改用回退策略#{i}定位成功（快手可能已改版，建议核对选择器）"))
return
except Exception as e: # noqa: BLE001
last_err = e
dbg = await _dump_page_debug(page, "desc_not_found")
raise RuntimeError(
f"未能定位快手『描述』编辑区（疑似发布页改版）。已保存截图/HTML 到 {dbg}，"
f"请据此更新选择器。最后错误: {last_err}")
def _print_ks_qrcode(qrcode_content: str, qrcode_path: Path) -> None:
try:
print_terminal_qrcode(qrcode_content, qrcode_path, "快手APP", compact=False, border=2)
@@ -161,12 +217,33 @@ async def cookie_auth(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 已失效，得重新登录一下"))
await page.wait_for_timeout(3000)
# 检查是否被重定向到登录页
if "passport.kuaishou.com" in page.url:
kuaishou_logger.info(_msg("🥹", "cookie 已失效（跳到登录页）"))
return False
kuaishou_logger.success(_msg("🥳", "cookie 有效"))
return True
# 检查是否停留在介绍页（未登录状态显示"立即登录"按钮）
login_btn = page.get_by_text("立即登录")
if await login_btn.count() > 0:
kuaishou_logger.info(_msg("🥹", "cookie 已失效（介绍页）"))
return False
# 正向证明：上传按钮存在 = 真正已登录
try:
upload_btn = page.locator("button[class^='_upload-btn']")
await upload_btn.wait_for(state="visible", timeout=10000)
kuaishou_logger.success(_msg("🥳", "cookie 有效"))
return True
except Exception:
# 兜底：旧版检测（"机构服务"元素出现在未登录介绍页）
if await _is_ks_cookie_invalid(page):
kuaishou_logger.info(_msg("🥹", "cookie 已失效（机构服务页）"))
return False
# 都没命中：保守判定为失效，避免假阳性
kuaishou_logger.warning(_msg("😵", "无法确认 cookie 有效性，按失效处理"))
return False
except Exception as exc:
kuaishou_logger.warning(_msg("😵", f"cookie 校验时出错，按失效处理: {exc}"))
return False
@@ -359,25 +436,59 @@ class KSBaseUploader(BaseVideoUploader):
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"]')
"""关闭快手创作者平台的 Joyride 引导遮罩。
# 判断是否显示
Joyride 有两个关键元素：
1. tooltip (alertdialog) — 引导提示框，有关闭按钮
2. spotlight (react-joyride__spotlight) — 聚光灯遮罩层，拦截点击事件
两者可能独立存在。必须都关掉才能正常操作页面。
"""
closed = False
# 方式1：点击 tooltip 的关闭/跳过按钮
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 引导遮罩，正在关闭...")
# 尝试多种关闭按钮 selector
close_selectors = [
'[aria-label="Skip"], [data-action="skip"], button[title="Skip"]',
'button:text("跳过")',
'button:text("我知道了")',
'button:text("关闭")',
'button:text("下一步")', # 有时需要多步跳过
]
for sel in close_selectors:
btn = page.locator('div[role="alertdialog"]').locator(sel)
if await btn.count() > 0:
await btn.first.click(force=True)
await asyncio.sleep(0.5)
break
closed = True
# 点击关闭按钮（X），使用多个可靠特征
close_button = page.locator('div[role="alertdialog"]').locator(
'[aria-label="Skip"], [data-action="skip"], button[title="Skip"]'
)
# 方式2：直接移除 Joyride portal（兜底，确保 spotlight 不再拦截）
joyride_portal = page.locator('div#react-joyride-portal')
if await joyride_portal.count() > 0:
try:
await page.evaluate("document.getElementById('react-joyride-portal')?.remove()")
print("✅ 已移除 Joyride portal 遮罩")
closed = True
except Exception:
pass
await close_button.click(force=True)
# 方式3：移除 spotlight 元素
spotlight = page.locator('div.react-joyride__spotlight')
if await spotlight.count() > 0:
try:
await page.evaluate("document.querySelectorAll('.react-joyride__spotlight').forEach(e => e.remove())")
print("✅ 已移除 Joyride spotlight")
closed = True
except Exception:
pass
# 等待遮罩消失
await joyride_tooltip.wait_for(state="hidden", timeout=5000)
print("✅ 已关闭 Joyride 遮罩")
else:
if not closed:
print("未检测到 Joyride 遮罩，继续执行")
else:
await asyncio.sleep(0.5)
class KSVideo(KSBaseUploader):
@@ -393,6 +504,7 @@ class KSVideo(KSBaseUploader):
headless: bool = LOCAL_CHROME_HEADLESS,
thumbnail_path=None,
desc: str | None = None,
collection_name: str | None = None,
):
super().__init__(
publish_date=publish_date,
@@ -406,6 +518,46 @@ class KSVideo(KSBaseUploader):
self.tags = tags or []
self.thumbnail_path = thumbnail_path
self.desc = desc or ""
self.collection_name = collection_name
async def apply_collection(self, page: Page) -> None:
"""在发布表单页选择"加入合集"下拉框（Ant Design Select，label 属性=合集名）。
锚点用 label 文字"加入合集"精确定位紧邻的 ant-select 容器，避免误选页面上
其它下拉框（服务类型/关联热点/作者声明/添加地点，同页面还有好几个 ant-select）。
找不到匹配名字的合集选项时按 Escape 收起下拉，保持未选状态直接发布（界面允许留空，
不阻断主发布流程）。
"""
if not self.collection_name:
return
try:
trigger = page.locator(
'label:text-is("加入合集")'
).locator("xpath=following-sibling::div[contains(@class,'ant-select')]").first
if await trigger.count() == 0:
kuaishou_logger.warning(_msg("😵", "未找到\"加入合集\"下拉框，跳过归集"))
return
await trigger.locator(".ant-select-selector").click(timeout=8000)
await page.wait_for_timeout(800)
option = page.locator(f'div.ant-select-item-option[label="{self.collection_name}"]')
if await option.count() == 0:
kuaishou_logger.warning(
_msg("😵", f"合集下拉框未找到「{self.collection_name}」，跳过归集，保持未选状态")
)
await page.keyboard.press("Escape")
await page.wait_for_timeout(300)
return
await option.first.click(timeout=8000)
await page.wait_for_timeout(500)
kuaishou_logger.success(_msg("🥳", f"已选择合集：{self.collection_name}"))
except Exception as exc:
kuaishou_logger.warning(_msg("😵", f"选择合集失败，跳过归集继续发布: {exc}"))
try:
await page.keyboard.press("Escape")
except Exception:
pass
async def validate_upload_args(self):
await self.validate_base_args()
@@ -494,7 +646,9 @@ class KSVideo(KSBaseUploader):
await self.close_guide_overlay(page)
kuaishou_logger.info(_msg("✍️", "小人开始填描述和话题"))
await page.get_by_text("描述").locator("xpath=following-sibling::div").click()
# 再次检查并关闭 Joyride（可能在文件上传后才弹出）
await self.close_guide_overlay(page)
await _focus_desc_editor(page)
await page.keyboard.press("Backspace")
await page.keyboard.press("Control+KeyA")
await page.keyboard.press("Delete")
@@ -532,6 +686,8 @@ class KSVideo(KSBaseUploader):
await self.set_thumbnail(page)
await self.apply_collection(page)
if self.publish_strategy == KUAISHOU_PUBLISH_STRATEGY_SCHEDULED and self.publish_date != 0:
await self.set_schedule_time(page, self.publish_date)
@@ -634,7 +790,7 @@ class KSNote(KSBaseUploader):
await self.close_guide_overlay(page)
kuaishou_logger.info(_msg("✍️", "小人开始填写图文内容和话题"))
await page.get_by_text("描述").locator("xpath=following-sibling::div").click()
await _focus_desc_editor(page)
await page.keyboard.press("Backspace")
await page.keyboard.press("Control+KeyA")
await page.keyboard.press("Delete")