一、OpenClaw监控WordPress评论
1. 核心应用逻辑
监控的本质是定时查询WordPress评论接口，对比新旧数据，发现新增/待审核/违规评论并触发相应操作(如邮件提醒、自动审核、关键词过滤)。
2. 实现步骤与核心代码
前置准备
确保WordPress开启REST API(默认开启)
为 OpenClaw创建有「编辑评论」权限的API用户
安装依赖：pip install requests python-dotenv schedule
核心代码(监控 + 自动处理评论)
import requests
import schedule
import time
import json
from datetime import datetime
# 配置项
WP_API_URL = "https://你的WP域名/wp-json/wp/v2/comments"
WP_AUTH = ("你的API用户名", "你的API密码")
# 本地缓存（记录已监控过的评论ID，避免重复处理）
MONITORED_COMMENT_IDS = set()
# 违规关键词列表
FORBIDDEN_KEYWORDS = ["广告", "刷单", "链接", "微信"]
# 提醒接收邮箱（可选，需额外配置SMTP）
ALERT_EMAIL = "你的邮箱@example.com"
def get_latest_comments():
"""从WP API获取最新评论（默认返回10条，可调整per_page参数）"""
params = {
"per_page": 20, # 每次获取20条评论
"status": "hold", # 只监控待审核评论：hold(待审核)/approved(已通过)/spam(垃圾)
"orderby": "date",
"order": "desc" # 按时间倒序
}
try:
response = requests.get(WP_API_URL, auth=WP_AUTH, params=params)
response.raise_for_status()
return response.json()
except Exception as e:
print(f"获取评论失败：{e}")
return []
def check_new_comments():
"""检查并处理新评论"""
global MONITORED_COMMENT_IDS
comments = get_latest_comments()
if not comments:
return
for comment in comments:
comment_id = comment["id"]
# 跳过已监控过的评论
if comment_id in MONITORED_COMMENT_IDS:
continue
# 标记为已监控
MONITORED_COMMENT_IDS.add(comment_id)
# 提取评论核心信息
comment_content = comment["content"]["rendered"].lower()
comment_post_id = comment["post"]
comment_author = comment["author_name"]
# 1. 关键词检测（判断是否为违规评论）
is_spam = any(keyword in comment_content for keyword in FORBIDDEN_KEYWORDS)
# 2. 自动处理评论
if is_spam:
# 标记为垃圾评论
update_comment_status(comment_id, "spam")
print(f"[{datetime.now()}] 发现违规评论 ID:{comment_id}，已标记为垃圾")
else:
# 自动审核通过
update_comment_status(comment_id, "approved")
print(f"[{datetime.now()}] 新评论 ID:{comment_id} 已自动审核通过")
# 3. 发送提醒（可选，需自行实现send_email函数）
# send_email(ALERT_EMAIL, f"新评论提醒", f"评论ID:{comment_id} 作者:{comment_author} 内容:{comment_content}")
def update_comment_status(comment_id, status):
"""更新评论状态"""
url = f"{WP_API_URL}/{comment_id}"
data = {"status": status}
headers = {"Content-Type": "application/json"}
try:
response = requests.post(
url,
auth=WP_AUTH,
headers=headers,
data=json.dumps(data)
)
response.raise_for_status()
return True
except Exception as e:
print(f"更新评论{comment_id}状态失败：{e}")
return False
# 定时任务：每5分钟检查一次新评论
schedule.every(5).minutes.do(check_new_comments)
if __name__ == "__main__":
print(f"评论监控已启动，开始时间：{datetime.now()}")
# 先加载历史评论ID（避免启动时重复处理）
initial_comments = get_latest_comments()
for c in initial_comments:
MONITORED_COMMENT_IDS.add(c["id"])
# 持续运行定时任务
while True:
schedule.run_pending()
time.sleep(1)3. 关键优化点
持久化缓存：上述代码用set临时存储已监控ID，生产环境建议用json文件/SQLite持久化，避免程序重启后重复处理
频率控制：监控间隔建议≥5 分钟，避免频繁调用API增加服务器负载
多状态监控：可修改status参数为all，同时监控已通过 / 垃圾评论，实现全量评论审计
二、OpenClaw批量修改WordPress产品数据
1. 核心应用逻辑
WordPress产品数据通常基于WooCommerce插件，批量修改的核心是通过WooCommerce REST API 读取产品列表 → 按规则修改(价格/库存/标签)→ 批量提交更新。
2. 实现步骤与核心代码
前置准备
安装WooCommerce插件(确保启用REST API)
获取WooCommerce API 密钥：WooCommerce → 设置 → 高级 → REST API → 添加密钥(分配「读写」权限)
安装依赖：pip install requests python-dotenv
核心代码(批量修改产品价格 / 库存)
import requests
import json
from requests.auth import HTTPBasicAuth
# WooCommerce API 配置
WC_API_URL = "https://你的WP域名/wp-json/wc/v3/products"
# WooCommerce API 密钥（不是WP用户名密码）
WC_CONSUMER_KEY = "你的Consumer Key"
WC_CONSUMER_SECRET = "你的Consumer Secret"
WC_AUTH = HTTPBasicAuth(WC_CONSUMER_KEY, WC_CONSUMER_SECRET)
def get_products_by_category(category_id, per_page=50):
"""按分类获取产品列表（支持分页）"""
params = {
"category": category_id, # 产品分类ID
"per_page": per_page, # 每页获取数量（最大100）
"page": 1 # 页码
}
all_products = []
while True:
try:
response = requests.get(WC_API_URL, auth=WC_AUTH, params=params)
response.raise_for_status()
products = response.json()
if not products:
break
all_products.extend(products)
params["page"] += 1
except Exception as e:
print(f"获取产品失败：{e}")
break
return all_products
def batch_update_products(products, update_rule):
"""批量修改产品数据"""
success_count = 0
fail_count = 0
fail_ids = []
for product in products:
product_id = product["id"]
update_url = f"{WC_API_URL}/{product_id}"
headers = {"Content-Type": "application/json"}
# 构建修改数据（按规则更新）
update_data = {}
if "price" in update_rule:
# 示例：价格上涨10%
old_price = float(product["price"]) if product["price"] else 0
new_price = old_price * (1 + update_rule["price"]["rate"])
update_data["regular_price"] = str(round(new_price, 2))
if "stock" in update_rule:
# 示例：设置库存为固定值
update_data["stock_quantity"] = update_rule["stock"]["value"]
if "tags" in update_rule:
# 示例：添加新标签
old_tags = [tag["name"] for tag in product["tags"]]
new_tags = old_tags + update_rule["tags"]["add"]
update_data["tags"] = new_tags
# 提交修改
try:
response = requests.put(
update_url,
auth=WC_AUTH,
headers=headers,
data=json.dumps(update_data)
)
response.raise_for_status()
success_count += 1
print(f"产品ID:{product_id} 修改成功")
except Exception as e:
fail_count += 1
fail_ids.append(product_id)
print(f"产品ID:{product_id} 修改失败：{e}")
# 输出统计结果
print(f"\n批量修改完成：成功{success_count}个，失败{fail_count}个")
if fail_ids:
print(f"失败产品ID：{fail_ids}")
if __name__ == "__main__":
# 1. 定义修改规则
update_rule = {
"price": {"rate": 0.1}, # 价格上涨10%
"stock": {"value": 100}, # 库存设置为100
"tags": {"add": ["促销", "热销"]} # 添加标签
}
# 2. 获取需要修改的产品（分类ID=5的所有产品）
target_products = get_products_by_category(category_id=5)
print(f"共获取到 {len(target_products)} 个产品待修改")
# 3. 批量修改
batch_update_products(target_products, update_rule)3. 关键优化点
分页处理：WooCommerce API默认每页最多返回100个产品，需通过page参数分页获取，避免漏改
幂等性保障：修改前记录产品原始数据，修改失败时可回滚;建议先测试1-2个产品，再批量操作
字段适配：WooCommerce产品字段较多(如sale_price促销价、sku商品编码)，可根据需求扩展update_rule
速率限制：批量修改时添加time.sleep(0.5)，避免每秒请求数超过服务器限制(通常≤20次/秒)
监控WP评论：核心是「定时调用评论API + 对比历史数据 + 自动处理」，重点控制监控频率和缓存持久化，避免重复操作;
批量修改产品数据：基于WooCommerce REST API，先筛选目标产品，再按规则批量更新，重点关注分页、幂等性和速率限制;
通用注意事项：两类场景均需使用专用API账号(最小权限原则)，操作前备份数据，生产环境添加完整日志记录。
