commit b3b1593efa9233f8b8d31600d8b79e5fb924310e Author: zhijun song <13861419@qq.com> Date: Thu Jul 9 13:55:56 2026 +0800 首次提交代码 diff --git a/SKILL.md b/SKILL.md new file mode 100644 index 0000000..3d47cdd --- /dev/null +++ b/SKILL.md @@ -0,0 +1,115 @@ +--- +name: publish-video-shipinhao +description: Use when Codex needs to install, configure, use, or troubleshoot OpenClaw, WorkBuddy, or similar automation products for uploading, drafting, scheduling, publishing, or verifying WeChat Channels videos through 微信视频号/视频号助手 browser automation. +--- + +# Publish Video Shipinhao + +## Overview + +Automate 微信视频号 video publishing through the 视频号助手 web console with Playwright or a compatible product adapter. Treat the web UI as semi-automated: login, CAPTCHA, account selection, product setup, and final publish confirmation may require the human operator. + +## Safety Rules + +- Verify product installation and usage steps from official docs or user-provided package docs before installing OpenClaw, WorkBuddy, or related products. +- Do not bypass QR login, CAPTCHA, rate limits, account checks, or platform policy prompts. +- Do not click the final publish button unless the user explicitly asks for live publishing or passes `--publish`. +- Prefer `--validate-config` first, then an upload/fill run without `--publish`, then final publish after reviewing the screenshot. +- Save screenshots and logs for every run so the user can verify what was submitted. + +## Quick Start + +1. Read `references/config.example.json` when creating a new publish config. +2. If the user mentions OpenClaw, WorkBuddy, or another automation product, read `references/product-adapters.md` before installing or running it. +3. Install Python dependencies when the target environment has not already done so: + +```bash +python -m pip install -r requirements.txt +python -m playwright install chromium +``` + +Skill installation does not automatically install Python packages unless the host product explicitly runs `requirements.txt`; verify this for OpenClaw, WorkBuddy, or any other adapter. +4. Ask the user for missing required fields: `video_path`, `title`, target product/runtime, and whether this should be draft-only, scheduled, or live publishing. +5. Run product environment checks first when a non-default product adapter is requested. +6. Run validation before opening the browser: + +```bash +python scripts/publish.py --config path/to/config.json --validate-config +``` + +7. For first-time login or expired sessions, run headed and let the user scan the QR code: + +```bash +python scripts/publish.py --config path/to/config.json --login-only --headed +``` + +8. Upload and fill metadata without final publish: + +```bash +python scripts/publish.py --config path/to/config.json --headed +``` + +9. Publish only after explicit approval: + +```bash +python scripts/publish.py --config path/to/config.json --headed --publish +``` + +Use `--save-draft` instead of `--publish` when the user wants a draft saved in 视频号助手. + +## Workflow + +1. Confirm intent: install/setup only, draft, scheduled publish, or live publish. +2. Select the adapter: built-in Playwright script, OpenClaw, WorkBuddy, or a generic browser-automation product. +3. Validate installation docs, local files, and metadata. Reject missing files, empty short title, short titles outside the 6-16 character range, or publish requests without explicit confirmation. +4. Open 视频号助手 with persisted browser/session state. Pause for manual QR login when needed. +5. Navigate to the create-post page, upload the video, optionally upload the cover, and fill title/description/topics. +6. If scheduling is requested, attempt the visible schedule controls; if selectors fail, stop for manual review instead of guessing. +7. Capture a screenshot before any final action. +8. Click `存草稿` or `发表/发布` only when the requested mode requires it. +9. Verify success by checking the toast, URL, list page, or a post-management entry. If verification is ambiguous, report the screenshot and manual check needed. + +## Product Adapters + +Use `references/product-adapters.md` for OpenClaw, WorkBuddy, and related products. Keep product-specific setup outside the main workflow: + +- Treat each product as an adapter with `install`, `configure`, `verify`, `run`, and `collect evidence` steps. +- Prefer official docs, bundled README files, or user-provided install packages over memory. +- Keep adapter-specific commands in the user's project/config, not hard-coded into this skill. +- Require a smoke test before using any installed product to publish a real video. +- Fall back to `scripts/publish.py` when the product cannot expose a controllable browser session, selector map, or evidence capture. + +## Bundled Script + +Use `scripts/publish.py` for the browser workflow. It supports: + +- `--validate-config`: validate JSON and required fields without launching a browser. +- `--skip-file-checks`: useful for config-shape tests when media files are not present. +- `--login-only`: open 视频号助手 and save the QR-login session. +- `--headed`: show the browser, recommended for login and final publishing. +- `--publish`: click the final publish button after filling the form. +- `--save-draft`: click the draft button after filling the form. +- `--review-seconds N`: keep the browser open before final action or close. + +When selectors fail, prefer adding page-specific selectors in the config under `selectors` or the product adapter profile instead of hard-coding a new guess in the skill instructions. + +## Dependencies + +The Python dependency manifest is `requirements.txt`. It currently installs Playwright only. Browser binaries are not installed by `pip`; run `python -m playwright install chromium` after installing Python packages. + +## Configuration + +Load `references/config.example.json` when preparing a new config. Required fields are: + +- `video_path`: absolute or working-directory-relative path to the video file. +- `title`: 视频号 short title, 6-16 characters. + +Common optional fields are `description`, `topics`, `cover_path`, `publish_at`, `visibility`, `browser.storage_state`, `product_adapter`, and `selectors`. + +## Common Mistakes + +- Running headless for first login. Use `--headed` so the user can scan the QR code. +- Installing OpenClaw, WorkBuddy, or a related product from memory. Check official or package docs first. +- Treating upload completion as publish completion. Always wait for a visible success signal or manual confirmation. +- Guessing a changed selector and clicking the wrong button. Stop, screenshot, and inspect the page when controls are ambiguous. +- Forgetting that platform rules and account permissions can block publishing even when upload automation succeeds. diff --git a/agents/openai.yaml b/agents/openai.yaml new file mode 100644 index 0000000..44ba196 --- /dev/null +++ b/agents/openai.yaml @@ -0,0 +1,4 @@ +interface: + display_name: "WeChat Channels Product Publish" + short_description: "Install and use product adapters for WeChat Channels publishing." + default_prompt: "Use $publish-video-shipinhao to install or use an OpenClaw, WorkBuddy, or Playwright adapter for WeChat Channels video publishing." diff --git a/references/config.example.json b/references/config.example.json new file mode 100644 index 0000000..96d92ed --- /dev/null +++ b/references/config.example.json @@ -0,0 +1,40 @@ +{ + "video_path": "C:/path/to/video.mp4", + "title": "视频标题示例", + "description": "视频正文或简介。可以包含换行。", + "topics": ["视频号", "自动发布"], + "cover_path": "C:/path/to/cover.jpg", + "publish_at": null, + "visibility": "public", + "product_adapter": { + "name": "playwright", + "product": "built-in", + "docs_url": "", + "install_commands": [ + "python -m pip install -r requirements.txt", + "python -m playwright install chromium" + ], + "verify_commands": ["python -c \"import playwright; print('playwright ok')\""], + "run_command": "python scripts/publish.py --config path/to/config.json --headed", + "evidence": ["before-final-action screenshot", "run log"], + "notes": "Use name=openclaw or name=workbuddy when the user wants a specific product adapter." + }, + "browser": { + "storage_state": ".auth/wechat_channels_state.json", + "headless": false, + "slow_mo_ms": 100, + "timeout_ms": 60000, + "review_seconds": 0 + }, + "selectors": { + "create_url": "https://channels.weixin.qq.com/platform/post/create", + "video_input": "input[type='file']", + "cover_input": "", + "title": "", + "description": "", + "save_draft_button": "", + "publish_button": "", + "schedule_button": "", + "schedule_input": "" + } +} diff --git a/references/product-adapters.md b/references/product-adapters.md new file mode 100644 index 0000000..cb28fa3 --- /dev/null +++ b/references/product-adapters.md @@ -0,0 +1,129 @@ +# Product Adapters + +Use this reference when the user wants OpenClaw, WorkBuddy, or a related product to install, configure, or run the WeChat Channels publishing workflow. + +## Adapter Contract + +Every product adapter must define: + +| Field | Purpose | +| --- | --- | +| `name` | Stable adapter id such as `openclaw`, `workbuddy`, or `playwright`. | +| `product` | Human product name and version when known. | +| `docs_url` | Official docs, bundled README, or user-provided package path used as the source of truth. | +| `install_commands` | Commands copied from current docs or package instructions. | +| `verify_commands` | Non-destructive checks proving the product is installed and usable. | +| `run_command` | Command or UI workflow that starts the video publishing task. | +| `evidence` | Expected screenshots, logs, output files, or status pages after a run. | + +If a field is unknown, ask for docs/package context or browse official sources before inventing commands. + +## Installation Flow + +1. Identify the product, edition, version, operating system, and whether the user wants local, cloud, desktop, CLI, or plugin usage. +2. Read current official docs or the package's README/install guide. Prefer product-provided instructions over memory. +3. Check prerequisites without changing the system: Python, Node.js, browser runtime, package manager, account login, license/token, and workspace permissions. +4. Install only after the user agrees to commands that download packages, modify global state, or open GUI apps. +5. Run `verify_commands` and capture the important output. +6. Run a smoke test that does not publish: open 视频号助手, validate config, or fill a draft without final submit. +7. Record product-specific commands in the user's config or notes so future runs do not rely on chat memory. + +## OpenClaw Adapter + +Use `name: "openclaw"` when the user explicitly requests OpenClaw. + +Expected shape: + +```json +{ + "product_adapter": { + "name": "openclaw", + "product": "OpenClaw", + "docs_url": "https://official-docs-or-package-readme", + "install_commands": [], + "verify_commands": [], + "run_command": "", + "evidence": ["install log", "smoke-test screenshot", "run log"] + } +} +``` + +Map OpenClaw to this workflow by confirming: + +- How OpenClaw installs skills, plugins, browser automation actions, or task bundles. +- Where it stores workspace files, browser profiles, cookies, and screenshots. +- Whether it can call Python scripts directly, import Playwright actions, or needs a wrapper command. +- How to pass `config.example.json` fields into an OpenClaw task. +- How to stop before final publish and expose a manual review checkpoint. + +If OpenClaw cannot provide browser-session persistence, file upload, screenshots, and manual review pauses, use it only for orchestration and keep `scripts/publish.py` as the publishing executor. + +## WorkBuddy Adapter + +Use `name: "workbuddy"` when the user explicitly requests WorkBuddy. + +Expected shape: + +```json +{ + "product_adapter": { + "name": "workbuddy", + "product": "WorkBuddy", + "docs_url": "https://official-docs-or-package-readme", + "install_commands": [], + "verify_commands": [], + "run_command": "", + "evidence": ["install log", "task result", "before-final-action screenshot"] + } +} +``` + +Map WorkBuddy to this workflow by confirming: + +- Whether WorkBuddy is a desktop app, browser extension, CLI, cloud workflow, or agent runtime. +- How it imports a skill, script, prompt, or reusable task. +- How it grants file access to the video, cover, config, and output screenshot directory. +- How it exposes manual QR login and final publish review. +- Which command or UI step proves the installation worked. + +When WorkBuddy is primarily a user-facing app, write product setup steps as operator instructions and keep destructive or account-changing actions behind explicit confirmation. + +## Generic Browser-Automation Product + +Use this profile for other products: + +```json +{ + "product_adapter": { + "name": "generic", + "product": "Product name and version", + "docs_url": "Official docs or package path", + "install_commands": [], + "verify_commands": [], + "run_command": "", + "evidence": [] + } +} +``` + +Minimum capabilities for direct publishing: + +- Open a persistent browser session. +- Upload local files into a web form. +- Fill text fields and click visible controls. +- Pause for QR login, CAPTCHA, account selection, and final review. +- Save screenshots and logs. +- Return a success/failure signal that can be verified. + +If a product lacks any minimum capability, use it for setup/orchestration only and run the bundled Playwright script for the actual browser task. + +## Verification Checklist + +Before saying a product adapter is ready: + +- Product source of truth is recorded: official URL, package README, or local docs path. +- Install command or manual installation steps are recorded. +- Non-destructive verification command has passed. +- Video publish config validates. +- A dry run produced a screenshot before final action. +- Final publish remains opt-in with `--publish` or an explicit user instruction. diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..472d6be --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +playwright>=1.45 diff --git a/scripts/publish.py b/scripts/publish.py new file mode 100644 index 0000000..1e7692f --- /dev/null +++ b/scripts/publish.py @@ -0,0 +1,566 @@ +#!/usr/bin/env python3 +"""Upload, draft, or publish a WeChat Channels video with Playwright. + +The script intentionally keeps final publishing behind an explicit --publish +flag because the WeChat Channels UI can require manual review. +""" + +from __future__ import annotations + +import argparse +import json +import re +import sys +import time +from datetime import datetime +from pathlib import Path +from typing import Any, Iterable + + +PLATFORM_URL = "https://channels.weixin.qq.com/platform" +DEFAULT_CREATE_URL = "https://channels.weixin.qq.com/platform/post/create" +VIDEO_EXTENSIONS = {".mp4", ".mov", ".m4v", ".avi", ".wmv", ".flv", ".mkv", ".webm"} +IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".webp"} + + +class ConfigError(ValueError): + """Raised when the publish configuration is invalid.""" + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Publish video to WeChat Channels via browser automation.") + parser.add_argument("--config", required=False, help="Path to publish config JSON.") + parser.add_argument("--validate-config", action="store_true", help="Validate config and exit.") + parser.add_argument("--skip-file-checks", action="store_true", help="Validate JSON shape without checking media files.") + parser.add_argument("--print-example", action="store_true", help="Print an example config and exit.") + parser.add_argument("--login-only", action="store_true", help="Open platform, wait for login, save storage state, and exit.") + parser.add_argument("--headed", action="store_true", help="Run with a visible browser window.") + parser.add_argument("--publish", action="store_true", help="Click the final publish button.") + parser.add_argument("--save-draft", action="store_true", help="Click the save-draft button.") + parser.add_argument("--review-seconds", type=int, default=None, help="Seconds to pause for manual review.") + parser.add_argument("--output-dir", default="runs", help="Directory for screenshots and run logs.") + return parser.parse_args() + + +def example_config() -> dict[str, Any]: + return { + "video_path": "C:/path/to/video.mp4", + "title": "视频标题示例", + "description": "视频正文或简介。可以包含换行。", + "topics": ["视频号", "自动发布"], + "cover_path": "C:/path/to/cover.jpg", + "publish_at": None, + "visibility": "public", + "product_adapter": { + "name": "playwright", + "product": "built-in", + "docs_url": "", + "install_commands": [ + "python -m pip install -r requirements.txt", + "python -m playwright install chromium", + ], + "verify_commands": ["python -c \"import playwright; print('playwright ok')\""], + "run_command": "python scripts/publish.py --config path/to/config.json --headed", + "evidence": ["before-final-action screenshot", "run log"], + "notes": "Use name=openclaw or name=workbuddy when the user wants a specific product adapter.", + }, + "browser": { + "storage_state": ".auth/wechat_channels_state.json", + "headless": False, + "slow_mo_ms": 100, + "timeout_ms": 60000, + "review_seconds": 0, + }, + "selectors": { + "create_url": DEFAULT_CREATE_URL, + "video_input": "input[type='file']", + "cover_input": "", + "title": "", + "description": "", + "save_draft_button": "", + "publish_button": "", + "schedule_button": "", + "schedule_input": "", + }, + } + + +def load_config(config_path: str | None) -> tuple[dict[str, Any], Path]: + if not config_path: + raise ConfigError("--config is required unless --print-example is used.") + path = Path(config_path).expanduser() + if not path.exists(): + raise ConfigError(f"Config file not found: {path}") + try: + data = json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError as exc: + raise ConfigError(f"Invalid JSON in {path}: {exc}") from exc + if not isinstance(data, dict): + raise ConfigError("Config root must be a JSON object.") + return data, path.resolve() + + +def resolve_path(value: str | None, base_dir: Path) -> Path | None: + if not value: + return None + path = Path(value).expanduser() + if not path.is_absolute(): + path = base_dir / path + return path.resolve() + + +def validate_media_file(path: Path, allowed_exts: set[str], label: str) -> None: + if not path.exists(): + raise ConfigError(f"{label} file not found: {path}") + if not path.is_file(): + raise ConfigError(f"{label} path is not a file: {path}") + if path.suffix.lower() not in allowed_exts: + allowed = ", ".join(sorted(allowed_exts)) + raise ConfigError(f"{label} file extension must be one of {allowed}: {path}") + if path.stat().st_size <= 0: + raise ConfigError(f"{label} file is empty: {path}") + + +def normalize_topics(value: Any) -> list[str]: + if value is None: + return [] + if isinstance(value, str): + value = [item.strip() for item in re.split(r"[,，\s]+", value) if item.strip()] + if not isinstance(value, list): + raise ConfigError("topics must be a list or comma-separated string.") + topics: list[str] = [] + for item in value: + if not isinstance(item, str): + raise ConfigError("topics must contain only strings.") + cleaned = item.strip().lstrip("#") + if cleaned: + topics.append(cleaned) + return topics + + +def normalize_string_list(value: Any, field_name: str) -> list[str]: + if value is None: + return [] + if not isinstance(value, list): + raise ConfigError(f"{field_name} must be a list of strings.") + cleaned: list[str] = [] + for item in value: + if not isinstance(item, str): + raise ConfigError(f"{field_name} must contain only strings.") + if item.strip(): + cleaned.append(item.strip()) + return cleaned + + +def normalize_product_adapter(value: Any) -> dict[str, Any]: + if value is None: + value = {"name": "playwright", "product": "built-in"} + if not isinstance(value, dict): + raise ConfigError("product_adapter must be an object.") + + name = str(value.get("name") or "playwright").strip().lower() + if not re.fullmatch(r"[a-z0-9][a-z0-9_-]{0,63}", name): + raise ConfigError("product_adapter.name must use letters, digits, underscore, or hyphen.") + + return { + "name": name, + "product": str(value.get("product") or "").strip(), + "docs_url": str(value.get("docs_url") or "").strip(), + "install_commands": normalize_string_list(value.get("install_commands"), "product_adapter.install_commands"), + "verify_commands": normalize_string_list(value.get("verify_commands"), "product_adapter.verify_commands"), + "run_command": str(value.get("run_command") or "").strip(), + "notes": str(value.get("notes") or "").strip(), + "evidence": normalize_string_list(value.get("evidence"), "product_adapter.evidence"), + } + + +def validate_config(data: dict[str, Any], base_dir: Path, skip_files: bool = False) -> dict[str, Any]: + title = str(data.get("title") or "").strip() + if not title: + raise ConfigError("title is required.") + if len(title) < 6: + raise ConfigError("title is under 6 characters; length must match the 视频号 short-title field.") + if len(title) > 16: + raise ConfigError("title is over 16 characters; shorten it for the 视频号 short-title field.") + + video_path = resolve_path(data.get("video_path"), base_dir) + if video_path is None: + raise ConfigError("video_path is required.") + cover_path = resolve_path(data.get("cover_path"), base_dir) + + if not skip_files: + validate_media_file(video_path, VIDEO_EXTENSIONS, "video") + if cover_path: + validate_media_file(cover_path, IMAGE_EXTENSIONS, "cover") + + visibility = str(data.get("visibility") or "public").strip().lower() + if visibility not in {"public", "private", "friends", "default"}: + raise ConfigError("visibility must be public, private, friends, or default.") + + publish_at = data.get("publish_at") + if publish_at not in (None, "") and not isinstance(publish_at, str): + raise ConfigError("publish_at must be null or a string accepted by the WeChat UI.") + + browser = data.get("browser") or {} + if not isinstance(browser, dict): + raise ConfigError("browser must be an object.") + selectors = data.get("selectors") or {} + if not isinstance(selectors, dict): + raise ConfigError("selectors must be an object.") + product_adapter = normalize_product_adapter(data.get("product_adapter")) + + return { + "video_path": video_path, + "cover_path": cover_path, + "title": title, + "description": str(data.get("description") or "").strip(), + "topics": normalize_topics(data.get("topics")), + "visibility": visibility, + "publish_at": str(publish_at).strip() if publish_at else "", + "product_adapter": product_adapter, + "browser": browser, + "selectors": selectors, + } + + +def import_playwright(): + try: + from playwright.sync_api import TimeoutError as PlaywrightTimeoutError + from playwright.sync_api import sync_playwright + except ImportError as exc: + raise RuntimeError( + "Playwright is not installed. Run: python -m pip install playwright && python -m playwright install chromium" + ) from exc + return sync_playwright, PlaywrightTimeoutError + + +def run_dir(output_dir: str) -> Path: + stamp = datetime.now().strftime("%Y%m%d-%H%M%S") + path = Path(output_dir).expanduser() / f"shipinhao-{stamp}" + path.mkdir(parents=True, exist_ok=True) + return path.resolve() + + +def screenshot(page: Any, directory: Path, name: str) -> Path: + path = directory / f"{name}.png" + page.screenshot(path=str(path), full_page=True) + print(f"[screenshot] {path}") + return path + + +def first_nonempty(*values: str | None) -> str: + for value in values: + if value: + return value + return "" + + +def locator_exists(locator: Any) -> bool: + try: + return locator.count() > 0 + except Exception: + return False + + +def fill_first(page: Any, selectors: Iterable[str], value: str, label: str, timeout_ms: int) -> bool: + if not value: + return True + for selector in selectors: + if not selector: + continue + locator = page.locator(selector).first + if not locator_exists(locator): + continue + try: + locator.wait_for(state="visible", timeout=timeout_ms) + locator.fill(value, timeout=timeout_ms) + print(f"[fill] {label}: {selector}") + return True + except Exception as exc: + print(f"[warn] Could not fill {label} with {selector}: {exc}") + print(f"[warn] Could not find a field for {label}; manual fill may be required.") + return False + + +def click_first(page: Any, selectors: Iterable[str], label: str, timeout_ms: int) -> bool: + for selector in selectors: + if not selector: + continue + try: + if selector.startswith("text="): + locator = page.get_by_text(selector.removeprefix("text="), exact=True).first + else: + locator = page.locator(selector).first + if not locator_exists(locator): + continue + locator.wait_for(state="visible", timeout=timeout_ms) + if hasattr(locator, "is_enabled") and not locator.is_enabled(timeout=timeout_ms): + print(f"[warn] {label} control is disabled: {selector}") + continue + locator.click(timeout=timeout_ms) + print(f"[click] {label}: {selector}") + return True + except Exception as exc: + print(f"[warn] Could not click {label} with {selector}: {exc}") + return False + + +def set_file_input(page: Any, configured_selector: str, file_path: Path, kind: str, timeout_ms: int) -> bool: + selectors = [configured_selector] if configured_selector else [] + selectors.append("input[type='file']") + wanted = "image" if kind == "cover" else "video" + + for selector in selectors: + if not selector: + continue + inputs = page.locator(selector) + try: + count = inputs.count() + except Exception as exc: + print(f"[warn] Could not inspect file inputs with {selector}: {exc}") + continue + for index in range(count): + candidate = inputs.nth(index) + try: + accept = (candidate.get_attribute("accept", timeout=timeout_ms) or "").lower() + except Exception: + accept = "" + if wanted == "video" and accept and "video" not in accept and ".mp4" not in accept: + continue + if wanted == "image" and accept and "image" not in accept and ".jpg" not in accept and ".png" not in accept: + continue + try: + candidate.set_input_files(str(file_path), timeout=timeout_ms) + print(f"[upload] {kind}: {file_path}") + return True + except Exception as exc: + print(f"[warn] Could not set {kind} file on {selector}[{index}]: {exc}") + return False + + +def composed_description(description: str, topics: list[str]) -> str: + tags = " ".join(f"#{topic}" for topic in topics) + if description and tags: + return f"{description}\n\n{tags}" + return description or tags + + +def raise_if_validation_errors(page: Any, directory: Path, label: str) -> None: + patterns = [ + "标题至少", + "标题超过", + "标题过长", + "请上传", + "请填写", + "不能为空", + ] + for pattern in patterns: + try: + locator = page.get_by_text(re.compile(pattern)).first + if locator.count() > 0 and locator.is_visible(timeout=1000): + screenshot(page, directory, f"{label}-validation-error") + raise RuntimeError(f"Page validation error after {label}: {pattern}") + except RuntimeError: + raise + except Exception: + continue + + +def wait_for_login(page: Any, timeout_ms: int) -> None: + print("[login] If a QR code appears, scan it in WeChat and wait for the console to load.") + try: + page.wait_for_load_state("networkidle", timeout=timeout_ms) + except Exception: + pass + if "login" in page.url.lower(): + print("[login] Waiting for login to finish...") + page.wait_for_url(re.compile(r"channels\.weixin\.qq\.com/platform"), timeout=max(timeout_ms, 120000)) + + +def apply_visibility(page: Any, visibility: str, timeout_ms: int) -> None: + labels = { + "public": ["text=公开", "text=所有人可见"], + "private": ["text=仅自己可见", "text=私密"], + "friends": ["text=朋友可见", "text=仅朋友可见"], + "default": [], + } + for selector in labels.get(visibility, []): + if click_first(page, [selector], f"visibility {visibility}", timeout_ms): + return + if visibility != "default": + print(f"[warn] Could not set visibility={visibility}; verify manually.") + + +def apply_schedule(page: Any, selectors: dict[str, Any], publish_at: str, timeout_ms: int) -> None: + if not publish_at: + return + schedule_button = first_nonempty(str(selectors.get("schedule_button") or ""), "text=定时发布", "text=定时") + clicked = click_first(page, [schedule_button, "text=定时发表"], "schedule", timeout_ms) + if not clicked: + print("[warn] Could not open schedule controls; set publish time manually.") + return + schedule_input = first_nonempty( + str(selectors.get("schedule_input") or ""), + "input[placeholder*='时间']", + "input[placeholder*='日期']", + "input[type='datetime-local']", + ) + filled = fill_first(page, [schedule_input], publish_at, "publish_at", timeout_ms) + if not filled: + print("[warn] Could not fill schedule time; set it manually before publishing.") + + +def automation(config: dict[str, Any], config_path: Path, args: argparse.Namespace) -> int: + sync_playwright, PlaywrightTimeoutError = import_playwright() + del PlaywrightTimeoutError + + browser_config = config["browser"] + selectors = config["selectors"] + timeout_ms = int(browser_config.get("timeout_ms") or 60000) + slow_mo = int(browser_config.get("slow_mo_ms") or 0) + headless = bool(browser_config.get("headless", False)) and not args.headed + review_seconds = args.review_seconds + if review_seconds is None: + review_seconds = int(browser_config.get("review_seconds") or 0) + + state_path = resolve_path(browser_config.get("storage_state") or ".auth/wechat_channels_state.json", config_path.parent) + assert state_path is not None + state_path.parent.mkdir(parents=True, exist_ok=True) + run_path = run_dir(args.output_dir) + + with sync_playwright() as playwright: + browser = playwright.chromium.launch(headless=headless, slow_mo=slow_mo) + context_args: dict[str, Any] = { + "locale": "zh-CN", + "viewport": {"width": 1440, "height": 1000}, + } + if state_path.exists(): + context_args["storage_state"] = str(state_path) + context = browser.new_context(**context_args) + context.set_default_timeout(timeout_ms) + page = context.new_page() + + try: + page.goto(PLATFORM_URL, wait_until="domcontentloaded", timeout=timeout_ms) + wait_for_login(page, timeout_ms) + context.storage_state(path=str(state_path)) + print(f"[state] Saved login state to {state_path}") + screenshot(page, run_path, "after-login") + + if args.login_only: + return 0 + + create_url = str(selectors.get("create_url") or DEFAULT_CREATE_URL) + page.goto(create_url, wait_until="domcontentloaded", timeout=timeout_ms) + try: + page.wait_for_load_state("networkidle", timeout=timeout_ms) + except Exception: + pass + screenshot(page, run_path, "create-page") + + if not set_file_input(page, str(selectors.get("video_input") or ""), config["video_path"], "video", timeout_ms): + screenshot(page, run_path, "video-input-not-found") + raise RuntimeError("Could not find a video upload input. Add selectors.video_input in the config.") + + print("[upload] Waiting for upload controls to settle...") + time.sleep(3) + + if config["cover_path"]: + if not set_file_input(page, str(selectors.get("cover_input") or ""), config["cover_path"], "cover", timeout_ms): + print("[warn] Could not upload cover automatically; set it manually if required.") + + title_selectors = [ + str(selectors.get("title") or ""), + "textarea[placeholder*='标题']", + "input[placeholder*='标题']", + "[contenteditable='true'][placeholder*='标题']", + "[contenteditable='true'][aria-label*='标题']", + ] + fill_first(page, title_selectors, config["title"], "title", timeout_ms) + + description_selectors = [ + str(selectors.get("description") or ""), + "textarea[placeholder*='描述']", + "textarea[placeholder*='简介']", + "textarea[placeholder*='正文']", + "textarea[placeholder*='内容']", + "[placeholder*='描述']", + "[data-placeholder*='描述']", + "[aria-placeholder*='描述']", + "[contenteditable='true'][placeholder*='描述']", + "[contenteditable='true'][aria-label*='描述']", + "xpath=//*[normalize-space()='视频描述']/following::*[@contenteditable='true' or self::textarea or self::input][1]", + ] + fill_first(page, description_selectors, composed_description(config["description"], config["topics"]), "description", timeout_ms) + + apply_visibility(page, config["visibility"], timeout_ms) + apply_schedule(page, selectors, config["publish_at"], timeout_ms) + + before_action = screenshot(page, run_path, "before-final-action") + print(f"[review] Review screenshot before final action: {before_action}") + if review_seconds > 0: + print(f"[review] Waiting {review_seconds} seconds for manual review...") + time.sleep(review_seconds) + + if args.publish and args.save_draft: + raise RuntimeError("Use only one of --publish or --save-draft.") + if args.publish: + publish_selector = first_nonempty(str(selectors.get("publish_button") or ""), "text=发表", "text=发布") + if not click_first(page, [publish_selector, "button:has-text('发表')", "button:has-text('发布')"], "publish", timeout_ms): + screenshot(page, run_path, "publish-button-not-found") + raise RuntimeError("Publish button not found. Verify manually or add selectors.publish_button.") + time.sleep(5) + screenshot(page, run_path, "after-publish-click") + raise_if_validation_errors(page, run_path, "publish") + print("[done] Publish click completed. Verify status in 视频号助手.") + elif args.save_draft: + draft_selector = first_nonempty(str(selectors.get("save_draft_button") or ""), "text=存草稿", "text=保存草稿") + if not click_first(page, [draft_selector, "button:has-text('草稿')"], "save draft", timeout_ms): + screenshot(page, run_path, "draft-button-not-found") + raise RuntimeError("Draft button not found. Verify manually or add selectors.save_draft_button.") + time.sleep(3) + screenshot(page, run_path, "after-save-draft") + raise_if_validation_errors(page, run_path, "save-draft") + print("[done] Draft click completed. Verify status in 视频号助手.") + else: + print("[done] Filled upload form. No final publish action was taken.") + + context.storage_state(path=str(state_path)) + return 0 + finally: + context.close() + browser.close() + + +def main() -> int: + args = parse_args() + if args.print_example: + print(json.dumps(example_config(), ensure_ascii=False, indent=2)) + return 0 + + try: + raw_config, config_path = load_config(args.config) + config = validate_config(raw_config, config_path.parent, skip_files=args.skip_file_checks) + if args.validate_config: + summary = { + "video_path": str(config["video_path"]), + "cover_path": str(config["cover_path"]) if config["cover_path"] else None, + "title": config["title"], + "topics": config["topics"], + "visibility": config["visibility"], + "publish_at": config["publish_at"] or None, + "product_adapter": config["product_adapter"]["name"], + } + print(json.dumps({"ok": True, "summary": summary}, ensure_ascii=False, indent=2)) + return 0 + return automation(config, config_path, args) + except ConfigError as exc: + print(f"[config-error] {exc}", file=sys.stderr) + return 2 + except RuntimeError as exc: + print(f"[runtime-error] {exc}", file=sys.stderr) + return 3 + + +if __name__ == "__main__": + raise SystemExit(main())