# -*- 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
|
