1. add tencent video uploader

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)

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)

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 = '科技'