1. add tencent video uploader
This commit is contained in:
pookz@stme
@@ -0,0 +1,28 @@
|
||||
import asyncio
|
||||
from pathlib import Path
|
||||
|
||||
from conf import BASE_DIR
|
||||
from tencent_uploader.main import weixin_setup, TencentVideo
|
||||
from utils.constant import TencentZoneTypes
|
||||
from utils.files_times import generate_schedule_time_next_day, get_title_and_hashtags
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
filepath = Path(BASE_DIR) / "videos"
|
||||
account_file = Path(BASE_DIR / "tencent_uploader" / "account.json")
|
||||
# 获取视频目录
|
||||
folder_path = Path(filepath)
|
||||
# 获取文件夹中的所有文件
|
||||
files = list(folder_path.glob("*.mp4"))
|
||||
file_num = len(files)
|
||||
publish_datetimes = generate_schedule_time_next_day(file_num, 1, daily_times=[16])
|
||||
cookie_setup = asyncio.run(weixin_setup(account_file, handle=True))
|
||||
category = TencentZoneTypes.LIFESTYLE.value # 标记原创需要否则不需要传
|
||||
for index, file in enumerate(files):
|
||||
title, tags = get_title_and_hashtags(str(file))
|
||||
# 打印视频文件名、标题和 hashtag
|
||||
print(f"视频文件名：{file}")
|
||||
print(f"标题：{title}")
|
||||
print(f"Hashtag：{tags}")
|
||||
app = TencentVideo(title, file, tags, publish_datetimes[index], account_file, category)
|
||||
asyncio.run(app.main(), debug=False)
|
||||
@@ -0,0 +1,264 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
import json
|
||||
from datetime import datetime
|
||||
|
||||
from playwright.async_api import Playwright, async_playwright
|
||||
import os
|
||||
import asyncio
|
||||
|
||||
from utils.files_times import get_absolute_path
|
||||
|
||||
|
||||
def format_str_for_short_title(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 cookie_auth(account_file):
|
||||
async with async_playwright() as playwright:
|
||||
browser = await playwright.chromium.launch(headless=True)
|
||||
context = await browser.new_context(storage_state=account_file)
|
||||
# 创建一个新的页面
|
||||
page = await context.new_page()
|
||||
# 访问指定的 URL
|
||||
await page.goto("https://channels.weixin.qq.com/platform/post/create")
|
||||
try:
|
||||
await page.wait_for_selector('div.title-name:has-text("视频号小店")', timeout=5000) # 等待5秒
|
||||
print("[+] 等待5秒 cookie 失效")
|
||||
return False
|
||||
except:
|
||||
print("[+] cookie 有效")
|
||||
return True
|
||||
|
||||
|
||||
async def save_storage_state(account_file: str):
|
||||
async with async_playwright() as p:
|
||||
browser = await p.chromium.launch(headless=False)
|
||||
context = await browser.new_context()
|
||||
page = await context.new_page()
|
||||
await page.goto("https://channels.weixin.qq.com")
|
||||
print("请在浏览器中扫码登录...")
|
||||
await asyncio.sleep(20) # 给用户60秒时间进行扫码登录
|
||||
|
||||
# 保存存储状态到文件
|
||||
storage_state = await context.storage_state()
|
||||
with open(account_file, 'w') as f:
|
||||
f.write(json.dumps(storage_state))
|
||||
await browser.close()
|
||||
|
||||
|
||||
async def weixin_setup(account_file, handle=False):
|
||||
account_file = get_absolute_path(account_file, "tencent_uploader")
|
||||
if not os.path.exists(account_file) or not await cookie_auth(account_file):
|
||||
if not handle:
|
||||
# Todo alert message
|
||||
return False
|
||||
print('[+] cookie文件不存在或已失效，即将自动打开浏览器，请扫码登录，登陆后会自动生成cookie文件')
|
||||
# await save_storage_state(account_file)
|
||||
os.system('python3 -m playwright install')
|
||||
os.system(f'playwright codegen channels.weixin.qq.com --save-storage={account_file}') # 生成cookie文件
|
||||
return True
|
||||
|
||||
|
||||
class TencentVideo(object):
|
||||
def __init__(self, title, file_path, tags, publish_date: datetime, account_file, category=None):
|
||||
self.title = title # 视频标题
|
||||
self.file_path = file_path
|
||||
self.tags = tags
|
||||
self.publish_date = publish_date
|
||||
self.account_file = account_file
|
||||
self.category = category
|
||||
self.local_executable_path = "C:/Program Files/Google/Chrome/Application/chrome.exe"
|
||||
self.date_format = '%Y年%m月%d日 %H:%M'
|
||||
|
||||
async def set_schedule_time_tencent(self, page, publish_date):
|
||||
print("click schedule")
|
||||
|
||||
label_element = page.locator("label").filter(has_text="定时").nth(1)
|
||||
await label_element.click()
|
||||
|
||||
await page.click('input[placeholder="请选择发表时间"]')
|
||||
|
||||
current_month = str(publish_date.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(publish_date.day):
|
||||
await element.click()
|
||||
break
|
||||
|
||||
# 输入小时部分（假设选择11小时）
|
||||
await page.click('input[placeholder="请选择时间"]')
|
||||
await page.keyboard.press("Control+KeyA")
|
||||
await page.keyboard.type(str(publish_date.hour))
|
||||
|
||||
# 选择标题栏（令定时时间生效）
|
||||
await page.locator("div.input-editor").click()
|
||||
|
||||
async def handle_upload_error(self, page):
|
||||
print("视频出错了，重新上传中")
|
||||
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.file_path)
|
||||
|
||||
async def upload(self, playwright: Playwright) -> None:
|
||||
# 使用 Chromium (这里使用系统内浏览器，用chromium 会造成h264错误
|
||||
browser = await playwright.chromium.launch(headless=False, executable_path=self.local_executable_path)
|
||||
# 创建一个浏览器上下文，使用指定的 cookie 文件
|
||||
context = await browser.new_context(storage_state=f"{self.account_file}")
|
||||
|
||||
# 创建一个新的页面
|
||||
page = await context.new_page()
|
||||
# 访问指定的 URL
|
||||
await page.goto("https://channels.weixin.qq.com/platform/post/create")
|
||||
print('[+]正在上传-------{}.mp4'.format(self.title))
|
||||
# 等待页面跳转到指定的 URL，没进入，则自动等待到超时
|
||||
await page.wait_for_url("https://channels.weixin.qq.com/platform/post/create")
|
||||
# await page.wait_for_selector('input[type="file"]', timeout=10000)
|
||||
file_input = page.locator('input[type="file"]')
|
||||
await file_input.set_input_files(self.file_path)
|
||||
# 填充标题和话题
|
||||
await self.add_title_tags(page)
|
||||
# 添加商品
|
||||
# await self.add_product(page)
|
||||
# 合集功能
|
||||
await self.add_collection(page)
|
||||
# 原创选择
|
||||
await self.add_original(page)
|
||||
# 检测上传状态
|
||||
await self.detact_upload_status(page)
|
||||
if self.publish_date != 0:
|
||||
await self.set_schedule_time_tencent(page, self.publish_date)
|
||||
# 添加短标题
|
||||
await self.add_short_title(page)
|
||||
|
||||
await self.click_publish(page)
|
||||
|
||||
await context.storage_state(path=f"{self.account_file}") # 保存cookie
|
||||
print(' [-]cookie更新完毕！')
|
||||
await asyncio.sleep(2) # 这里延迟是为了方便眼睛直观的观看
|
||||
# 关闭浏览器上下文和浏览器实例
|
||||
await context.close()
|
||||
await browser.close()
|
||||
print('[+]正在监控执行计划中.......')
|
||||
|
||||
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 = format_str_for_short_title(self.title)
|
||||
await short_title_element.fill(short_title)
|
||||
|
||||
async def click_publish(self, page):
|
||||
while True:
|
||||
try:
|
||||
publish_buttion = page.locator('div.form-btns button:has-text("发表")')
|
||||
if await publish_buttion.count():
|
||||
await publish_buttion.click()
|
||||
await page.wait_for_url("https://channels.weixin.qq.com/platform/post/list", timeout=1500)
|
||||
print(" [-]视频发布成功")
|
||||
break
|
||||
except Exception as e:
|
||||
current_url = page.url
|
||||
if "https://channels.weixin.qq.com/platform/post/list" in current_url:
|
||||
print(" [-]视频发布成功")
|
||||
break
|
||||
else:
|
||||
print(f" [-] Exception: {e}")
|
||||
print(" [-] 视频正在发布中...")
|
||||
await page.screenshot(full_page=True)
|
||||
await asyncio.sleep(0.5)
|
||||
|
||||
async def detact_upload_status(self, page):
|
||||
while True:
|
||||
# 匹配删除按钮，代表视频上传完毕，如果不存在，代表视频正在上传，则等待
|
||||
try:
|
||||
# 匹配删除按钮，代表视频上传完毕
|
||||
if "weui-desktop-btn_disabled" not in await page.get_by_role("button", name="发表").get_attribute(
|
||||
'class'):
|
||||
print(" [-]视频上传完毕")
|
||||
break
|
||||
else:
|
||||
print(" [-] 正在上传视频中...")
|
||||
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():
|
||||
print(" [-] 发现上传出错了...")
|
||||
await self.handle_upload_error(page)
|
||||
except:
|
||||
print(" [-] 正在上传视频中...")
|
||||
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")
|
||||
print(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 asyncio.sleep(1)
|
||||
if await page.locator('button:has-text("声明原创"):visible').count():
|
||||
await page.locator('button:has-text("声明原创"):visible').click()
|
||||
|
||||
async def main(self):
|
||||
async with async_playwright() as playwright:
|
||||
await self.upload(playwright)
|
||||
@@ -0,0 +1,28 @@
|
||||
import enum
|
||||
|
||||
|
||||
class TencentZoneTypes(enum.Enum):
|
||||
LIFESTYLE = '生活'
|
||||
CUTE_KIDS = '萌娃'
|
||||
MUSIC = '音乐'
|
||||
KNOWLEDGE = '知识'
|
||||
EMOTION = '情感'
|
||||
TRAVEL_SCENERY = '旅行风景'
|
||||
FASHION = '时尚'
|
||||
FOOD = '美食'
|
||||
LIFE_HACKS = '生活技巧'
|
||||
DANCE = '舞蹈'
|
||||
MOVIES_TV_SHOWS = '影视综艺'
|
||||
SPORTS = '运动'
|
||||
FUNNY = '搞笑'
|
||||
CELEBRITIES = '明星名人'
|
||||
NEWS_INFO = '新闻资讯'
|
||||
GAMING = '游戏'
|
||||
AUTOMOTIVE = '车'
|
||||
ANIME = '二次元'
|
||||
TALENT = '才艺'
|
||||
CUTE_PETS = '萌宠'
|
||||
INDUSTRY_MACHINERY_CONSTRUCTION = '机械'
|
||||
ANIMALS = '动物'
|
||||
PARENTING = '育儿'
|
||||
TECHNOLOGY = '科技'
|
||||
Binary file not shown.
Reference in New Issue
Block a user