diff --git a/sau_cli.py b/sau_cli.py index 36e945d..4d74d45 100644 --- a/sau_cli.py +++ b/sau_cli.py @@ -26,6 +26,13 @@ from uploader.ks_uploader.main import ( cookie_auth as kuaishou_cookie_auth, ks_setup, ) +from uploader.tencent_uploader.main import ( + TENCENT_PUBLISH_STRATEGY_IMMEDIATE, + TENCENT_PUBLISH_STRATEGY_SCHEDULED, + TencentVideo, + cookie_auth as tencent_cookie_auth, + tencent_setup, +) from uploader.xiaohongshu_uploader.main import ( XIAOHONGSHU_PUBLISH_STRATEGY_IMMEDIATE, XIAOHONGSHU_PUBLISH_STRATEGY_SCHEDULED, @@ -132,6 +139,23 @@ class BilibiliVideoUploadRequest: publish_date: datetime | int +@dataclass(slots=True) +class TencentVideoUploadRequest: + account_name: str + video_file: Path + title: str + description: str + tags: list[str] + publish_date: datetime | int + thumbnail_file: Path | None = None + short_title: str | None = None + category: str | None = None + is_draft: bool = False + publish_strategy: str = TENCENT_PUBLISH_STRATEGY_IMMEDIATE + debug: bool = True + headless: bool = True + + def has_interactive_terminal() -> bool: return sys.stdin.isatty() and sys.stdout.isatty() @@ -234,6 +258,18 @@ async def check_bilibili_account(account_name: str) -> bool: return result.returncode == 0 +async def login_tencent_account(account_name: str, headless: bool = True) -> dict: + account_file = resolve_account_file("tencent", account_name) + return await tencent_setup(str(account_file), handle=True, return_detail=True, headless=headless) + + +async def check_tencent_account(account_name: str) -> bool: + account_file = resolve_account_file("tencent", account_name) + if not account_file.exists(): + return False + return await tencent_cookie_auth(str(account_file)) + + async def upload_video(request: DouyinVideoUploadRequest) -> Path: account_file = resolve_account_file("douyin", request.account_name) is_ready = await douyin_setup(str(account_file), handle=False) @@ -408,6 +444,34 @@ async def upload_bilibili_video(request: BilibiliVideoUploadRequest) -> Path: return account_file +async def upload_tencent_video(request: TencentVideoUploadRequest) -> Path: + account_file = resolve_account_file("tencent", request.account_name) + is_ready = await tencent_setup(str(account_file), handle=False) + if not is_ready: + raise RuntimeError( + f"Tencent/WeChat Channels cookie is missing or expired: {account_file}. " + f"Run `sau tencent login --account {request.account_name}` first." + ) + + app = TencentVideo( + title=request.title, + file_path=str(request.video_file), + tags=request.tags, + publish_date=request.publish_date, + account_file=str(account_file), + category=request.category, + is_draft=request.is_draft, + desc=request.description, + thumbnail_path=str(request.thumbnail_file) if request.thumbnail_file else None, + short_title=request.short_title, + publish_strategy=request.publish_strategy, + debug=request.debug, + headless=request.headless, + ) + await app.tencent_upload_video() + return account_file + + def existing_file_path(value: str) -> Path: path = Path(value) if not path.is_file(): @@ -541,6 +605,28 @@ def build_parser() -> argparse.ArgumentParser: bilibili_upload_video_parser.add_argument("--tid", required=True, type=int, help="Bilibili category id") bilibili_upload_video_parser.add_argument("--tags", default="", help="Comma-separated tags, such as tag1,tag2") bilibili_upload_video_parser.add_argument("--schedule", type=schedule_value, help=f"Schedule time in {schedule_help}") + + tencent_parser = platform_parsers.add_parser("tencent", help="Tencent/WeChat Channels operations") + tencent_actions = tencent_parser.add_subparsers(dest="action", required=True) + + for action_name in ("login", "check"): + action_parser = tencent_actions.add_parser(action_name, help=f"Tencent/WeChat Channels {action_name}") + action_parser.add_argument("--account", required=True, help="Tencent user-defined account_name") + if action_name == "login": + add_runtime_flags(action_parser) + + tencent_upload_video_parser = tencent_actions.add_parser("upload-video", help="Upload one video to WeChat Channels") + tencent_upload_video_parser.add_argument("--account", required=True, help="Tencent user-defined account_name") + tencent_upload_video_parser.add_argument("--file", required=True, type=existing_file_path, help="Video file path") + tencent_upload_video_parser.add_argument("--title", required=True, help="Video title") + tencent_upload_video_parser.add_argument("--desc", default="", help="Optional video description") + tencent_upload_video_parser.add_argument("--tags", default="", help="Comma-separated tags, such as tag1,tag2") + tencent_upload_video_parser.add_argument("--schedule", type=schedule_value, help=f"Schedule time in {schedule_help}") + tencent_upload_video_parser.add_argument("--thumbnail", type=existing_file_path, help="Optional thumbnail path") + tencent_upload_video_parser.add_argument("--short-title", help="Optional WeChat Channels short title") + tencent_upload_video_parser.add_argument("--category", help="Optional original content category") + tencent_upload_video_parser.add_argument("--draft", action="store_true", help="Save as draft instead of publishing") + add_runtime_flags(tencent_upload_video_parser) return parser @@ -728,6 +814,43 @@ async def dispatch(args: argparse.Namespace) -> int: raise RuntimeError(f"Unsupported Bilibili action: {args.action}") + if args.platform == "tencent": + if args.action == "login": + result = await login_tencent_account(args.account, headless=args.headless) + if not result["success"]: + raise RuntimeError(result["message"]) + print(f"Tencent/WeChat Channels login flow completed: {result['account_file']}") + return 0 + + if args.action == "check": + is_valid = await check_tencent_account(args.account) + print("valid" if is_valid else "invalid") + return 0 if is_valid else 1 + + publish_strategy = TENCENT_PUBLISH_STRATEGY_SCHEDULED if args.schedule else TENCENT_PUBLISH_STRATEGY_IMMEDIATE + + if args.action == "upload-video": + request = TencentVideoUploadRequest( + account_name=args.account, + video_file=args.file, + title=args.title, + description=args.desc, + tags=parse_tags(args.tags), + publish_date=args.schedule or 0, + thumbnail_file=args.thumbnail, + short_title=args.short_title, + category=args.category, + is_draft=args.draft, + publish_strategy=publish_strategy, + debug=args.debug, + headless=args.headless, + ) + await upload_tencent_video(request) + print(f"Tencent/WeChat Channels video upload submitted: {request.video_file}") + return 0 + + raise RuntimeError(f"Unsupported Tencent/WeChat Channels action: {args.action}") + raise RuntimeError(f"Unsupported platform: {args.platform}") diff --git a/uploader/tencent_uploader/main.py b/uploader/tencent_uploader/main.py index 8a24140..1f35c03 100644 --- a/uploader/tencent_uploader/main.py +++ b/uploader/tencent_uploader/main.py @@ -537,8 +537,10 @@ class TencentBaseUploader(BaseVideoUploader): await collection_elements.first.click() async def apply_original_statement(self, page: Page) -> None: + original_set = False if await page.get_by_label("视频为原创").count(): await page.get_by_label("视频为原创").check() + original_set = True try: label_locator = await page.locator('label:has-text("我已阅读并同意 《视频号原创声明使用条款》")').is_visible() @@ -548,30 +550,89 @@ class TencentBaseUploader(BaseVideoUploader): if label_locator: await page.get_by_label("我已阅读并同意 《视频号原创声明使用条款》").check() await page.get_by_role("button", name="声明原创").click() + original_set = True - 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() + declaration_entry = page.locator( + 'div.label span:has-text("声明原创"), ' + 'div:has-text("声明原创"):has(input.ant-checkbox-input), ' + 'div:has-text("原创声明"):has(input.ant-checkbox-input)' + ).first + if await declaration_entry.count(): + original_checkbox = page.locator("div.declare-original-checkbox input.ant-checkbox-input").first + if await original_checkbox.count() and not await original_checkbox.is_disabled(): + await original_checkbox.click() + await page.wait_for_timeout(500) 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() + await page.locator("div.declare-original-dialog input.ant-checkbox-input:visible").first.click() original_type_form = page.locator('div.original-type-form > div.form-label:has-text("原创类型"):visible') if await original_type_form.count(): + category = getattr(self, "category", None) 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() + option = None + if category: + option = page.locator( + "ul.weui-desktop-dropdown__list " + f'li.weui-desktop-dropdown__list-ele:has-text("{category}")' + ).first + if not await option.count(): + option = None + if option is None: + option = page.locator( + "ul.weui-desktop-dropdown__list " + "li.weui-desktop-dropdown__list-ele:visible" + ).first + if await option.count(): + await option.click() await page.wait_for_timeout(1000) declare_button = page.locator('button:has-text("声明原创"):visible') if await declare_button.count(): - await declare_button.click() + await declare_button.first.click() + original_set = True + await page.wait_for_timeout(1000) + + if not original_set: + for original_text in ("声明原创", "原创声明", "视频为原创"): + try: + modern_original = page.locator(f'text="{original_text}"').first + if await modern_original.count() and await modern_original.is_visible(): + await modern_original.click() + original_set = True + await page.wait_for_timeout(1000) + break + except Exception: + continue + + content_declaration = page.locator('text="内容声明"').first + try: + if await content_declaration.count() and await content_declaration.is_visible(): + await content_declaration.click() + for option_text in ("无需声明", "不声明", "无"): + option = page.locator(f'text="{option_text}"').first + if await option.count() and await option.is_visible(): + await option.click() + tencent_logger.info(_msg("🧾", f"内容声明已选择: {option_text}")) + break + else: + tencent_logger.info(_msg("🧾", "当前页面未发现内容声明字段")) + except Exception as exc: + tencent_logger.warning(_msg("😵", f"内容声明设置失败，继续前先人工确认页面: {exc}")) + + if not original_set: + try: + diagnostic_path = Path(BASE_DIR) / "debug_tencent_original_missing.png" + await page.screenshot(path=str(diagnostic_path), full_page=True) + visible_text = (await page.locator("body").inner_text())[-4000:] + tencent_logger.warning(_msg("😵", f"未确认声明原创，诊断截图: {diagnostic_path}")) + tencent_logger.warning(_msg("🧾", f"页面末尾文本: {visible_text}")) + except Exception as exc: + tencent_logger.warning(_msg("😵", f"生成原创声明诊断信息失败: {exc}")) + raise RuntimeError("未能在视频号发布页找到并设置“声明原创”，已停止发布") async def wait_for_upload_complete(self, page: Page) -> None: while True: @@ -736,7 +797,6 @@ class TencentVideo(TencentBaseUploader): 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、视频文件和发布时间")) @@ -754,6 +814,7 @@ class TencentVideo(TencentBaseUploader): 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.apply_original_statement(page) await self.set_thumbnail(page) if self.publish_strategy == TENCENT_PUBLISH_STRATEGY_SCHEDULED and self.publish_date != 0: