# -*- coding: utf-8 -*- from playwright.async_api import Playwright, Page import asyncio import datetime from auto_media_publisher.uploaders.uploader_base import UploaderBase CONTENT_UPLOAD_URL="https://baijiahao.baidu.com/builder/rc/edit?type=videoV2" VIDEO_PUBLISH_SECCESSFULL_URL = "https://baijiahao.baidu.com/builder/rc/clue**" class UploaderBaijiahao(UploaderBase): async def virtual_scroll_select(self,page,text, wrapper_index = 0,panel_selector="div.rc-virtual-list-holder-inner", item_selector_template='div.cheetah-select-item-option-content:has-text("{text}")', step=200, # 每次滚动像素 max_attempts=15, direction="down" # "down" 或 "up" ): await page.locator('div.select-wrap').nth(wrapper_index).click() await asyncio.sleep(1) panel = page.locator(panel_selector).nth(wrapper_index) item_selector = item_selector_template.format(text=text) for i in range(max_attempts): # 尝试查找 items = page.locator(item_selector) if await items.count() > 0: # 找到了，滚下去保证可见，再点击 await items.first.scroll_into_view_if_needed() await items.first.click() return # 没找到就滚一轮 delta = step if direction == "down" else -step await panel.evaluate(f"(el) => el.scrollTop += {delta}") await asyncio.sleep(0.2) await asyncio.sleep(1) async def set_schedule_time(self, page): """ TODO 时间选择，日后在处理 百家号的时间选择不准确，目前是随机 """ scheduled_time_day = f"{self.scheduled_time.month}月{self.scheduled_time.day}日" if self.scheduled_time.day >9 else f"{self.scheduled_time.month}月0{self.scheduled_time.day}日" scheduled_time_hour = f"{self.scheduled_time.hour}点" scheduled_time_min = f"{self.scheduled_time.minute}分" await page.locator("div.cheetah-modal-confirm-body-wrapper").wait_for(state='visible',timeout=30000) current_hour = datetime.datetime.now().hour current_min = datetime.datetime.now().minute direction_hour = "down" if self.scheduled_time.hour > current_hour else "up" direction_minute = "down" if self.scheduled_time.minute > current_min else "up" print(direction_hour) print(dir) try: await page.locator('div.select-wrap').nth(0).click() await asyncio.sleep(1) target = page.locator(f'div[title="{scheduled_time_day}"]') #await target.wait_for(state='visible') await target.click() await asyncio.sleep(1) await self.virtual_scroll_select(page,scheduled_time_hour,wrapper_index=1,direction=direction_hour) await self.virtual_scroll_select(page,scheduled_time_min,wrapper_index=2,direction=direction_minute) #await page.locator('div.select-wrap').nth(2).click() #await asyncio.sleep(1) #target = page.locator(f'div.cheetah-select-item-option-content:has-text("{scheduled_time_min}")') #await target.wait_for(state='visible') #await target.scroll_into_view_if_needed() #await target.click() #await asyncio.sleep(1) except Exception as e: self.logger.error(f"[Baidu BaiJiaHao] 设置时间发生错误.{e}") #await page.locator("div.cheetah-modal-confirm-btns > button").nth(1).click() async def handle_upload_error(self, page): # 日后实现，目前没遇到 return print("视频出错了，重新上传中") async def uploading_video(self, page): while True: upload_failed = await page.locator('div .cover-overlay:has-text("上传失败")').count() if upload_failed: self.logger.error("发现上传出错了...") # await self.handle_upload_error(page) # 假设这是处理上传错误的函数 return False uploading = await page.locator('div .cover-overlay:has-text("上传中")').count() if uploading: self.logger.info("正在上传视频中...") await asyncio.sleep(2) # 等待2秒再次检查 continue # 检查上传是否成功 if not uploading and not upload_failed: self.logger.info("视频上传完毕") return True async def set_schedule_publish(self, page): while True: try: self.logger.info("开始点击发布定时...") buttons = page.get_by_role("button", name="定时发布") count = await buttons.count() for i in range(count): btn = buttons.nth(i) if await btn.is_visible(): await btn.click() break await asyncio.sleep(2) # 等待2秒再次检查 await self.set_schedule_time(page) except Exception as e: self.logger.error(f"定时发布失败: {e}") raise # 重新抛出异常，让重试装饰器捕获 async def publish_video(self, page: Page): if self.scheduled_time != 0: # 定时发布 await self.set_schedule_publish(page) else: # 立即发布 await self.direct_publish(page) async def direct_publish(self, page): try: publish_button = page.locator("button >> text=发布") if await publish_button.count(): await publish_button.click() except Exception as e: self.logger.error(f"直接发布视频失败: {e}") raise # 重新抛出异常，让重试装饰器捕获 async def add_title_tags(self, page): title_page = page.get_by_placeholder('添加标题获得更多推荐') if len(self.title) <= 8: self.title += " 你不知道的" await title_page.fill(self.title[:30]) async def _upload_core(self) -> bool: try: page = await self.context.new_page() await page.goto(CONTENT_UPLOAD_URL, timeout=120000) self.logger.info(f"正在上传-------{self.path_media}") await page.wait_for_url(CONTENT_UPLOAD_URL, timeout=120000) # 点击 "上传视频" 按钮 upload_div = page.locator(".PL3VsCg3MxIPY5UYiCS_") await upload_div.wait_for(state="visible", timeout=10000) async with page.expect_file_chooser() as fc_info: await upload_div.click() file_chooser = await fc_info.value await file_chooser.set_files(self.path_media) await asyncio.sleep(2) # 等待页面跳转到指定的 URL for i in range(10): try: await page.locator("div#formMain:visible").wait_for(timeout=30000) break except: self.logger.info(f"第 {i+1} 次尝试：正在等待进入视频发布页面...") await asyncio.sleep(0.1) else: self.logger.warning("超过最大尝试次数，仍未进入视频发布页面。") # 填充标题和话题 # 这里为了避免页面变化，故使用相对位置定位：作品标题父级右侧第一个元素的input子元素 await asyncio.sleep(1) self.logger.info("正在填充标题和话题...") await self.add_title_tags(page) upload_status = await self.uploading_video(page) if not upload_status: self.logger.error(f"发现上传出错了... 文件:{self.path_media}") raise # 判断视频封面图是否生成成功 while True: self.logger.info("正在确认封面完成, 准备去点击定时/发布...") if await page.locator("div.erVanTLKMOvnU7OgFaoD img").count(): self.logger.info("封面已完成，点击定时/发布...") break else: self.logger.info("等待封面生成...") await asyncio.sleep(3) await self.publish_video(page) await page.wait_for_timeout(2000) if await page.locator('div.passMod_dialog-page >> text=百度安全验证:visible').count(): self.logger.error("出现验证，退出") raise Exception("出现验证，退出") await page.wait_for_url(VIDEO_PUBLISH_SECCESSFULL_URL, timeout=20000) self.logger.info("视频发布成功") await self.context.storage_state(path=self.cookie_path) # 保存cookie self.logger.info('cookie更新完毕！') await asyncio.sleep(2) # 这里延迟是为了方便眼睛直观的观看 return True except Exception as e: self.logger.error(f"{self.user} upload video {self.path_media} fails, error is :{e}") return False