diff --git a/examples/upload_video_to_tencent_video.py b/examples/upload_video_to_tencent_video.py new file mode 100644 index 0000000..b80b15c --- /dev/null +++ b/examples/upload_video_to_tencent_video.py @@ -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) diff --git a/tencent_uploader/__init__.py b/tencent_uploader/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tencent_uploader/main.py b/tencent_uploader/main.py new file mode 100644 index 0000000..c05619d --- /dev/null +++ b/tencent_uploader/main.py @@ -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) diff --git a/utils/constant.py b/utils/constant.py new file mode 100644 index 0000000..c8e9523 --- /dev/null +++ b/utils/constant.py @@ -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 = '科技' diff --git a/videos/demo.mp4 b/videos/demo.mp4 index c36300f..8b63326 100644 Binary files a/videos/demo.mp4 and b/videos/demo.mp4 differ