# -*- coding: utf-8 -*- from playwright.async_api import Page import asyncio from auto_media_publisher.uploaders.uploader_base import UploaderBase CONTENT_UPLOAD_URL="https://channels.weixin.qq.com/platform/post/create" VIDEO_PUBLISH_SECCESSFULL_URL = "https://channels.weixin.qq.com/platform/post/list" class UploaderShipinhao(UploaderBase): async def set_schedule_time_tencent(self, page): label_element = page.locator("label").filter(has_text="定时").nth(1) await label_element.click() await page.click('input[placeholder="请选择发表时间"]') str_month = str(self.scheduled_time.month) if self.scheduled_time.month > 9 else "0" + str(self.scheduled_time.month) current_month = str_month + "月" # 获取当前的月份 page_month = await page.inner_text('span.weui-desktop-picker__panel__label:has-text("月")') # 检查当前月份是否与目标月份相同 if page_month != current_month: await page.click('button.weui-desktop-btn__icon__right') # 获取页面元素 elements = await page.query_selector_all('table.weui-desktop-picker__table a') # 遍历元素并点击匹配的元素 for element in elements: if 'weui-desktop-picker__disabled' in await element.evaluate('el => el.className'): continue text = await element.inner_text() if text.strip() == str(self.scheduled_time.day): await element.click() break # 输入小时部分（假设选择11小时） await page.click('input[placeholder="请选择时间"]') await page.keyboard.press("Control+KeyA") await page.keyboard.type(str(self.scheduled_time.hour)) # 选择标题栏（令定时时间生效） await page.locator("div.input-editor").click() async def handle_upload_error(self, page): self.logger.info("视频出错了，重新上传中") await page.locator('div.media-status-content div.tag-inner:has-text("删除")').click() await page.get_by_role('button', name="删除", exact=True).click() file_input = page.locator('input[type="file"]') await file_input.set_input_files(self.path_media) async def add_short_title(self, page): short_title_element = page.get_by_text("短标题", exact=True).locator("..").locator( "xpath=following-sibling::div").locator( 'span input[type="text"]') if await short_title_element.count(): short_title = self.format_str_for_short_title(self.title) await short_title_element.fill(short_title) async def click_publish(self, page, max_retry_times=20): retry_times = 0 while retry_times < max_retry_times: try: publish_button = page.locator('div.form-btns button:has-text("发表")') if await publish_button.count(): await publish_button.click() await page.wait_for_url(VIDEO_PUBLISH_SECCESSFULL_URL, timeout=1500) self.logger.info(" [-] 视频发布成功") return except Exception as e: current_url = page.url if VIDEO_PUBLISH_SECCESSFULL_URL in current_url: self.logger.info(" [-] 视频发布成功（通过URL判断）") return else: retry_times += 1 self.logger.warning(f" [-] 第 {retry_times} 次重试，异常: {e}") self.logger.info(" [-] 视频正在发布中...") await asyncio.sleep(0.5) self.logger.error(" [-] 超过最大重试次数，发布失败") async def detect_upload_status(self, page): while True: # 匹配删除按钮，代表视频上传完毕，如果不存在，代表视频正在上传，则等待 try: # 匹配删除按钮，代表视频上传完毕 if "weui-desktop-btn_disabled" not in await page.get_by_role("button", name="发表").get_attribute( 'class'): self.logger.info(" [-]视频上传完毕") break else: self.logger.info(" [-] 正在上传视频中...") await asyncio.sleep(2) # 出错了视频出错 if await page.locator('div.status-msg.error').count() and await page.locator( 'div.media-status-content div.tag-inner:has-text("删除")').count(): self.logger.error(" [-] 发现上传出错了...准备重试") await self.handle_upload_error(page) except: self.logger.info(" [-] 正在上传视频中...") await asyncio.sleep(2) async def add_title_tags(self, page): await page.locator("div.input-editor").click() await page.keyboard.type(self.title) await page.keyboard.press("Enter") for index, tag in enumerate(self.tags, start=1): await page.keyboard.type("#" + tag) await page.keyboard.press("Space") self.logger.info(f"成功添加hashtag: {len(self.tags)}") async def add_collection(self, page): collection_elements = page.get_by_text("添加到合集").locator("xpath=following-sibling::div").locator( '.option-list-wrap > div') if await collection_elements.count() > 1: await page.get_by_text("添加到合集").locator("xpath=following-sibling::div").click() await collection_elements.first.click() async def add_original(self, page): if await page.get_by_label("视频为原创").count(): await page.get_by_label("视频为原创").check() # 检查 "我已阅读并同意 《视频号原创声明使用条款》" 元素是否存在 label_locator = await page.locator('label:has-text("我已阅读并同意 《视频号原创声明使用条款》")').is_visible() if label_locator: await page.get_by_label("我已阅读并同意 《视频号原创声明使用条款》").check() await page.get_by_role("button", name="声明原创").click() # 2023年11月20日 wechat更新: 可能新账号或者改版账号，出现新的选择页面 """ if await page.locator('div.label span:has-text("声明原创")').count() and self.category: if not await page.locator('div.declare-original-checkbox input.ant-checkbox-input').is_disabled(): await page.locator('div.declare-original-checkbox input.ant-checkbox-input').click() if not await page.locator( 'div.declare-original-dialog label.ant-checkbox-wrapper.ant-checkbox-wrapper-checked:visible').count(): await page.locator('div.declare-original-dialog input.ant-checkbox-input:visible').click() if await page.locator('div.original-type-form > div.form-label:has-text("原创类型"):visible').count(): await page.locator('div.form-content:visible').click() # 下拉菜单 await page.locator( f'div.form-content:visible ul.weui-desktop-dropdown__list li.weui-desktop-dropdown__list-ele:has-text("{self.category}")').first.click() await page.wait_for_timeout(1000) if await page.locator('button:has-text("声明原创"):visible').count(): await page.locator('button:has-text("声明原创"):visible').click() """ def format_str_for_short_title(self,origin_title: str) -> str: allowed_special_chars = "《》“”:+?%°" filtered_chars = [char if char.isalnum() or char in allowed_special_chars else ' ' if char == ',' else '' for char in origin_title] formatted_string = ''.join(filtered_chars) if len(formatted_string) > 16: formatted_string = formatted_string[:16] elif len(formatted_string) < 6: formatted_string += ' ' * (6 - len(formatted_string)) return formatted_string 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(f'[+]正在上传-------{self.title}.mp4') await page.wait_for_url(CONTENT_UPLOAD_URL, timeout=120000) file_input = page.locator('input[type="file"]') await file_input.set_input_files(self.path_media) # 填充标题和话题 await self.add_title_tags(page) # 添加商品 # await self.add_product(page) # 合集功能 await self.add_collection(page) # 原创选择 await self.add_original(page) # 检测上传状态 await self.detect_upload_status(page) if self.scheduled_time != 0: await self.set_schedule_time_tencent(page) # 添加短标题 await self.add_short_title(page) await self.click_publish(page) await self.context.storage_state(path=f"{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