WordPress内容同步到微信公众号脚本

wordpress写的东西想要自动同步到微信公众号去，我又不想安装插件使用其他的平台。自己手写了一个python脚本自动进行同步。该脚本可将最近20篇的文章自动同步到微信公众号的草稿箱，然后通过微信公众号的草稿箱进行发布即可。本文详细介绍了脚本及其相关步骤，按照本文的教程进行操作就可以完成发布，不用在wordpress里面写了再去微信公众号里面再写一遍了。

一、开启网站REST API接口。

二、网站建立一个同步的账号。

三、微信公众号平台开发者平台申请API。

四、编辑python脚本信息。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
WordPress -> 微信公众号 自动同步工具
=====================================
自动从 WordPress 获取新文章，处理 HTML 内容和图片，
通过微信公众号 API 创建草稿并可选自动发布。

特性：
- 自动获取 WordPress 新文章（REST API 轮询）
- 文章内图片自动下载并上传到微信素材库，替换 URL
- 封面图自动上传为永久素材
- HTML 格式清理，保留微信支持的标签和内联样式
- 支持守护进程模式（定时轮询）和单次运行模式
- 状态文件记录已同步文章，避免重复

用法：
python wp2wechat.py --config config.json # 单次同步
python wp2wechat.py --config config.json --daemon # 守护进程模式
python wp2wechat.py --config config.json --post-id 123 # 同步指定文章
python wp2wechat.py --config config.json --test # 测试连接

