#!/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())