# -*- coding: utf-8 -*- from playwright.async_api import Page import asyncio from auto_media_publisher.uploaders.uploader_base import UploaderBase CONTENT_UPLOAD_URL="https://member.bilibili.com/platform/upload/video/frame" VIDEO_PUBLISH_SECCESSFULL_URL = "https://cp.kuaishou.com/article/manage/video?status=2&from=publish" class UploaderBilibili(UploaderBase): async def set_schedule_time(self, page): try: await page.locator(".switch-roll").first.click() await asyncio.sleep(1) scheduled_date = self.scheduled_time.strftime("%Y-%m-%d") scheduled_month = self.scheduled_time.strftime("%Y年%m月") exact_day = str(self.scheduled_time.day) await page.locator(".date-show-icon").first.click() await asyncio.sleep(1) if scheduled_month != await page.locator("p.date-picker-nav-title").text_content(): await page.locator(".next-btn-day.icon-sprite.icon-sprite-ic_into").click() await asyncio.sleep(1) await page.get_by_text(exact_day, exact=True).click() await asyncio.sleep(1) exact_hour = f"{self.scheduled_time.hour:02d}" exact_min = f"{self.scheduled_time.minute:02d}" await page.locator(".date-picker-timer > .date-show-icon").click() await asyncio.sleep(0.5) await page.locator('div.time-picker-panel-select-wrp').nth(0).locator(f'span:nth-child({self.scheduled_time.hour + 1})').click() minute_index = ((self.scheduled_time.minute // 5) * 5 // 5) + 1 # 序号从1开始 await page.locator('div.time-picker-panel-select-wrp').nth(1).locator(f'span:nth-child({minute_index})').click() await asyncio.sleep(1) except Exception as e: self.logger.error(f'[ERROR ] {self.user} , {self.platform} ,设置 发布时间 发生错误：{e}') async def set_cover_image(self,page): try: # 通过点击 div.content 触发文件选择框的打开 upload_div = page.locator("div.cover-main-upload div.content") 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_cover) # 确保上传完成 await page.locator('button:has-text("完成")').click(timeout=30000) # 点击“完成”按钮，确保上传完成 except Exception as e: self.logger.error(f'[ERROR ] {self.user} , {self.platform} ,设置 封面 发生错误：{e}') async def upload_media_and_wait(self,page): try: upload_div = page.get_by_text("拖拽到此处也可上传上传视频") 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) max_retries = 60 # 设置最大重试次数,最大等待时间为 2 分钟 retry_count = 0 while retry_count < max_retries: try: number = await page.locator('span.success', has_text="上传完成").count() if number > 0: self.logger.info("视频上传完毕") break else: if retry_count % 5 == 0: self.logger.info("正在上传视频中...") await asyncio.sleep(2) except Exception as e: self.logger.error(f"检查上传状态时发生错误: {e}") await asyncio.sleep(2) # 等待 2 秒后重试 retry_count += 1 if retry_count == max_retries: self.logger.warning("超过最大重试次数，视频上传可能未完成。") except Exception as e: self.logger.error(f'[ERROR ] {self.user} , {self.platform} ,设置 封面 发生错误：{e}') async def set_title_and_description(self,page): try: self.logger.info("正在填充标题和话题...") await page.get_by_placeholder("请输入稿件标题").click() await page.get_by_placeholder("请输入稿件标题").press("ControlOrMeta+a") await page.get_by_placeholder("请输入稿件标题").fill(self.title) await page.locator('div[data-placeholder="填写更全面的相关信息，让更多的人能找到你的视频吧"]').click() description_tags = self.description + ' ' + ' '.join(f"#{tag}" for tag in self.tags) await page.locator('div[data-placeholder="填写更全面的相关信息，让更多的人能找到你的视频吧"]').fill(description_tags) # 快手只能添加3个话题 for index, tag in enumerate(self.tags[:7], start=1): self.logger.info("正在添加第%s个话题" % index) await page.get_by_placeholder("按回车键Enter创建标签").nth(0).click() await page.keyboard.type(tag) await page.keyboard.press("Enter") await asyncio.sleep(1) except Exception as e: self.logger.error(f'[ERROR ] {self.user} , {self.platform} ,设置 标题和描述 发生错误：{e}') async def _upload_core(self) -> bool: try: # 创建一个新的页面 page = await self.context.new_page() # 访问指定的 URL await page.goto(CONTENT_UPLOAD_URL, timeout=120000) self.logger.info('正在上传-------{}.mp4'.format(self.title)) await page.wait_for_url(CONTENT_UPLOAD_URL, timeout=120000) await self.upload_media_and_wait(page) await self.set_title_and_description(page) await self.set_cover_image(page) # select CLass if self.category: try: await page.locator(".select-controller").first.click() await asyncio.sleep(1) await page.get_by_title(self.category).click() except Exception as e: self.logger.error(f'[ bilibili ] setting catalog error. {e}') # 定时发布 if self.scheduled_time != 0: await self.set_schedule_time(page) # 判断视频是否发布成功 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) await page.get_by_role("button", name="再投一个").wait_for(timeout=10000) self.logger.info("视频发布成功") break except Exception as e: self.logger.info(f"视频正在发布中... 错误: {e}") await page.screenshot(full_page=True) await asyncio.sleep(1) await self.context.storage_state(path=self.cookie_path) # 保存cookie 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