依赖：pip install -r requirements.txt
"""

import argparse
import json
import logging
import os
import re
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urljoin, urlparse

import requests
from bs4 import BeautifulSoup, Comment

# 脚本所在目录（用于解析默认相对路径）
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))

# ============================================================================
# 配置管理
# ============================================================================

class Config:
"""加载并管理配置。"""

DEFAULTS = {
# WordPress 设置
"wordpress_url": "", # WordPress 站点地址，如 https://blog.example.com
"wp_username": "", # WordPress 用户名（公开站点可留空）
"wp_app_password": "", # WordPress 应用密码（公开站点可留空）

# 微信公众号设置
"wechat_appid": "", # 微信公众号 AppID
"wechat_secret": "", # 微信公众号 AppSecret

# 同步设置
"auto_publish": False, # 是否自动发布（False=仅创建草稿）
"poll_interval": 300, # 轮询间隔（秒），daemon 模式下使用
"state_file": "state.json", # 状态文件路径
"image_cache_dir": ".img_cache",# 图片缓存目录
"sync_drafts": False, # 是否同步草稿状态的文章（默认只同步已发布）

# 文章设置
"author_name": "", # 默认作者名（留空则使用 WordPress 作者名）
"need_open_comment": 0, # 是否开启评论（0/1）
"only_fans_can_comment": 0, # 是否仅粉丝可评论（0/1）
"default_cover_url": "", # 默认封面图 URL（文章无特色图时使用）

# 日志
"log_file": "wp2wechat.log", # 日志文件路径
"log_level": "INFO", # 日志级别
}

def __init__(self, config_path=None):
self.data = dict(self.DEFAULTS)
self._config_dir = SCRIPT_DIR # 默认相对于脚本目录
if config_path and os.path.exists(config_path):
self._config_dir = os.path.dirname(os.path.abspath(config_path))
with open(config_path, "r", encoding="utf-8") as f:
user_config = json.load(f)
self.data.update(user_config)

# 环境变量覆盖（前缀 WP2WECHAT_）
for key in self.data:
env_key = f"WP2WECHAT_{key.upper()}"
if env_key in os.environ:
val = os.environ[env_key]
# 尝试转换布尔值和整数
if isinstance(self.data[key], bool):
val = val.lower() in ("true", "1", "yes")
elif isinstance(self.data[key], int):
val = int(val)
self.data[key] = val

# 将相对路径解析为基于配置文件目录的绝对路径
for path_key in ("state_file", "image_cache_dir", "log_file"):
val = self.data.get(path_key, "")
if val and not os.path.isabs(val):
self.data[path_key] = os.path.join(self._config_dir, val)

def __getattr__(self, name):
if name.startswith("_"):
raise AttributeError(name)
if name in self.data:
return self.data[name]
raise AttributeError(f"'{type(self).__name__}' 没有属性 '{name}'")

def validate(self):
"""校验必填配置。"""
errors = []
if not self.wordpress_url:
errors.append("wordpress_url（WordPress 站点地址）必填")
if not self.wechat_appid:
errors.append("wechat_appid（微信 AppID）必填")
if not self.wechat_secret:
errors.append("wechat_secret（微信 AppSecret）必填")
if errors:
raise ValueError("配置错误:\n " + "\n ".join(errors))
return True

# ============================================================================
# 日志设置
# ============================================================================

def setup_logging(config):
"""配置日志输出（同时输出到文件和控制台）。"""
level = getattr(logging, config.log_level.upper(), logging.INFO)
fmt = "%(asctime)s [%(levelname)s] %(message)s"
datefmt = "%Y-%m-%d %H:%M:%S"

handlers = []
if config.log_file:
handlers.append(logging.FileHandler(config.log_file, encoding="utf-8"))
handlers.append(logging.StreamHandler(sys.stdout))

logging.basicConfig(
level=level,
format=fmt,
datefmt=datefmt,
handlers=handlers,
)

# ============================================================================
# 微信公众号 API 客户端
# ============================================================================

class WeChatClient:
"""微信公众号 API 封装。"""

BASE_URL = "https://api.weixin.qq.com/cgi-bin"

def __init__(self, appid, secret, cache_dir=".img_cache"):
self.appid = appid
self.secret = secret
self._token = None
self._token_expires_at = 0
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(parents=True, exist_ok=True)
self.session = requests.Session()

def _get_token(self):
"""获取 access_token，带缓存（提前 5 分钟刷新）。"""
now = time.time()
if self._token and now < self._token_expires_at - 300:
return self._token

url = f"{self.BASE_URL}/token"
params = {
"grant_type": "client_credential",
"appid": self.appid,
"secret": self.secret,
}
resp = self.session.get(url, params=params, timeout=30)
data = resp.json()

if "access_token" not in data:
err = data.get("errmsg", "未知错误")
raise RuntimeError(f"获取 access_token 失败: [{data.get('errcode')}] {err}")

self._token = data["access_token"]
self._token_expires_at = now + data.get("expires_in", 7200)
logging.info("微信 access_token 获取成功，有效期 %s 秒", data.get("expires_in", 7200))
return self._token

def upload_content_image(self, image_data, filename, mime_type):
"""
上传文章内图片（用于正文中引用）。
返回微信图片 URL。
接口: media/uploadimg
"""
token = self._get_token()
url = f"{self.BASE_URL}/media/uploadimg"
params = {"access_token": token}
files = {"media": (filename, image_data, mime_type)}

resp = self.session.post(url, params=params, files=files, timeout=60)
data = resp.json()

if "url" not in data:
err = data.get("errmsg", "未知错误")
raise RuntimeError(f"上传内容图片失败: [{data.get('errcode')}] {err}")

logging.debug("内容图片上传成功: %s", data["url"])
return data["url"]

def upload_permanent_image(self, image_data, filename, mime_type):
"""
上传永久图片素材（用于文章封面 thumb）。
返回 {"media_id": "...", "url": "..."}。
接口: material/add_material
"""
token = self._get_token()
url = f"{self.BASE_URL}/material/add_material"
params = {"access_token": token, "type": "image"}
files = {"media": (filename, image_data, mime_type)}

resp = self.session.post(url, params=params, files=files, timeout=60)
data = resp.json()

if "media_id" not in data:
err = data.get("errmsg", "未知错误")
raise RuntimeError(f"上传永久图片素材失败: [{data.get('errcode')}] {err}")

logging.debug("永久素材上传成功: media_id=%s", data["media_id"])
return {"media_id": data["media_id"], "url": data.get("url", "")}

def add_draft(self, article):
"""
新建草稿。
article: dict，包含 title, content, thumb_media_id, digest, author 等。
返回草稿 media_id。
接口: draft/add
"""
token = self._get_token()
url = f"{self.BASE_URL}/draft/add"
params = {"access_token": token}

body = {"articles": [article]}
# 必须用 ensure_ascii=False，否则中文会被转义为 \uXXXX，微信不解析
resp = self.session.post(
url, params=params,
data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
headers={"Content-Type": "application/json; charset=utf-8"},
timeout=60,
)
data = resp.json()

if "media_id" not in data:
err = data.get("errmsg", "未知错误")
raise RuntimeError(f"创建草稿失败: [{data.get('errcode')}] {err}")

logging.info("草稿创建成功: media_id=%s", data["media_id"])
return data["media_id"]

def publish(self, media_id):
"""
发布草稿。
接口: freepublish/submit
"""
token = self._get_token()
url = f"{self.BASE_URL}/freepublish/submit"
params = {"access_token": token}

resp = self.session.post(
url, params=params, json={"media_id": media_id}, timeout=60
)
data = resp.json()

if data.get("errcode", 0) != 0:
err = data.get("errmsg", "未知错误")
raise RuntimeError(f"发布失败: [{data.get('errcode')}] {err}")

logging.info("文章发布成功: publish_id=%s", data.get("publish_id", ""))
return data

def get_material_count(self):
"""获取素材总数（用于测试连接）。"""
token = self._get_token()
url = f"{self.BASE_URL}/material/get_materialcount"
params = {"access_token": token}
resp = self.session.get(url, params=params, timeout=30)
return resp.json()

def get_published_articles(self, count=20):
"""
获取已发布文章列表（标题）。
接口: freepublish/batchget
返回标题集合。
"""
token = self._get_token()
url = f"{self.BASE_URL}/freepublish/batchget"
params = {"access_token": token}

titles = set()
offset = 0
while True:
body = {"offset": offset, "count": min(count, 20), "no_content": 1}
resp = self.session.post(
url, params=params,
data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
headers={"Content-Type": "application/json; charset=utf-8"},
timeout=30,
)
data = resp.json()

if data.get("errcode", 0) != 0:
logging.warning("获取已发布文章失败: [%s] %s",
data.get("errcode"), data.get("errmsg"))
break

article_list = data.get("article_list", {}).get("article", [])
if not article_list:
break

for article in article_list:
content = article.get("content", {})
for item in content.get("news_item", []):
title = item.get("title", "").strip()
if title:
titles.add(title)

total = data.get("article_list", {}).get("total_count", 0)
offset += len(article_list)
if offset >= total or len(article_list) < min(count, 20):
break

logging.info("微信已发布文章: %d 篇", len(titles))
return titles

def get_draft_articles(self, count=20):
"""
获取草稿箱文章列表（标题）。
接口: draft/batchget
返回标题集合。
"""
token = self._get_token()
url = f"{self.BASE_URL}/draft/batchget"
params = {"access_token": token}

titles = set()
offset = 0
while True:
body = {"offset": offset, "count": min(count, 20), "no_content": 1}
resp = self.session.post(
url, params=params,
data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
headers={"Content-Type": "application/json; charset=utf-8"},
timeout=30,
)
data = resp.json()

if data.get("errcode", 0) != 0:
logging.warning("获取草稿列表失败: [%s] %s",
data.get("errcode"), data.get("errmsg"))
break

items = data.get("item", [])
if not items:
break

for item in items:
content = item.get("content", {})
for news in content.get("news_item", []):
title = news.get("title", "").strip()
if title:
titles.add(title)

total = data.get("total_count", 0)
offset += len(items)
if offset >= total or len(items) < min(count, 20):
break

logging.info("微信草稿箱文章: %d 篇", len(titles))
return titles

def get_existing_titles(self):
"""获取微信上已有的所有文章标题（已发布 + 草稿）。"""
published = self.get_published_articles()
drafts = self.get_draft_articles()
return published | drafts

# ============================================================================
# WordPress API 客户端
# ============================================================================

class WordPressClient:
"""WordPress REST API 客户端。"""

def __init__(self, base_url, username="", app_password=""):
self.base_url = base_url.rstrip("/")
self.api_base = f"{self.base_url}/wp-json/wp/v2"
self.session = requests.Session()

if username and app_password:
self.session.auth = (username, app_password)

def get_recent_posts(self, per_page=20, after=None):
"""
获取最新文章列表。
after: ISO 8601 日期字符串，只返回此日期之后的文章。
"""
params = {
"per_page": per_page,
"orderby": "date",
"order": "desc",
"_embed": "", # 包含嵌入数据（作者、特色图等）
"status": "publish",
}
if after:
params["after"] = after

resp = self.session.get(
f"{self.api_base}/posts", params=params, timeout=30
)
resp.raise_for_status()
return resp.json()

def get_post(self, post_id):
"""获取单篇文章详情。"""
resp = self.session.get(
f"{self.api_base}/posts/{post_id}",
params={"_embed": ""},
timeout=30,
)
resp.raise_for_status()
return resp.json()

def get_featured_image_url(self, post):
"""
从 _embedded 数据中提取特色图 URL。
"""
embedded = post.get("_embedded", {})

# 方式1: wp:featuredmedia
featured = embedded.get("wp:featuredmedia", [])
if featured:
# 获取大尺寸图
media_details = featured[0].get("media_details", {})
sizes = media_details.get("sizes", {})
# 优先 full > large > medium_large > medium
for size_name in ("full", "large", "medium_large", "medium"):
if size_name in sizes:
return sizes[size_name]["source_url"]
# 回退到原始 URL
return featured[0].get("source_url", "")

# 方式2: 从内容中找第一张图
content_html = post.get("content", {}).get("rendered", "")
soup = BeautifulSoup(content_html, "html.parser")
img = soup.find("img")
if img:
src = img.get("src") or img.get("data-src", "")
if src and not src.startswith("http"):
src = urljoin(self.base_url, src)
return src

return ""

def get_author_name(self, post):
"""从 _embedded 数据中获取作者名。"""
embedded = post.get("_embedded", {})
authors = embedded.get("author", [])
if authors:
return authors[0].get("name", "")
return ""

def download_image(self, url):
"""
下载图片，返回 (binary_data, mime_type, filename)。
"""
resp = self.session.get(url, timeout=60)
resp.raise_for_status()

mime_type = resp.headers.get("Content-Type", "image/jpeg")
# 清理 mime_type（可能包含 charset）
mime_type = mime_type.split(";")[0].strip()

# 从 URL 提取文件名
parsed = urlparse(url)
path = parsed.path
filename = os.path.basename(path) or "image.jpg"
# URL 解码文件名
filename = requests.utils.unquote(filename)

# 如果没有扩展名，根据 mime_type 补充
if "." not in filename:
ext_map = {
"image/jpeg": ".jpg",
"image/png": ".png",
"image/gif": ".gif",
"image/bmp": ".bmp",
}
filename += ext_map.get(mime_type, ".jpg")

return resp.content, mime_type, filename

# ============================================================================
# HTML 内容处理器
# ============================================================================

class ContentProcessor:
"""
处理 WordPress HTML 内容，使其适配微信公众号。
- 下载并重新上传所有图片
- 清理不支持的 HTML 标签和属性
- 保留内联样式
"""

# 微信支持的 HTML 标签
SUPPORTED_TAGS = {
"p", "div", "section", "span", "br", "hr",
"h1", "h2", "h3", "h4", "h5", "h6",
"strong", "b", "em", "i", "u", "s", "sub", "sup",
"a", "img", "blockquote", "pre", "code",
"ul", "ol", "li",
"table", "thead", "tbody", "tr", "td", "th",
"font", "center", "del", "ins",
}

# 需要完全移除的标签（连同内容）
REMOVE_TAGS = {"script", "style", "iframe", "noscript", "form", "input", "button"}

# 图片 URL 缓存（避免同一张图上传多次）
def __init__(self, wp_client, wechat_client, cache_dir=".img_cache"):
self.wp = wp_client
self.wechat = wechat_client
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(parents=True, exist_ok=True)
self._image_url_cache = {} # 原始URL -> 微信URL

def _upload_image_with_retry(self, url, is_cover=False, max_retries=3):
"""下载并上传图片，带重试和缓存。"""
# 检查缓存
cache_key = f"{url}{'_cover' if is_cover else ''}"
if cache_key in self._image_url_cache:
return self._image_url_cache[cache_key]

last_error = None
for attempt in range(1, max_retries + 1):
try:
# 下载图片
image_data, mime_type, filename = self.wp.download_image(url)

# 检查图片大小（微信限制约 10MB）
if len(image_data) > 10 * 1024 * 1024:
logging.warning("图片过大 (%.1fMB)，跳过: %s", len(image_data) / 1024 / 1024, url)
return None

if is_cover:
result = self.wechat.upload_permanent_image(image_data, filename, mime_type)
self._image_url_cache[cache_key] = result
return result
else:
wechat_url = self.wechat.upload_content_image(image_data, filename, mime_type)
self._image_url_cache[cache_key] = wechat_url
return wechat_url

except Exception as e:
last_error = e
logging.warning("图片上传失败 (第%d次): %s - %s", attempt, url, e)
if attempt < max_retries:
time.sleep(2 * attempt)

logging.error("图片上传最终失败，跳过: %s - %s", url, last_error)
return None

def _process_images(self, soup, base_url):
"""处理 HTML 中所有图片：下载 -> 上传到微信 -> 替换 URL。"""
img_count = 0
fail_count = 0

for img in soup.find_all("img"):
# 获取图片 URL（兼容懒加载）
src = img.get("src") or img.get("data-src") or ""
if not src:
# 从 srcset 中取最大图
srcset = img.get("srcset", "")
if srcset:
# 取最后一个（通常是最大的）
urls = srcset.split(",")
if urls:
last = urls[-1].strip().split(" ")
src = last[0] if last else ""

if not src:
continue

# 处理相对 URL
if not src.startswith("http"):
src = urljoin(base_url, src)

# 尝试获取原图（去掉 WordPress 尺寸后缀）
original_src = self._get_full_size_url(src)

# 上传到微信
wechat_url = self._upload_image_with_retry(original_src)
if not wechat_url:
# 尝试原始 URL
if original_src != src:
wechat_url = self._upload_image_with_retry(src)

if wechat_url:
img["src"] = wechat_url
img_count += 1
else:
fail_count += 1
logging.warning("图片替换失败，将移除该图片: %s", src)
img.decompose()
continue

# 清理不需要的属性
for attr in list(img.attrs):
if attr not in ("src", "alt", "style", "width", "height"):
del img[attr]

if img_count or fail_count:
logging.info("图片处理完成: 成功 %d, 失败 %d", img_count, fail_count)

def _get_full_size_url(self, url):
"""
尝试将 WordPress 缩略图 URL 转换为原图 URL。
例如: image-300x200.jpg -> image.jpg
"""
# 匹配 -WIDTHxHEIGHT 后缀
pattern = r"-\d+x\d+(\.(?:jpg|jpeg|png|gif|bmp|webp))$"
match = re.search(pattern, url, re.IGNORECASE)
if match:
return url[:match.start()] + match.group(1)
return url

def _clean_html(self, soup):
"""清理 HTML，移除微信不支持的标签和属性。"""

# 移除注释
for comment in soup.find_all(string=lambda t: isinstance(t, Comment)):
comment.extract()

# 移除不需要的标签（连同内容）
for tag_name in self.REMOVE_TAGS:
for tag in soup.find_all(tag_name):
tag.decompose()

# 处理 <picture> 标签 -> 提取 <img>
for picture in soup.find_all("picture"):
img = picture.find("img")
if img:
img = img.extract()
picture.replace_with(img)
else:
picture.decompose()

# 处理 <figure> -> 转换为 <section>
for figure in soup.find_all("figure"):
figure.name = "section"
for figcaption in soup.find_all("figcaption"):
figcaption.name = "p"

# 处理 <video> / <audio> -> 转为链接
for media_tag in soup.find_all(["video", "audio"]):
source = media_tag.find("source")
src = source.get("src", "") if source else media_tag.get("src", "")
if src:
link = soup.new_tag("a", href=src)
link.string = f"[媒体文件: {src}]"
media_tag.replace_with(link)
else:
media_tag.decompose()

# 清理不支持标签的属性，或将不支持的标签转为 div
for tag in soup.find_all(True):
if tag.name not in self.SUPPORTED_TAGS:
# 尝试转为 div（保留内容）
tag.name = "div"

# 清理属性（保留 style, href, src, alt, colspan, rowspan 等）
allowed_attrs = {
"style", "href", "src", "alt", "title",
"colspan", "rowspan", "width", "height",
"align", "valign", "color", "face", "size",
"target", "border", "cellpadding", "cellspacing",
}
for attr in list(tag.attrs):
if attr not in allowed_attrs:
del tag[attr]

# 移除空链接的空属性
for a in soup.find_all("a"):
if not a.get("href"):
a.unwrap()

def process(self, html_content, base_url):
"""
处理文章 HTML 内容。
返回处理后的 HTML 字符串。
"""
if not html_content:
return ""

soup = BeautifulSoup(html_content, "html.parser")

# 先处理图片（需要下载上传）
self._process_images(soup, base_url)

# 清理 HTML
self._clean_html(soup)

# 获取处理后的 HTML
result = str(soup)

# 微信对 content 有大小限制（约 120KB）
if len(result.encode("utf-8")) > 120000:
logging.warning("文章内容较大 (%d KB)，可能超过微信限制", len(result.encode("utf-8")) // 1024)

return result

def process_cover_image(self, image_url, base_url):
"""
处理封面图：下载并上传为微信永久素材。
返回 {"media_id": "...", "url": "..."} 或 None。
"""
if not image_url:
return None

# 处理相对 URL
if not image_url.startswith("http"):
image_url = urljoin(base_url, image_url)

# 获取原图
original_url = self._get_full_size_url(image_url)

result = self._upload_image_with_retry(original_url, is_cover=True)
if not result and original_url != image_url:
result = self._upload_image_with_retry(image_url, is_cover=True)

return result

@staticmethod
def generate_digest(html_content, max_length=120):
"""从 HTML 内容生成摘要。"""
soup = BeautifulSoup(html_content, "html.parser")
text = soup.get_text(separator="", strip=True)
# 清理多余空白
text = re.sub(r"\s+", " ", text)
# 微信 digest 限制 120 字节，中文字占 3 字节，需按字节截断
return ContentProcessor._truncate_to_bytes(text, 120)

@staticmethod
def _truncate_to_bytes(text, max_bytes):
"""按 UTF-8 字节长度截断字符串。"""
encoded = text.encode("utf-8")
if len(encoded) <= max_bytes:
return text
# 截断后预留 "..." 的 3 字节
cut = max_bytes - 3
# 逐字符截断，避免截断到多字节字符中间
result = b""
for char in text:
char_bytes = char.encode("utf-8")
if len(result) + len(char_bytes) > cut:
break
result += char_bytes
return result.decode("utf-8") + "..."

# ============================================================================
# 状态管理
# ============================================================================

class StateManager:
"""管理同步状态，避免重复同步。"""

def __init__(self, state_file="state.json"):
self.state_file = Path(state_file)
self.state = self._load()

def _load(self):
"""加载状态文件。"""
if self.state_file.exists():
try:
with open(self.state_file, "r", encoding="utf-8") as f:
return json.load(f)
except (json.JSONDecodeError, IOError) as e:
logging.warning("状态文件读取失败，将重新创建: %s", e)
return {
"synced_posts": {}, # {post_id: {"date": "...", "wechat_media_id": "..."}}
"last_sync": None,
}

def save(self):
"""保存状态文件。"""
self.state["last_sync"] = datetime.now(timezone.utc).isoformat()
with open(self.state_file, "w", encoding="utf-8") as f:
json.dump(self.state, f, ensure_ascii=False, indent=2)

def is_synced(self, post_id):
"""检查文章是否已同步。"""
return str(post_id) in self.state["synced_posts"]

def mark_synced(self, post_id, media_id, publish_id=None, source="script"):
"""标记文章为已同步。"""
self.state["synced_posts"][str(post_id)] = {
"date": datetime.now(timezone.utc).isoformat(),
"wechat_media_id": media_id,
"publish_id": publish_id,
"source": source,
}
self.save()

def get_last_sync_date(self):
"""获取上次同步时间。"""
return self.state.get("last_sync")

# ============================================================================
# 同步引擎
# ============================================================================

class SyncEngine:
"""协调 WordPress -> 微信 的同步流程。"""

def __init__(self, config):
self.config = config
self.wp = WordPressClient(
config.wordpress_url,
config.wp_username,
config.wp_app_password,
)
self.wechat = WeChatClient(
config.wechat_appid,
config.wechat_secret,
config.image_cache_dir,
)
self.processor = ContentProcessor(
self.wp, self.wechat, config.image_cache_dir
)
self.state = StateManager(config.state_file)

def test_connection(self):
"""测试 WordPress 和微信连接。"""
print("\n=== 连接测试 ===\n")

# 测试 WordPress
print("[1/2] 测试 WordPress 连接...")
try:
posts = self.wp.get_recent_posts(per_page=1)
print(f" ✓ WordPress 连接成功，共获取到 {len(posts)} 篇文章")
if posts:
print(f" 最新文章: {posts[0]['title']['rendered']}")
except Exception as e:
print(f" ✗ WordPress 连接失败: {e}")
return False

# 测试微信
print("\n[2/2] 测试微信公众号连接...")
try:
count = self.wechat.get_material_count()
print(f" ✓ 微信公众号连接成功")
print(f" 素材统计: 图片 {count.get('image_count', '?')} 个, 视频 {count.get('video_count', '?')} 个")
except Exception as e:
print(f" ✗ 微信公众号连接失败: {e}")
return False

print("\n=== 测试完成 ===\n")
return True

def sync_post(self, post):
"""
同步单篇文章到微信公众号。
返回 (success, media_id)。
"""
post_id = post["id"]
title = post.get("title", {}).get("rendered", f"文章-{post_id}")
# 清理标题中的 HTML
title = BeautifulSoup(title, "html.parser").get_text(strip=True)
# 微信标题限制 64 字节
title = ContentProcessor._truncate_to_bytes(title, 64)

content_html = post.get("content", {}).get("rendered", "")
post_url = post.get("link", "")
excerpt = post.get("excerpt", {}).get("rendered", "")
digest = BeautifulSoup(excerpt, "html.parser").get_text(strip=True)
if not digest:
digest = ContentProcessor.generate_digest(content_html)

# 微信摘要限制 120 字节（generate_digest 已处理，这里兜底）
digest = ContentProcessor._truncate_to_bytes(digest, 120)

# 作者名
author = self.config.author_name or self.wp.get_author_name(post) or ""

logging.info("=" * 60)
logging.info("开始同步文章 [ID:%s]: %s", post_id, title)
logging.info("原文链接: %s", post_url)
logging.info("=" * 60)

try:
# 1. 处理封面图
logging.info("步骤 1/4: 处理封面图")
cover_url = self.wp.get_featured_image_url(post)
if not cover_url and self.config.default_cover_url:
cover_url = self.config.default_cover_url

thumb_media_id = None
if cover_url:
cover_result = self.processor.process_cover_image(
cover_url, self.config.wordpress_url
)
if cover_result:
thumb_media_id = cover_result["media_id"]
logging.info("封面图上传成功: %s", thumb_media_id)
else:
logging.warning("封面图上传失败")
else:
logging.warning("文章无特色图，且未设置默认封面图")

if not thumb_media_id:
logging.error("缺少封面图素材 ID，无法创建草稿（微信要求必须有封面图）")
return False, None

# 2. 处理正文内容和图片
logging.info("步骤 2/4: 处理正文内容和图片")
processed_content = self.processor.process(
content_html, self.config.wordpress_url
)

if not processed_content.strip():
logging.error("处理后的文章内容为空")
return False, None

# 3. 创建草稿
logging.info("步骤 3/4: 创建微信公众号草稿")
article = {
"title": title,
"author": author,
"digest": digest,
"content": processed_content,
"content_source_url": post_url,
"thumb_media_id": thumb_media_id,
"need_open_comment": self.config.need_open_comment,
"only_fans_can_comment": self.config.only_fans_can_comment,
}

media_id = self.wechat.add_draft(article)
logging.info("草稿创建成功: media_id=%s", media_id)

# 4. 发布（可选）
publish_id = None
if self.config.auto_publish:
logging.info("步骤 4/4: 自动发布")
try:
result = self.wechat.publish(media_id)
publish_id = result.get("publish_id", "")
except Exception as e:
logging.error("自动发布失败（草稿已保存）: %s", e)
else:
logging.info("步骤 4/4: 跳过发布（auto_publish=False），草稿已保存到微信公众号后台")

# 标记已同步
self.state.mark_synced(post_id, media_id, publish_id)

logging.info("文章同步完成: %s", title)
return True, media_id

except Exception as e:
logging.error("文章同步失败 [ID:%s]: %s", post_id, e, exc_info=True)
return False, None

def sync_new_posts(self, limit=20):
"""
同步所有未同步的新文章。
返回 (成功数, 失败数)。
"""
# 获取最新文章
logging.info("正在从 WordPress 获取最新文章...")
posts = self.wp.get_recent_posts(per_page=limit)

if not posts:
logging.info("WordPress 暂无文章")
return 0, 0

# 过滤已同步的文章
new_posts = [p for p in posts if not self.state.is_synced(p["id"])]
logging.info("共 %d 篇文章，其中 %d 篇待同步", len(posts), len(new_posts))

if not new_posts:
logging.info("没有新文章需要同步")
return 0, 0

# 获取微信上已有的文章标题（已发布 + 草稿），用于跳过手动发布的文章
logging.info("正在检查微信已发布/草稿文章，跳过已存在的...")
try:
existing_titles = self.wechat.get_existing_titles()
except Exception as e:
logging.warning("获取微信已有文章失败，将不进行标题去重: %s", e)
existing_titles = set()

# 进一步过滤：标题在微信上已存在的也跳过
final_posts = []
skipped_count = 0
for post in new_posts:
raw_title = post.get("title", {}).get("rendered", "")
title = BeautifulSoup(raw_title, "html.parser").get_text(strip=True)
if title and title in existing_titles:
logging.info("跳过（微信已存在）: %s [ID:%s]", title, post["id"])
self.state.mark_synced(post["id"], media_id=None, publish_id=None,
source="manual")
skipped_count += 1
else:
final_posts.append(post)

logging.info("跳过 %d 篇已存在文章，实际待同步 %d 篇",
skipped_count, len(final_posts))

if not final_posts:
logging.info("没有新文章需要同步")
return 0, 0

success_count = 0
fail_count = 0

# 按时间正序同步（旧文章先发）
final_posts.sort(key=lambda p: p.get("date", ""))

for i, post in enumerate(final_posts):
success, _ = self.sync_post(post)
if success:
success_count += 1
else:
fail_count += 1

# 文章之间稍作停顿，避免 API 频率限制
if i < len(final_posts) - 1:
time.sleep(3)

logging.info("同步完成: 成功 %d 篇, 跳过 %d 篇, 失败 %d 篇",
success_count, skipped_count, fail_count)
return success_count, fail_count

def sync_specific_post(self, post_id):
"""同步指定的单篇文章。"""
logging.info("正在获取文章 [ID:%s]...", post_id)
try:
post = self.wp.get_post(post_id)
except Exception as e:
logging.error("获取文章失败: %s", e)
return False

success, _ = self.sync_post(post)
return success

def run_daemon(self):
"""守护进程模式：定时轮询新文章。"""
interval = self.config.poll_interval
logging.info("启动守护进程模式，轮询间隔: %d 秒", interval)

while True:
try:
self.sync_new_posts()
except Exception as e:
logging.error("轮询同步异常: %s", e, exc_info=True)

logging.info("等待 %d 秒后再次检查...", interval)
time.sleep(interval)

# ============================================================================
# 命令行入口
# ============================================================================

def main():
parser = argparse.ArgumentParser(
description="WordPress -> 微信公众号自动同步工具",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
示例:
python wp2wechat.py --config config.json # 单次同步所有新文章
python wp2wechat.py --config config.json --daemon # 守护进程模式
python wp2wechat.py --config config.json --post-id 123 # 同步指定文章
python wp2wechat.py --config config.json --test # 测试连接
python wp2wechat.py --config config.json --list # 列出最近文章
python wp2wechat.py --config config.json --mark-synced 7497,7500 # 标记已手动发布的文章
python wp2wechat.py --config config.json --list-synced # 查看已同步列表
""",
)
parser.add_argument(
"--config", "-c", default=None,
help="配置文件路径 (默认: 脚本目录下的 config.json)"
)
parser.add_argument(
"--daemon", "-d", action="store_true",
help="守护进程模式，定时轮询新文章"
)
parser.add_argument(
"--post-id", "-p", type=int,
help="同步指定 ID 的文章"
)
parser.add_argument(
"--test", "-t", action="store_true",
help="测试 WordPress 和微信连接"
)
parser.add_argument(
"--list", "-l", action="store_true",
help="列出 WordPress 最近的文章"
)
parser.add_argument(
"--limit", type=int, default=20,
help="获取文章数量 (默认: 20)"
)
parser.add_argument(
"--mark-synced", "-m", metavar="IDS",
help="标记指定文章为已同步（逗号分隔的ID，如: 7497,7500,7510），"
"用于跳过已在微信公众号手动发布的文章"
)
parser.add_argument(
"--list-synced", action="store_true",
help="查看已同步文章列表"
)

