# -*- coding: utf-8 -*-
|
from playwright.async_api import Page
|
import asyncio
|
from auto_media_publisher.uploaders.uploader_base import UploaderBase
|
|
CONTENT_UPLOAD_URL="https://creator.xiaohongshu.com/publish/publish?from=menu"
|
VIDEO_PUBLISH_SECCESSFULL_URL = "https://creator.xiaohongshu.com/publish/success?source=&bind_status=not_bind&__debugger__=&proxy="
|
|
|
class UploaderXiaohongshu(UploaderBase):
|
|
async def set_schedule_time(self, page):
|
try:
|
|
scheduled_date_hour = self.scheduled_time.strftime("%Y-%m-%d %H:%M")
|
await page.locator("label").filter(has_text="定时发布").locator("span").nth(1).click()
|
await asyncio.sleep(1)
|
await page.get_by_placeholder("选择日期和时间").click()
|
await asyncio.sleep(1)
|
await page.keyboard.press("Control+KeyA")
|
await page.keyboard.type(str(scheduled_date_hour))
|
await page.keyboard.press("Enter")
|
await asyncio.sleep(1)
|
except Exception as e:
|
self.logger.error(f" [exception] 正在设置发布时间错误：{e}")
|
await asyncio.sleep(2)
|
|
async def set_tile_and_description(self,page):
|
try:
|
self.logger.info("正在填充标题和话题...")
|
await page.get_by_placeholder("填写标题会有更多赞哦～").click()
|
await page.get_by_placeholder("填写标题会有更多赞哦～").fill(self.title[:20])
|
await page.locator("#quillEditor div").click()
|
description_tags = self.description + ' ' + ' '.join(f"#{tag}" for tag in self.tags)
|
await page.locator("#quillEditor div").fill(description_tags)
|
await page.mouse.click(0, 0) # 页面左上角一般是空白
|
except Exception as e:
|
self.logger.error(f" [exception] 正在设置 标题和描述错误：{e}")
|
await asyncio.sleep(2)
|
|
async def set_cover_image(self,page):
|
try:# STEP 3 设置封面
|
self.logger.info("正在设置 封面...")
|
await page.get_by_text("设置封面", exact=True).wait_for()
|
await page.get_by_text("设置封面", exact=True).click()
|
img_upload_button = page.get_by_text("上传图片", exact=True)
|
await img_upload_button.wait_for(state='visible') # 确保按钮可见
|
|
async with page.expect_file_chooser() as fc_info:
|
await img_upload_button.click()
|
file_chooser = await fc_info.value
|
await file_chooser.set_files(self.path_cover)
|
await asyncio.sleep(2)
|
await page.get_by_role("button", name="确定").click(timeout=10000)
|
except Exception as e:
|
self.logger.error(f" [exception] 正在上传视频中... 错误：{e}")
|
await asyncio.sleep(2)
|
|
async def upload_media_and_wait(self,page):
|
try:
|
upload_button = page.locator("input[class^='upload-input']")
|
await upload_button.wait_for(state='visible') # 确保按钮可见
|
|
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.path_media)
|
await asyncio.sleep(2)
|
|
# wait for video upload finish
|
try:
|
locator = page.locator('div:has-text("上传成功")')
|
count = await locator.count()
|
if count > 0:
|
# 判断是否至少有一个匹配的元素是可见的
|
for i in range(count):
|
if await locator.nth(i).is_visible():
|
self.logger.info(" [-] 视频上传完毕")
|
break
|
else:
|
self.logger.info(" [-] 正在上传视频中（元素存在但不可见）...")
|
await asyncio.sleep(3)
|
continue
|
break # 跳出 while True
|
else:
|
self.logger.info(" [-] 正在上传视频中...")
|
await asyncio.sleep(3)
|
except Exception as e:
|
self.logger.error(f" [exception] 正在上传视频中... 错误：{e}")
|
await asyncio.sleep(2)
|
except Exception as e:
|
self.logger.error(f" [exception] 正在上传视频中... 错误：{e}")
|
await asyncio.sleep(2)
|
|
async def _upload_core(self) -> bool:
|
try:
|
page = await self.context.new_page()
|
await page.goto(CONTENT_UPLOAD_URL, timeout=120000)
|
self.logger.info('正在打开主页...')
|
await page.wait_for_url(CONTENT_UPLOAD_URL, timeout=120000)
|
|
await self.upload_media_and_wait(page)
|
|
await self.set_tile_and_description(page)
|
|
if self.path_cover:
|
await self.set_cover_image(page)
|
|
# 定时任务
|
if self.scheduled_time != 0:
|
self.logger.info("正在设置 定时发布 时间...")
|
await self.set_schedule_time(page)
|
|
# 判断视频是否发布成功
|
await page.get_by_role("button", name="定时发布").click()
|
await asyncio.sleep(1)
|
max_retry = 10
|
for _ in range(max_retry):
|
try:
|
current_url = page.url
|
if "success" in current_url: # 或者更精确匹配你需要的路径片段
|
self.logger.info(f"视频发布成功，跳转URL: {current_url}")
|
break
|
else:
|
self.logger.info(f"视频正在发布中... 当前URL: {current_url}")
|
await asyncio.sleep(1)
|
except Exception as e:
|
self.logger.warning(f"等待跳转异常: {e}")
|
await asyncio.sleep(1)
|
else:
|
self.logger.error("超过最大重试次数，发布可能失败。")
|
|
# 始终保存 cookie 并关闭浏览器
|
await self.context.storage_state(path=self.cookie_path)
|
self.logger.info('cookie更新完毕！')
|
await asyncio.sleep(3)
|
return True
|
except Exception as e:
|
self.logger.error(f"{self.user} upload video {self.path_media} fails, error is :{e}")
|
return False
|
