diff --git a/examples/get_tencent_cookie.py b/examples/get_tencent_cookie.py index 06361c4..4d5db2f 100644 --- a/examples/get_tencent_cookie.py +++ b/examples/get_tencent_cookie.py @@ -2,9 +2,10 @@ import asyncio from pathlib import Path from conf import BASE_DIR -from uploader.tencent_uploader.main import weixin_setup +from uploader.tencent_uploader.main import tencent_setup -if __name__ == '__main__': + +if __name__ == "__main__": account_file = Path(BASE_DIR / "cookies" / "tencent_uploader" / "account.json") account_file.parent.mkdir(exist_ok=True) - cookie_setup = asyncio.run(weixin_setup(str(account_file), handle=True)) + asyncio.run(tencent_setup(str(account_file), handle=True)) diff --git a/examples/upload_video_to_tencent.py b/examples/upload_video_to_tencent.py index 2e47a1e..16a8320 100644 --- a/examples/upload_video_to_tencent.py +++ b/examples/upload_video_to_tencent.py @@ -1,28 +1,109 @@ +""" +当前文件先保留为视频号 uploader 的调试入口 / 历史直连路径。 + +注意： +1. `uploader/tencent_uploader/main.py` 里的核心页面交互逻辑目前是骨架； +2. 你需要自己补视频里的 `fill_title_and_tags` / `wait_for_upload_complete` / `set_thumbnail` / `submit_publish` 等方法； +3. 你需要自己补图文里的 `switch_to_note_mode` / `upload_note_images` / `fill_note_title_and_tags` / `submit_publish` 等方法； +3. 补完后，这个 example 就可以直接作为本地调试入口继续使用。 +""" + import asyncio +from datetime import datetime, timedelta from pathlib import Path from conf import BASE_DIR -from uploader.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 +from uploader.tencent_uploader.main import TENCENT_PUBLISH_STRATEGY_IMMEDIATE +from uploader.tencent_uploader.main import TENCENT_PUBLISH_STRATEGY_SCHEDULED +from uploader.tencent_uploader.main import TencentNote +from uploader.tencent_uploader.main import TencentVideo -if __name__ == '__main__': - filepath = Path(BASE_DIR) / "videos" - account_file = Path(BASE_DIR / "cookies" / "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) +ACCOUNT_FILE = Path(BASE_DIR / "cookies" / "tencent_uploader" / "account.json") + + +def upload_video_to_tencent(): + video_file = Path(BASE_DIR) / "videos" / "demo.mp4" + thumbnail_path = video_file.with_suffix(".png") + app = TencentVideo( + title="视频号视频示例", + file_path=str(video_file), + tags=["视频号", "自动上传", "调试入口"], + publish_strategy=TENCENT_PUBLISH_STRATEGY_IMMEDIATE, + publish_date=0, + account_file=str(ACCOUNT_FILE), + desc="这里是你后面准备填写的视频简介示例", + thumbnail_path=str(thumbnail_path) if thumbnail_path.exists() else None, + short_title="视频号示例", + category=None, + is_draft=False, + ) + asyncio.run(app.tencent_upload_video()) + + +def upload_video_to_tencent_scheduled(): + video_file = Path(BASE_DIR) / "videos" / "demo.mp4" + thumbnail_path = video_file.with_suffix(".png") + publish_time = (datetime.now() + timedelta(hours=3)).replace(second=0, microsecond=0) + app = TencentVideo( + title="视频号定时发布示例", + file_path=str(video_file), + tags=["视频号", "定时发布", "调试入口"], + publish_strategy=TENCENT_PUBLISH_STRATEGY_SCHEDULED, + publish_date=publish_time, + account_file=str(ACCOUNT_FILE), + desc="这里是定时发布的视频简介示例", + thumbnail_path=str(thumbnail_path) if thumbnail_path.exists() else None, + short_title="定时发布示例", + category=None, + is_draft=False, + ) + asyncio.run(app.tencent_upload_video()) + + +def upload_note_to_tencent(): + image_candidates = [ + Path(BASE_DIR) / "videos" / "demo.png", + Path(BASE_DIR) / "videos" / "demo1.png", + Path(BASE_DIR) / "videos" / "demo2.png", + ] + image_paths = [str(path) for path in image_candidates if path.exists()] + app = TencentNote( + image_paths=image_paths, + note="视频号图文内容示例 #图文调试", + tags=["视频号图文", "自动上传", "调试入口"], + publish_strategy=TENCENT_PUBLISH_STRATEGY_IMMEDIATE, + publish_date=0, + account_file=str(ACCOUNT_FILE), + title="视频号图文示例", + is_draft=False, + ) + asyncio.run(app.tencent_upload_note()) + + +def upload_note_to_tencent_scheduled(): + image_candidates = [ + Path(BASE_DIR) / "videos" / "demo.png", + Path(BASE_DIR) / "videos" / "demo1.png", + Path(BASE_DIR) / "videos" / "demo2.png", + ] + image_paths = [str(path) for path in image_candidates if path.exists()] + publish_time = (datetime.now() + timedelta(hours=3)).replace(second=0, microsecond=0) + app = TencentNote( + image_paths=image_paths, + note="视频号图文定时发布示例 #图文调试", + tags=["视频号图文", "定时发布", "调试入口"], + publish_strategy=TENCENT_PUBLISH_STRATEGY_SCHEDULED, + publish_date=publish_time, + account_file=str(ACCOUNT_FILE), + title="视频号图文定时示例", + is_draft=False, + ) + asyncio.run(app.tencent_upload_note()) + + +if __name__ == "__main__": + upload_video_to_tencent() + # upload_video_to_tencent_scheduled() + # upload_note_to_tencent() + # upload_note_to_tencent_scheduled() diff --git a/uploader/tencent_uploader/__init__.py b/uploader/tencent_uploader/__init__.py index 9860b4a..c4da2de 100644 --- a/uploader/tencent_uploader/__init__.py +++ b/uploader/tencent_uploader/__init__.py @@ -2,4 +2,30 @@ from pathlib import Path from conf import BASE_DIR -Path(BASE_DIR / "cookies" / "tencent_uploader").mkdir(exist_ok=True) \ No newline at end of file +Path(BASE_DIR / "cookies" / "tencent_uploader").mkdir(exist_ok=True) + +from uploader.tencent_uploader.main import TENCENT_PUBLISH_STRATEGY_IMMEDIATE +from uploader.tencent_uploader.main import TENCENT_PUBLISH_STRATEGY_SCHEDULED +from uploader.tencent_uploader.main import TencentBaseUploader +from uploader.tencent_uploader.main import TencentNote +from uploader.tencent_uploader.main import TencentVideo +from uploader.tencent_uploader.main import cookie_auth +from uploader.tencent_uploader.main import format_str_for_short_title +from uploader.tencent_uploader.main import get_tencent_cookie +from uploader.tencent_uploader.main import tencent_cookie_gen +from uploader.tencent_uploader.main import tencent_setup +from uploader.tencent_uploader.main import weixin_setup + +__all__ = [ + "TENCENT_PUBLISH_STRATEGY_IMMEDIATE", + "TENCENT_PUBLISH_STRATEGY_SCHEDULED", + "TencentBaseUploader", + "TencentNote", + "TencentVideo", + "cookie_auth", + "format_str_for_short_title", + "get_tencent_cookie", + "tencent_cookie_gen", + "tencent_setup", + "weixin_setup", +] diff --git a/uploader/tencent_uploader/main.py b/uploader/tencent_uploader/main.py index 3f87ff8..8a24140 100644 --- a/uploader/tencent_uploader/main.py +++ b/uploader/tencent_uploader/main.py @@ -1,285 +1,876 @@ # -*- coding: utf-8 -*- -from datetime import datetime +from __future__ import annotations -from playwright.async_api import Playwright, async_playwright -import os import asyncio +import inspect +import os +from datetime import datetime +from pathlib import Path -from conf import LOCAL_CHROME_PATH, LOCAL_CHROME_HEADLESS +from patchright.async_api import Page +from patchright.async_api import Playwright +from patchright.async_api import async_playwright + +from conf import BASE_DIR, DEBUG_MODE, LOCAL_CHROME_HEADLESS, LOCAL_CHROME_PATH +from uploader.base_video import BaseVideoUploader from utils.base_social_media import set_init_script -from utils.files_times import get_absolute_path from utils.log import tencent_logger +TENCENT_LOGIN_URL = "https://channels.weixin.qq.com" +TENCENT_UPLOAD_URL = "https://channels.weixin.qq.com/platform/post/create" +TENCENT_MANAGE_URL = "https://channels.weixin.qq.com/platform/post/list" +TENCENT_PUBLISH_STRATEGY_IMMEDIATE = "immediate" +TENCENT_PUBLISH_STRATEGY_SCHEDULED = "scheduled" + + +def _msg(emoji: str, text: str) -> str: + return f"{emoji} {text}" + + +def _resolve_account_file(account_file: str | Path) -> str: + path = Path(account_file).expanduser() + if path.is_absolute(): + return str(path) + + if len(path.parts) == 1: + return str((Path(BASE_DIR) / "cookies" / "tencent_uploader" / path).resolve()) + + return str(path.resolve()) + + +async def _emit_qrcode_callback(qrcode_callback, payload: dict): + if not qrcode_callback: + return + + callback_result = qrcode_callback(payload) + if inspect.isawaitable(callback_result): + await callback_result + + +def _build_login_result( + success: bool, + status: str, + message: str, + account_file: str, + qrcode: dict | None = None, + current_url: str = "", +) -> dict: + return { + "success": success, + "status": status, + "message": message, + "account_file": str(account_file), + "qrcode": qrcode, + "current_url": current_url, + } + + +def _build_launch_kwargs(headless: bool) -> dict: + launch_kwargs = {"headless": headless} + if LOCAL_CHROME_PATH: + launch_kwargs["executable_path"] = LOCAL_CHROME_PATH + else: + launch_kwargs["channel"] = "chrome" + return launch_kwargs + + +def _get_qrcode_utils(): + from utils.login_qrcode import build_login_qrcode_path + from utils.login_qrcode import decode_qrcode_from_path + from utils.login_qrcode import print_terminal_qrcode + from utils.login_qrcode import remove_qrcode_file + from utils.login_qrcode import save_data_url_image + + return { + "build_login_qrcode_path": build_login_qrcode_path, + "decode_qrcode_from_path": decode_qrcode_from_path, + "print_terminal_qrcode": print_terminal_qrcode, + "remove_qrcode_file": remove_qrcode_file, + "save_data_url_image": save_data_url_image, + } + 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) - # 移除不允许的特殊字符 - 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)) + formatted_string += " " * (6 - len(formatted_string)) return formatted_string async def cookie_auth(account_file): + account_file = _resolve_account_file(account_file) async with async_playwright() as playwright: - browser = await playwright.chromium.launch(headless=LOCAL_CHROME_HEADLESS) - context = await browser.new_context(storage_state=account_file) - context = await set_init_script(context) - # 创建一个新的页面 - page = await context.new_page() - # 访问指定的 URL - await page.goto("https://channels.weixin.qq.com/platform/post/create") + browser = await playwright.chromium.launch(**_build_launch_kwargs(headless=True)) try: - await page.wait_for_selector('div.title-name:has-text("微信小店")', timeout=5000) # 等待5秒 - tencent_logger.error("[+] 等待5秒 cookie 失效") - return False - except: - tencent_logger.success("[+] cookie 有效") + context = await browser.new_context(storage_state=account_file) + context = await set_init_script(context) + page = await context.new_page() + await page.goto(TENCENT_UPLOAD_URL) + await page.wait_for_url(TENCENT_UPLOAD_URL, timeout=5000) + + login_markers = [ + page.get_by_text("扫码登录", exact=True).first, + page.get_by_text("发表视频", exact=True).first, + page.get_by_role("button", name="发表").first, + ] + + if await login_markers[0].count(): + tencent_logger.info(_msg("🥹", "cookie 已失效，得重新登录一下")) + return False + + tencent_logger.success(_msg("🥳", "cookie 有效")) return True - - -async def get_tencent_cookie(account_file): - async with async_playwright() as playwright: - options = { - 'args': [ - '--lang en-GB' - ], - 'headless': LOCAL_CHROME_HEADLESS, # Set headless option here - } - # Make sure to run headed. - browser = await playwright.chromium.launch(**options) - # Setup context however you like. - context = await browser.new_context() # Pass any options - # Pause the page, and start recording manually. - context = await set_init_script(context) - page = await context.new_page() - await page.goto("https://channels.weixin.qq.com") - await page.pause() - # 点击调试器的继续，保存cookie - await context.storage_state(path=account_file) - - -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 + except Exception as exc: + tencent_logger.warning(_msg("😵", f"cookie 校验时出错，按失效处理: {exc}")) return False - tencent_logger.info('[+] cookie文件不存在或已失效，即将自动打开浏览器，请扫码登录，登陆后会自动生成cookie文件') - await get_tencent_cookie(account_file) + finally: + await browser.close() + + +async def _extract_tencent_qrcode_src(page: Page) -> str: + if hasattr(page, "frame_locator"): + try: + iframe_locator = page.frame_locator('[src*="login-for-iframe"]') + qr_code_img = iframe_locator.locator('div#app img.qrcode').first + await qr_code_img.wait_for(state="visible", timeout=30000) + src = await qr_code_img.get_attribute("src") + if src and src.startswith("data:image/"): + return src + except Exception: + pass + + selector_candidates = [ + "div.login-qrcode-wrap img.qrcode", + "div.qrcode-wrap img.qrcode", + "img.qrcode", + 'img[src^="data:image/"]', + ] + for selector in selector_candidates: + qr_code_img = page.locator(selector).first + try: + if not await qr_code_img.count() or not await qr_code_img.is_visible(): + continue + src = await qr_code_img.get_attribute("src") + if src and src.startswith("data:image/"): + return src + except Exception: + continue + + raise RuntimeError("未获取到视频号登录二维码地址") + + +async def _save_tencent_qrcode(page: Page, account_file: str, previous_qrcode_path: Path | None = None, qrcode_callback=None) -> dict: + qrcode_utils = _get_qrcode_utils() + qrcode_src = await _extract_tencent_qrcode_src(page) + qrcode_path = qrcode_utils["save_data_url_image"]( + qrcode_src, + qrcode_utils["build_login_qrcode_path"](account_file, suffix="tencent_login_qrcode"), + ) + if previous_qrcode_path and previous_qrcode_path != qrcode_path: + if qrcode_utils["remove_qrcode_file"](previous_qrcode_path): + tencent_logger.info(_msg("🧹", f"临时二维码文件已清理: {previous_qrcode_path}")) + + tencent_logger.info(_msg("🖼️", f"二维码已经准备好啦，已保存到: {qrcode_path}")) + qrcode_content = qrcode_utils["decode_qrcode_from_path"](qrcode_path) + if qrcode_content: + qrcode_utils["print_terminal_qrcode"](qrcode_content, qrcode_path, "微信") + else: + tencent_logger.warning( + _msg( + "😵", + f"没能从二维码图片里解析出可打印内容，所以这次没法在终端重绘二维码；请直接打开 {qrcode_path} 扫码", + ) + ) + + qrcode_info = { + "image_path": str(qrcode_path), + "image_data_url": qrcode_src, + } + await _emit_qrcode_callback(qrcode_callback, qrcode_info) + return qrcode_info + + +async def _is_tencent_login_completed(page: Page) -> bool: + publish_markers = [ + page.locator('div:has-text("发表视频")').first, + page.locator('button:has-text("发表")').first, + page.locator('button:has-text("保存草稿")').first, + ] + for marker in publish_markers: + try: + if await marker.count() and await marker.is_visible(): + return True + except Exception: + continue + + if not (page.url.startswith(TENCENT_UPLOAD_URL) or page.url.startswith(TENCENT_MANAGE_URL)): + return False + + login_markers = [ + page.locator("div.login-qrcode-wrap").first, + page.locator("div.qrcode-wrap").first, + page.locator("img.qrcode").first, + page.locator('span:has-text("微信扫码登录 视频号助手")').first, + ] + for marker in login_markers: + try: + if await marker.count() and await marker.is_visible(): + return False + except Exception: + continue + return True -class TencentVideo(object): - def __init__(self, title, file_path, tags, publish_date: datetime, account_file, category=None, is_draft=False): - 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.headless = LOCAL_CHROME_HEADLESS - self.is_draft = is_draft # 是否保存为草稿 - self.local_executable_path = LOCAL_CHROME_PATH or None +async def _is_tencent_qrcode_expired(page: Page) -> bool: + tip_selectors = [ + 'div.mask.show p.refresh-tip:has-text("二维码已过期，点击刷新")', + 'div.mask.show p.refresh-tip:has-text("网络不可用，点击刷新")', + 'p.refresh-tip:has-text("二维码已过期，点击刷新")', + 'p.refresh-tip:has-text("网络不可用，点击刷新")', + ] + for selector in tip_selectors: + tip = page.locator(selector).first + try: + if await tip.count() and await tip.is_visible(): + return True + except Exception: + continue + return False - async def set_schedule_time_tencent(self, page, publish_date): + +async def _is_tencent_qrcode_scanned(page: Page) -> bool: + scanned_tips = [ + 'div.qr-tip div:has-text("已扫码")', + 'div.qr-tip div:has-text("需在手机上进行确认")', + ] + for selector in scanned_tips: + tip = page.locator(selector).first + try: + if await tip.count() and await tip.is_visible(): + return True + except Exception: + continue + return False + + +async def _refresh_tencent_qrcode(page: Page) -> None: + visible_refresh_selectors = [ + "div.login-qrcode-wrap div.mask.show div.refresh-wrap", + "div.login-qrcode-wrap div.mask.show .refresh-wrap", + ] + for selector in visible_refresh_selectors: + refresh_wrap = page.locator(selector).first + try: + if not await refresh_wrap.count() or not await refresh_wrap.is_visible(): + continue + await refresh_wrap.click() + return + except Exception: + continue + + tip_selectors = [ + 'div.mask.show p.refresh-tip:has-text("二维码已过期，点击刷新")', + 'div.mask.show p.refresh-tip:has-text("网络不可用，点击刷新")', + 'p.refresh-tip:has-text("二维码已过期，点击刷新")', + 'p.refresh-tip:has-text("网络不可用，点击刷新")', + ] + for selector in tip_selectors: + tip = page.locator(selector).first + try: + if not await tip.count() or not await tip.is_visible(): + continue + refresh_wrap = tip.locator("xpath=ancestor::div[contains(@class, 'refresh-wrap')]").first + if await refresh_wrap.count(): + await refresh_wrap.click() + else: + await tip.click() + return + except Exception: + continue + + fallback_refresh = page.locator("div.login-qrcode-wrap div.refresh-wrap").first + if await fallback_refresh.count(): + await fallback_refresh.click() + return + + raise RuntimeError("未找到可点击的视频号二维码刷新区域") + + +async def _wait_for_tencent_login( + page: Page, + account_file: str, + qrcode_info: dict, + qrcode_callback=None, + poll_interval: int = 3, + max_checks: int = 100, +) -> dict: + qrcode_path = Path(qrcode_info["image_path"]) + scanned_logged = False + for _ in range(max_checks): + if await _is_tencent_login_completed(page): + tencent_logger.info(_msg("🥳", f"扫码成功，已经跳转到登录后页面: {page.url}")) + return _build_login_result(True, "success", "视频号扫码登录成功", account_file, qrcode_info, page.url) + + if not scanned_logged and await _is_tencent_qrcode_scanned(page): + tencent_logger.info(_msg("📱", "已经扫码啦，还差手机端确认一下")) + scanned_logged = True + + if await _is_tencent_qrcode_expired(page): + tencent_logger.warning(_msg("😵", "二维码失效了，小人马上去刷新")) + await _refresh_tencent_qrcode(page) + await asyncio.sleep(1) + qrcode_info = await _save_tencent_qrcode( + page, + account_file, + previous_qrcode_path=qrcode_path, + qrcode_callback=qrcode_callback, + ) + qrcode_path = Path(qrcode_info["image_path"]) + + await asyncio.sleep(poll_interval) + + return _build_login_result(False, "timeout", "等待视频号扫码登录超时", account_file, qrcode_info, page.url) + + +async def tencent_cookie_gen( + account_file, + qrcode_callback=None, + poll_interval: int = 3, + max_checks: int = 100, + headless: bool = LOCAL_CHROME_HEADLESS, +): + account_file = _resolve_account_file(account_file) + Path(account_file).parent.mkdir(parents=True, exist_ok=True) + + async with async_playwright() as playwright: + browser = await playwright.chromium.launch(**_build_launch_kwargs(headless=headless)) + context = await browser.new_context() + qrcode_path = None + result = _build_login_result(False, "failed", "视频号登录失败", account_file) + try: + page = await context.new_page() + await page.goto(TENCENT_LOGIN_URL) + qrcode_info = await _save_tencent_qrcode(page, account_file, qrcode_callback=qrcode_callback) + qrcode_path = Path(qrcode_info["image_path"]) + tencent_logger.info(_msg("🧍", "请扫码，小人正在耐心等待登录完成")) + result = await _wait_for_tencent_login( + page, + account_file, + qrcode_info, + qrcode_callback=qrcode_callback, + poll_interval=poll_interval, + max_checks=max_checks, + ) + if result["success"]: + await asyncio.sleep(2) + await context.storage_state(path=account_file) + if not await cookie_auth(account_file): + result = _build_login_result( + False, + "cookie_invalid", + "视频号扫码流程结束，但 cookie 校验失败", + account_file, + qrcode_info, + page.url, + ) + return result + except Exception as exc: + result = _build_login_result( + False, + "failed", + str(exc), + account_file, + current_url=page.url if "page" in locals() else "", + ) + return result + finally: + qrcode_utils = _get_qrcode_utils() + if qrcode_utils["remove_qrcode_file"](qrcode_path): + tencent_logger.info(_msg("🧹", f"临时二维码文件已清理: {qrcode_path}")) + if not result["success"]: + tencent_logger.error(_msg("😢", f"登录失败: {result['message']}")) + await context.close() + await browser.close() + + +async def tencent_setup( + account_file, + handle=False, + return_detail=False, + qrcode_callback=None, + headless: bool = LOCAL_CHROME_HEADLESS, +): + account_file = _resolve_account_file(account_file) + if not os.path.exists(account_file) or not await cookie_auth(account_file): + if not handle: + result = _build_login_result(False, "cookie_invalid", "cookie文件不存在或已失效", account_file) + return result if return_detail else False + + tencent_logger.info(_msg("🥹", "cookie 失效了，准备打开浏览器重新登录")) + result = await tencent_cookie_gen(account_file, qrcode_callback=qrcode_callback, headless=headless) + return result if return_detail else result["success"] + + result = _build_login_result(True, "cookie_valid", "cookie有效", account_file) + return result if return_detail else True + + +async def get_tencent_cookie(account_file, qrcode_callback=None, headless: bool = LOCAL_CHROME_HEADLESS): + return await tencent_cookie_gen(account_file, qrcode_callback=qrcode_callback, headless=headless) + + +async def weixin_setup( + account_file, + handle=False, + return_detail=False, + qrcode_callback=None, + headless: bool = LOCAL_CHROME_HEADLESS, +): + return await tencent_setup( + account_file, + handle=handle, + return_detail=return_detail, + qrcode_callback=qrcode_callback, + headless=headless, + ) + + +class TencentBaseUploader(BaseVideoUploader): + def __init__( + self, + publish_date: datetime | int, + account_file, + publish_strategy: str = TENCENT_PUBLISH_STRATEGY_IMMEDIATE, + debug: bool = DEBUG_MODE, + headless: bool = LOCAL_CHROME_HEADLESS, + ): + self.publish_date = publish_date + self.account_file = _resolve_account_file(account_file) + self.publish_strategy = publish_strategy + self.debug = debug + self.headless = headless + self.local_executable_path = LOCAL_CHROME_PATH + + async def validate_base_args(self): + if not os.path.exists(self.account_file): + raise RuntimeError(f"cookie文件不存在，请先完成视频号登录: {self.account_file}") + if not await cookie_auth(self.account_file): + raise RuntimeError(f"cookie文件已失效，请先完成视频号登录: {self.account_file}") + if self.publish_strategy not in {TENCENT_PUBLISH_STRATEGY_IMMEDIATE, TENCENT_PUBLISH_STRATEGY_SCHEDULED}: + raise ValueError(f"不支持的发布策略: {self.publish_strategy}") + + if self.publish_strategy == TENCENT_PUBLISH_STRATEGY_SCHEDULED: + self.publish_date = self.validate_publish_date(self.publish_date) + else: + self.publish_date = 0 + + async def set_schedule_time_tencent(self, page: Page, publish_date: datetime): label_element = page.locator("label").filter(has_text="定时").nth(1) await label_element.click() - await page.click('input[placeholder="请选择发表时间"]') - str_month = str(publish_date.month) if publish_date.month > 9 else "0" + str(publish_date.month) - current_month = str_month + "月" - # 获取当前的月份 + current_month = publish_date.strftime("%m月") 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') + await page.click("button.weui-desktop-btn__icon__right") - # 获取页面元素 - elements = await page.query_selector_all('table.weui-desktop-picker__table a') - - # 遍历元素并点击匹配的元素 + 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'): + 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.keyboard.type(publish_date.strftime("%H")) await page.locator("div.input-editor").click() - async def handle_upload_error(self, page): - tencent_logger.info("视频出错了，重新上传中") - await page.locator('div.media-status-content div.tag-inner:has-text("删除")').click() - await page.get_by_role('button', name="删除", exact=True).click() + async def open_upload_page(self, page: Page) -> None: + await page.goto(TENCENT_UPLOAD_URL) + await page.wait_for_url(TENCENT_UPLOAD_URL) + + async def upload_video_file(self, page: Page, file_path: str) -> None: file_input = page.locator('input[type="file"]') - await file_input.set_input_files(self.file_path) + await file_input.set_input_files(file_path) - async def upload(self, playwright: Playwright) -> None: - # 使用 Chromium (这里使用系统内浏览器，用chromium 会造成h264错误 - browser = await playwright.chromium.launch(headless=self.headless, executable_path=self.local_executable_path) - # 创建一个浏览器上下文，使用指定的 cookie 文件 - context = await browser.new_context(storage_state=f"{self.account_file}") - context = await set_init_script(context) - - # 创建一个新的页面 - page = await context.new_page() - # 访问指定的 URL - await page.goto("https://channels.weixin.qq.com/platform/post/create") - tencent_logger.info(f'[+]正在上传-------{self.title}.mp4') - # 等待页面跳转到指定的 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.detect_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 - tencent_logger.success(' [-]cookie更新完毕！') - await asyncio.sleep(2) # 这里延迟是为了方便眼睛直观的观看 - # 关闭浏览器上下文和浏览器实例 - await context.close() - await browser.close() - - 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"]') + async def set_short_title(self, page: Page, title: str, short_title: str | None = None) -> None: + 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) + await short_title_element.fill(short_title or format_str_for_short_title(title)) - async def click_publish(self, page): - while True: - try: - if self.is_draft: - # 点击"保存草稿"按钮 - draft_button = page.locator('div.form-btns button:has-text("保存草稿")') - if await draft_button.count(): - await draft_button.click() - # 等待跳转到草稿箱页面或确认保存成功 - await page.wait_for_url("**/post/list**", timeout=5000) # 使用通配符匹配包含post/list的URL - tencent_logger.success(" [-]视频草稿保存成功") - else: - # 点击"发表"按钮 - publish_button = page.locator('div.form-btns button:has-text("发表")') - if await publish_button.count(): - await publish_button.click() - await page.wait_for_url("https://channels.weixin.qq.com/platform/post/list", timeout=5000) - tencent_logger.success(" [-]视频发布成功") - break - except Exception as e: - current_url = page.url - if self.is_draft: - # 检查是否在草稿相关的页面 - if "post/list" in current_url or "draft" in current_url: - tencent_logger.success(" [-]视频草稿保存成功") - break - else: - # 检查是否在发布列表页面 - if "https://channels.weixin.qq.com/platform/post/list" in current_url: - tencent_logger.success(" [-]视频发布成功") - break - tencent_logger.exception(f" [-] Exception: {e}") - tencent_logger.info(" [-] 视频正在发布中...") - await asyncio.sleep(0.5) - - async def detect_upload_status(self, page): - while True: - # 匹配删除按钮，代表视频上传完毕，如果不存在，代表视频正在上传，则等待 - try: - # 匹配删除按钮，代表视频上传完毕 - if "weui-desktop-btn_disabled" not in await page.get_by_role("button", name="发表").get_attribute( - 'class'): - tencent_logger.info(" [-]视频上传完毕") - break - else: - tencent_logger.info(" [-] 正在上传视频中...") - 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(): - tencent_logger.error(" [-] 发现上传出错了...准备重试") - await self.handle_upload_error(page) - except: - tencent_logger.info(" [-] 正在上传视频中...") - await asyncio.sleep(2) - - async def add_title_tags(self, page): + async def fill_title_and_tags(self, page: Page) -> None: 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): + for tag in self.tags: await page.keyboard.type("#" + tag) await page.keyboard.press("Space") - tencent_logger.info(f"成功添加hashtag: {len(self.tags)}") + tencent_logger.info(_msg("🏷️", 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') + async def fill_description(self, page: Page) -> None: + await page.keyboard.press("Enter") + await page.keyboard.type(self.desc) + tencent_logger.info(_msg("🏷️", f"成功添加 desc: {len(self.desc)}")) + + async def apply_collection(self, page: Page) -> None: + 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): + async def apply_original_statement(self, page: Page) -> None: if await page.get_by_label("视频为原创").count(): await page.get_by_label("视频为原创").check() - # 检查 "我已阅读并同意 《视频号原创声明使用条款》" 元素是否存在 - label_locator = await page.locator('label:has-text("我已阅读并同意 《视频号原创声明使用条款》")').is_visible() + + try: + label_locator = await page.locator('label:has-text("我已阅读并同意 《视频号原创声明使用条款》")').is_visible() + except Exception: + label_locator = False + 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 page.wait_for_timeout(1000) - if await page.locator('button:has-text("声明原创"):visible').count(): - await page.locator('button:has-text("声明原创"):visible').click() - async def main(self): + if await page.locator('div.label span:has-text("声明原创")').count() and getattr(self, "category", None): + 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() + checked_locator = page.locator( + "div.declare-original-dialog " + "label.ant-checkbox-wrapper.ant-checkbox-wrapper-checked:visible" + ) + if not await checked_locator.count(): + await page.locator("div.declare-original-dialog input.ant-checkbox-input:visible").click() + + original_type_form = page.locator('div.original-type-form > div.form-label:has-text("原创类型"):visible') + if await original_type_form.count(): + await page.locator("div.form-content:visible").click() + await page.locator( + "div.form-content:visible " + "ul.weui-desktop-dropdown__list " + f'li.weui-desktop-dropdown__list-ele:has-text("{self.category}")' + ).first.click() + await page.wait_for_timeout(1000) + + declare_button = page.locator('button:has-text("声明原创"):visible') + if await declare_button.count(): + await declare_button.click() + + async def wait_for_upload_complete(self, page: Page) -> None: + while True: + try: + publish_button = page.get_by_role("button", name="发表") + button_class = await publish_button.get_attribute("class") + if button_class and "weui-desktop-btn_disabled" not in button_class: + tencent_logger.info(_msg("🥳", "视频上传完毕")) + break + + tencent_logger.info(_msg("🏃", "正在上传视频中...")) + await asyncio.sleep(2) + + upload_failed = await page.locator("div.status-msg.error").count() + delete_button = await page.locator('div.media-status-content div.tag-inner:has-text("删除")').count() + if upload_failed and delete_button: + tencent_logger.error(_msg("😵", "发现上传出错了，准备重试")) + await self.handle_upload_error(page) + except Exception: + tencent_logger.info(_msg("🏃", "正在上传视频中...")) + await asyncio.sleep(2) + + async def submit_publish(self, page: Page) -> None: + while True: + try: + if getattr(self, "is_draft", False): + draft_button = page.locator('div.form-btns button:has-text("保存草稿")') + if await draft_button.count(): + await draft_button.click() + await page.wait_for_url("**/post/list**", timeout=5000) + tencent_logger.success(_msg("🥳", "视频草稿保存成功")) + else: + publish_button = page.locator('div.form-btns button:has-text("发表")') + if await publish_button.count(): + await publish_button.click() + await page.wait_for_url(TENCENT_MANAGE_URL, timeout=5000) + tencent_logger.success(_msg("🥳", "视频发布成功")) + break + except Exception as exc: + current_url = page.url + if getattr(self, "is_draft", False): + if "post/list" in current_url or "draft" in current_url: + tencent_logger.success(_msg("🥳", "视频草稿保存成功")) + break + else: + if TENCENT_MANAGE_URL in current_url: + tencent_logger.success(_msg("🥳", "视频发布成功")) + break + tencent_logger.exception(f" [-] Exception: {exc}") + tencent_logger.info(_msg("🏃", "视频正在发布中...")) + await asyncio.sleep(0.5) + + +class TencentVideo(TencentBaseUploader): + def __init__( + self, + title, + file_path, + tags, + publish_date: datetime | int, + account_file, + category=None, + is_draft=False, + desc: str | None = None, + thumbnail_path: str | None = None, + short_title: str | None = None, + publish_strategy: str = TENCENT_PUBLISH_STRATEGY_IMMEDIATE, + debug: bool = DEBUG_MODE, + headless: bool = LOCAL_CHROME_HEADLESS, + ): + super().__init__( + publish_date=publish_date, + account_file=account_file, + publish_strategy=publish_strategy, + debug=debug, + headless=headless, + ) + self.title = title + self.file_path = file_path + self.tags = tags or [] + self.category = category + self.is_draft = is_draft + self.desc = desc or "" + self.thumbnail_path = thumbnail_path + self.short_title = short_title + + async def validate_upload_args(self): + await self.validate_base_args() + if not self.title or not str(self.title).strip(): + raise ValueError("视频模式下，title 是必须的") + self.file_path = str(self.validate_video_file(self.file_path)) + if self.thumbnail_path: + self.thumbnail_path = str(self.validate_image_file(self.thumbnail_path)) + + async def handle_upload_error(self, page: Page) -> None: + tencent_logger.info(_msg("😵", "视频出错了，重新上传中")) + await page.locator('div.media-status-content div.tag-inner:has-text("删除")').click() + await page.get_by_role("button", name="删除", exact=True).click() + await self.upload_video_file(page, self.file_path) + + async def set_thumbnail(self, page: Page) -> None: + if not self.thumbnail_path: + return + + tencent_logger.info(_msg("🖼️", "小人准备设置封面")) + + cover_entry_selectors = [ + 'div.vertical-cover-wrap:has-text("个人主页卡片"):has-text("3:4")', + 'div.vertical-cover-wrap:has-text("3:4")', + 'div.vertical-cover-wrap:has-text("个人主页卡片")', + ] + for selector in cover_entry_selectors: + cover_entry = page.locator(selector).first + try: + if not await cover_entry.count(): + continue + await cover_entry.wait_for(state="visible", timeout=3000) + await cover_entry.click() + await page.wait_for_timeout(500) + break + except Exception: + continue + + cover_dialog = page.locator("div.weui-desktop-dialog").filter(has_text="编辑个人主页卡片").first + if not await cover_dialog.count(): + tencent_logger.info(_msg("🧍", "当前页面没有出现封面编辑弹窗，小人先跳过自定义封面")) + return + + try: + await cover_dialog.wait_for(state="visible", timeout=5000) + except Exception: + tencent_logger.warning(_msg("😵", "封面编辑弹窗暂时不可见，这次先跳过自定义封面")) + return + + file_input = cover_dialog.locator('.single-cover-uploader-wrap input[type="file"]').first + await file_input.wait_for(state="attached", timeout=10000) + await file_input.set_input_files(self.thumbnail_path) + await page.wait_for_timeout(1000) + + crop_dialog = page.locator("div.weui-desktop-dialog").filter(has_text="裁剪封面图").first + if await crop_dialog.count(): + try: + await crop_dialog.wait_for(state="visible", timeout=10000) + crop_confirm_button = crop_dialog.locator( + 'div.weui-desktop-dialog__ft button.weui-desktop-btn_primary:has-text("确定")' + ).first + if await crop_confirm_button.count(): + await crop_confirm_button.wait_for(state="visible", timeout=5000) + await crop_confirm_button.click() + await page.wait_for_timeout(1000) + except Exception as exc: + tencent_logger.warning(_msg("😵", f"封面裁剪确认时出错，小人继续尝试保存主弹窗: {exc}")) + + confirm_button = cover_dialog.locator( + 'div.weui-desktop-dialog__ft button.weui-desktop-btn_primary:has-text("确认")' + ).first + await confirm_button.wait_for(state="visible", timeout=10000) + await confirm_button.click() + tencent_logger.success(_msg("🥳", "封面已经设置完成")) + + async def prepare_video_for_publish(self, page: Page) -> None: + await self.fill_title_and_tags(page) + await self.fill_description(page) + await self.apply_collection(page) + await self.apply_original_statement(page) + + async def upload(self, playwright: Playwright) -> None: + tencent_logger.info(_msg("🧍", "小人先检查 cookie、视频文件和发布时间")) + await self.validate_upload_args() + tencent_logger.info(_msg("🥳", "上传前检查通过")) + + browser = await playwright.chromium.launch(**_build_launch_kwargs(headless=self.headless)) + context = await browser.new_context(storage_state=self.account_file) + + try: + page = await context.new_page() + await self.open_upload_page(page) + tencent_logger.info(_msg("🏃", f"小人开始搬运视频: {self.title}")) + + await self.upload_video_file(page, self.file_path) + await self.prepare_video_for_publish(page) + await self.wait_for_upload_complete(page) + await self.set_thumbnail(page) + + if self.publish_strategy == TENCENT_PUBLISH_STRATEGY_SCHEDULED and self.publish_date != 0: + await self.set_schedule_time_tencent(page, self.publish_date) + + await self.set_short_title(page, self.title, self.short_title) + await self.submit_publish(page) + + await context.storage_state(path=self.account_file) + tencent_logger.success(_msg("🥳", "cookie 更新完毕")) + finally: + await context.close() + await browser.close() + + async def tencent_upload_video(self): async with async_playwright() as playwright: await self.upload(playwright) + + async def main(self): + await self.tencent_upload_video() + + +class TencentNote(TencentBaseUploader): + def __init__( + self, + image_paths, + note, + tags, + publish_date: datetime | int, + account_file, + title: str | None = None, + publish_strategy: str = TENCENT_PUBLISH_STRATEGY_IMMEDIATE, + debug: bool = DEBUG_MODE, + headless: bool = LOCAL_CHROME_HEADLESS, + is_draft: bool = False, + ): + super().__init__( + publish_date=publish_date, + account_file=account_file, + publish_strategy=publish_strategy, + debug=debug, + headless=headless, + ) + self.image_paths = image_paths + self.note = note or "" + self.title = title or (self.note[:30] if self.note else "") + self.tags = tags or [] + self.is_draft = is_draft + + async def validate_upload_args(self): + await self.validate_base_args() + if not self.title or not str(self.title).strip(): + raise ValueError("图文模式下，title 是必须的") + if not self.image_paths: + raise ValueError("图文模式下，图片是必须的") + + if isinstance(self.image_paths, (str, Path)): + self.image_paths = [self.image_paths] + + normalized_image_paths = [] + for image_path in self.image_paths: + normalized_image_paths.append(str(self.validate_image_file(image_path))) + self.image_paths = normalized_image_paths + + async def switch_to_note_mode(self, page: Page) -> None: + raise NotImplementedError("请在 TencentNote.switch_to_note_mode 中补充视频号切换到图文发布模式的逻辑") + + async def upload_note_images(self, page: Page) -> None: + raise NotImplementedError("请在 TencentNote.upload_note_images 中补充视频号图文图片上传逻辑") + + async def fill_note_title_and_tags(self, page: Page) -> None: + raise NotImplementedError("请在 TencentNote.fill_note_title_and_tags 中补充视频号图文标题/话题填写逻辑") + + async def fill_note_body(self, page: Page) -> None: + return None + + async def prepare_note_for_publish(self, page: Page) -> None: + await self.fill_note_title_and_tags(page) + await self.fill_note_body(page) + await self.apply_collection(page) + await self.apply_original_statement(page) + + async def upload_note_content(self, page: Page) -> None: + await self.switch_to_note_mode(page) + await self.upload_note_images(page) + await self.prepare_note_for_publish(page) + + async def upload(self, playwright: Playwright) -> None: + tencent_logger.info(_msg("🧍", "小人先检查 cookie、图文图片和发布时间")) + await self.validate_upload_args() + tencent_logger.info(_msg("🥳", "图文上传前检查通过")) + + browser = await playwright.chromium.launch(**_build_launch_kwargs(headless=self.headless)) + context = await browser.new_context(storage_state=self.account_file) + context = await set_init_script(context) + + try: + page = await context.new_page() + await self.open_upload_page(page) + tencent_logger.info(_msg("🏃", f"小人开始搬运图文，共 {len(self.image_paths)} 张图片")) + + await self.upload_note_content(page) + + if self.publish_strategy == TENCENT_PUBLISH_STRATEGY_SCHEDULED and self.publish_date != 0: + await self.set_schedule_time_tencent(page, self.publish_date) + + await self.submit_publish(page) + + await context.storage_state(path=self.account_file) + tencent_logger.success(_msg("🥳", "cookie 更新完毕")) + finally: + await context.close() + await browser.close() + + async def tencent_upload_note(self): + async with async_playwright() as playwright: + await self.upload(playwright) + + async def main(self): + await self.tencent_upload_note()