args = parser.parse_args()

# 加载配置（默认使用脚本目录下的 config.json）
config_path = args.config or os.path.join(SCRIPT_DIR, "config.json")
config = Config(config_path)

try:
config.validate()
except ValueError as e:
print(f"\n配置错误:\n{e}\n")
print("请参考 config.example.json 创建配置文件。")
sys.exit(1)

setup_logging(config)
engine = SyncEngine(config)

# 执行命令
if args.test:
ok = engine.test_connection()
sys.exit(0 if ok else 1)

if args.list_synced:
synced = engine.state.state.get("synced_posts", {})
if not synced:
print("还没有已同步的文章记录")
else:
print(f"\n已同步文章 ({len(synced)} 篇):\n")
print(f"{'ID':<8} {'来源':<8} {'同步时间':<26} {'标题'}")
print("-" * 80)
# 获取文章标题
try:
posts = engine.wp.get_recent_posts(per_page=100)
title_map = {p["id"]: BeautifulSoup(
p.get("title", {}).get("rendered", ""), "html.parser"
).get_text(strip=True) for p in posts}
except Exception:
title_map = {}
for pid, info in sorted(synced.items(), key=lambda x: int(x[0])):
source = info.get("source", "script")
source_label = "手动" if source == "manual" else "脚本"
date = info.get("date", "")[:19]
title = title_map.get(int(pid), f"(文章ID:{pid})")
print(f"{pid:<8} {source_label:<8} {date:<26} {title}")
sys.exit(0)

