# -*- coding: utf-8 -*- from playwright.async_api import Page import asyncio from auto_media_publisher.uploaders.uploader_base import UploaderBase from datetime import datetime,date import re CONTENT_UPLOAD_URL="https://mp.weixin.qq.com/" PUBLISH_SECCESSFULL_URL = "https://mp.weixin.qq.com/cgi-bin/home?t=home/index" class UploaderGongzhonghao(UploaderBase): async def wai_for_upload(self,new_page,selector): await new_page.evaluate(f'''() => {{ const container = document.querySelector('{selector}'); if (container) {{ const input = container.querySelector('input[type="file"]'); // 从父元素开始查找 if (input) {{ input.style.display = 'block'; // 改变 display 为 block，使其可见 input.style.visibility = 'visible'; // 使其可见 input.style.opacity = '1'; // 确保它完全透明 }} }} }}''') async def set_schedule_time(self, new_page): try: save_publish= await new_page.wait_for_selector("#js_send button", timeout=60000) await save_publish.click(force=True) await asyncio.sleep(2) send_option_div = await new_page.wait_for_selector(".weui-desktop-dialog__ft", timeout=60000) if self.scheduled_time > datetime.now() : await new_page.locator(".mass-send__td-setting > .publish_container > .weui-desktop-form__controls > .mass-send__timer-wrp > .weui-desktop-switch > .weui-desktop-switch__box").click() delta_days = (self.scheduled_time.date() - date.today()).days if delta_days > 0: await new_page.locator("dt").filter(has_text="今天 今天").click() option_text = "明天" if delta_days==1 else f"月{self.scheduled_time.day}日" #day_selector = new_page.locator(f"li.weui-desktop-dropdown__list-ele_text").nth(delta_days) #await day_selector.click() await new_page.get_by_text(option_text).first.click() await new_page.get_by_role("textbox", name="请选择时间").click() await new_page.get_by_role("textbox", name="请选择时间").press("ControlOrMeta+a") await new_page.get_by_role("textbox", name="请选择时间").fill(self.scheduled_time.strftime("%H:%M")) await new_page.get_by_role("heading",name="发表").click() send_option_send = await send_option_div.wait_for_selector('text="发表"',state="attached", timeout=600000) await send_option_send.click(force=True) await asyncio.sleep(3) except Exception as e: self.logger.error(f'[ERROR ] {self.user} , {self.platform} ,设置 发布时间 发生错误：{e}') async def set_cover_image(self,new_page): try: # 等待并定位到 #js_cover_area 元素 cover_area = await new_page.wait_for_selector('#js_cover_area') # 悬停鼠标在 #js_cover_area 上 await cover_area.hover() await new_page.get_by_role("link", name="从图片库选择").click() image_upload_container = await new_page.wait_for_selector(".weui-desktop-dialog__bd", timeout=60000) await self.wai_for_upload(new_page,".weui-desktop-dialog__bd") image_input = await image_upload_container.wait_for_selector('input[type="file"]', timeout=60000) await image_input.set_input_files(self.path_cover) library_next_div = await new_page.wait_for_selector(".weui-desktop-dialog__ft", timeout=60000) await library_next_div.evaluate('''(div) => { return new Promise(resolve => { const btn = Array.from(div.querySelectorAll('button')).find(el => el.innerText.trim() === '下一步'); if (!btn) return; // 创建 MutationObserver 监听 class 变化 const observer = new MutationObserver(() => { if (!btn.classList.contains('weui-desktop-btn_disabled')) { observer.disconnect(); // 停止监听 resolve(true); } }); observer.observe(btn, { attributes: true, attributeFilter: ['class'] }); // 如果初始状态就满足条件，直接 resolve if (!btn.classList.contains('weui-desktop-btn_disabled')) { observer.disconnect(); resolve(true); } }); }''') # 找到按钮并点击 next_btn = await library_next_div.wait_for_selector('text="下一步"', state="attached", timeout=60000) await next_btn.click(force=True) # whole div has been changed library_next_div = await new_page.wait_for_selector(".weui-desktop-dialog__ft", timeout=60000) confirm_btn = await library_next_div.wait_for_selector('text="确认"',state="attached", timeout=600000) await confirm_btn.scroll_into_view_if_needed() # 确保按钮滚动到视图 await confirm_btn.click(force=True) #wait for setting cover image: preview_div_selector = 'div.js_cover_preview_new.select-cover__preview.first_appmsg_cover' while True: # 使用 evaluate 来检查 background-image 样式属性 has_background_image = await new_page.evaluate('''(selector) => { const div = document.querySelector(selector); if (div) { const bgImage = window.getComputedStyle(div).getPropertyValue('background-image'); return bgImage.includes('https'); } return false; }''', preview_div_selector) if has_background_image: print("背景图已加载，继续执行操作") break # 背景图片已加载，跳出循环 else: print("背景图未加载，继续等待") await asyncio.sleep(1) # 每秒检查一次 await asyncio.sleep(1) except Exception as e: self.logger.error(f'[ERROR ] {self.user} , {self.platform} ,设置 封面 发生错误：{e}') async def upload_media_and_wait(self,new_page): try: await new_page.get_by_role("listitem", name="文档导入").click() import_container = await new_page.wait_for_selector("#js_import_file_container", timeout=60000) await self.wai_for_upload(new_page,"#js_import_file_container") file_input = await import_container.wait_for_selector('input[type="file"]', timeout=60000) await file_input.set_input_files(self.path_media) # 替换为你的文件路径 await new_page.wait_for_selector("#js_import_file_container", state="hidden", timeout=60000) self.logger.info(f"{self.user},{self.path_media} 文件导入完成！") await asyncio.sleep(1) except Exception as e: self.logger.error(f'[ERROR ] {self.user} , {self.platform} ,设置 封面 发生错误：{e}') async def set_title_and_description(self,new_page): try: # 设置标题和作者，要在导入之后，否则会被修改掉 if self.title and len(self.title) >1: title_textarea = await new_page.wait_for_selector("#title", timeout=60000) await title_textarea.scroll_into_view_if_needed() await title_textarea.click() await new_page.keyboard.press('Control+A') # 对于 Mac，可以使用 'Meta+A' await new_page.keyboard.press('Backspace') await title_textarea.type(self.title, delay=50) # delay 可以控制输入的速度 author_input = await new_page.wait_for_selector("#author", timeout=60000) await author_input.fill(self.author) description_input = await new_page.wait_for_selector("#js_description", timeout=60000) await description_input.fill(self.description) await asyncio.sleep(1) except Exception as e: self.logger.error(f'[ERROR ] {self.user} , {self.platform} ,设置 标题和描述 发生错误：{e}') async def set_tags(self,new_page): try: await new_page.locator("#js_article_tags_area").get_by_text("未添加").click() await new_page.locator('input[placeholder="请选择合集"]').click() await asyncio.sleep(2) # 只选第一个 await new_page.locator('li.select-opt-li').nth(0).click() ''' for tag in self.tags: try: await new_page.get_by_placeholder("输入后按回车分割").fill(tag) await new_page.get_by_placeholder("输入后按回车分割").press("Enter") except Exception as e: print(f"添加标签 '{tag}' 失败: {str(e)}") ''' await new_page.get_by_role("button", name="确认").click() await asyncio.sleep(3) except Exception as e: self.logger.error(f'[ERROR ] {self.user} , {self.platform} ,设置 tag 发生错误：{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(f'[ gongzhognhao ] {self.user} , {self.scheduled_time} 正在打开-文章上传页面') # 点击“文章” await page.locator(".new-creation__menu-content").first.click() # 等待新窗口出现并切换 new_page = await page.context.wait_for_event("page") await new_page.wait_for_load_state() await self.upload_media_and_wait(new_page) await self.set_title_and_description(new_page) await self.set_cover_image(new_page) if self.tags: await self.set_tags(new_page) # 定时发布 if self.scheduled_time: await self.set_schedule_time(new_page) # 判断文章是否发布成功 await self.context.storage_state(path=self.cookie_path) # 保存cookie self.logger.info(f'{self.user},{self.platform} cookie更新完毕！') await asyncio.sleep(1) # 这里延迟是为了方便眼睛直观的观看 await new_page.get_by_role("button", name="继续发表").click() await asyncio.sleep(3) max_attempts = 10 for attempt in range(max_attempts): try: await new_page.wait_for_url(lambda url: url.startswith(PUBLISH_SECCESSFULL_URL), timeout=50000) self.logger.info(f"{self.user},{self.path_media}文章发布成功，页面已跳转到首页！(尝试次数: {attempt + 1})") return True except TimeoutError: print(f"尝试 {attempt + 1}/{max_attempts}：未跳转到目标页面，继续等待...") if attempt == max_attempts - 1: print("达到最大尝试次数，文章发布失败！") return False except Exception as e: self.logger.error(f"{self.user} upload docx {self.path_media} fails, error is :{e}") return False