Files
publish-video-shipinhao/scripts/publish.py
2026-07-09 13:55:56 +08:00

567 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/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())