if args.mark_synced:
ids = [s.strip() for s in args.mark_synced.split(",") if s.strip()]
# 获取文章标题用于日志
title_map = {}
try:
posts = engine.wp.get_recent_posts(per_page=100)
title_map = {p["id"]: BeautifulSoup(
p.get("title", {}).get("rendered", ""), "html.parser"
).get_text(strip=True) for p in posts}
except Exception:
pass
for id_str in ids:
try:
pid = int(id_str)
engine.state.mark_synced(pid, media_id=None, publish_id=None,
source="manual")
title = title_map.get(pid, "")
print(f" ✓ 标记 [ID:{pid}] 为已同步: {title}")
except ValueError:
print(f" ✗ 无效的 ID: {id_str}")
print(f"\n共标记 {len(ids)} 篇文章为已同步，下次同步时将自动跳过。")
sys.exit(0)

if args.list:
logging.info("正在获取 WordPress 最新文章...")
posts = engine.wp.get_recent_posts(per_page=args.limit)
if not posts:
print("暂无文章")
else:
print(f"\n{'ID':<8} {'日期':<20} {'标题'}")
print("-" * 70)
for p in posts:
pid = p["id"]
date = p.get("date", "")[:19]
title = BeautifulSoup(
p.get("title", {}).get("rendered", ""), "html.parser"
).get_text(strip=True)
synced = "✓" if engine.state.is_synced(pid) else " "
print(f"{synced} {pid:<6} {date:<20} {title}")
sys.exit(0)

if args.post_id:
ok = engine.sync_specific_post(args.post_id)
sys.exit(0 if ok else 1)

if args.daemon:
engine.run_daemon()
else:
# 单次同步
success, fail = engine.sync_new_posts(limit=args.limit)
print(f"\n同步完成: 成功 {success} 篇, 失败 {fail} 篇")
if fail > 0:
print("请查看日志文件了解失败原因。")
sys.exit(0 if fail == 0 else 1)

if __name__ == "__main__":
main()

五、将相关信息填入脚本中。

网站的地址、账号、密码、公众号的AppID、公众号的AppSecret填写完毕后运行。

六、微信公众号草稿箱查看同步情况。

七、微信公众号进行发布即可。