微信朋友圈定时发布的工程实现：定向可见、发布窗口与失败处理

做私域运营的团队大概都见过这类排期表：早 9 点一条产品动态、下午 3 点一条客户见证、晚上 8 点一条活动预告，而且不同内容只给不同的客户组看——新客户看引流款，老客户看复购款，同行一律屏蔽。

人工执行这套排期有两个必然翻车的点：一是忘发、错发，二是“定向可见”靠人手点分组，选错一次就是事故。这件事本质上是个工程问题：发布是定时任务，可见性是配置项，失败处理是重试策略。

这篇文章按工程落地的顺序，把朋友圈自动发布从接口能力、定向设计、调度策略到完整代码一次讲完。代码可直接跑，文末附上线前的评审清单。

阶段一：能力盘点——接口能做什么、边界在哪

先说结论：文字朋友圈的发布是一个 POST 接口就能覆盖的事，且官方协议层原生支持定向可见，不需要任何“分组可见”的黑科技。

我用通道是 WTAPI——基于 iPad/Mac 协议的微信机器人接口服务，把消息、好友、群、朋友圈这些能力做成标准 HTTP API（下文简称“平台”）。说直白点：你的程序调用它给的网址，就能用这个微信号收消息、发消息、发朋友圈，扫码登录、掉线重连这些协议层的活它来干。朋友圈发布走的是它的 sns 模块（接口路径以文档站 weiti.apifox.cn 为准），参数全景如下：

两个必须提前知道的边界，都写在文档原文里：

阶段二：定向可见的分层设计

定向可见不要在发圈的代码里现想，先做数据分层：

第一步，给好友打标签。 标签是平台标签接口的能力（给好友打标、用户分层），把“新客户/老客户/同行/内部测试”做成标签体系，这一步是所有定向动作的地基。

第二步，发布时按标签定向。 工程上有两条路线：

• 白名单式（allowTagIds）：只让指定标签看。适合灰度验证和精准内容——比如一条复购活动只对“老客户”标签可见。
• 黑名单式（disableTagIds）：全网可见但排除指定标签。适合“同行屏蔽”这类排除型需求。

我的实践判断：优先白名单，黑名单兜底。白名单的出错后果是“少部分人没看到”，黑名单的出错后果是“不该看的人看到了”——前者可补救，后者是事故。另外同一条朋友圈不要同时混用 allow 和 disable 两组参数，两组的优先级语义建议先实测确认再上生产，别赌。

第三步，先拿小号验证定向语义。 朋友圈没有“文件传输助手”这种自测对象，替代做法是：allowWxIds 只填自己的小号 wxid，发一条只有你能看到的朋友圈——发布链路、定向语义一次验完，对外零暴露。

阶段三：发布窗口与失败处理

定时发布最忌讳的是“整点准点发”——每天 9:00:00 整一条，机器行为特征过于整齐。调度层做两件事：

失败处理用最朴素的策略：网络类错误指数退避重试（最多 2 次），业务类错误（参数/登录态问题）直接告警不重试——重试解决不了登录态失效，只会刷屏日志。

阶段四：完整代码

依赖只有 pip install requests。发布函数、调度函数、主流程如下，可直接跑：

# WTAPI 接口平台 | 文档: https://weiti.apifox.cn
# 本文依赖:
# – 发文字朋友圈: POST /finder/v2/api/sns/sendTextSns
# (必填 appId/content; 定向: allowWxIds/disableWxIds/atWxIds; 标签: allowTagIds/disableTagIds)
# – 边界: 新设备登录后 1-3 天朋友圈功能受限; 频次红线见文档「防封必看」模块
# 下一步: 文档侧栏「API模块 / 朋友圈接口」核对参数, 标签分层用「标签接口」
# 官网: www.chuapi.com

import random
import time as time_mod
from dataclasses import dataclass, field
from datetime import datetime, timedelta, date

import requests

SNS_TEXT_URL = "https://wx.chuapi.com/finder/v2/api/sns/sendTextSns"

DAILY_LIMIT = 3 # 单日发布上限, 硬编码防上游抽风
PUBLISH_WINDOWS = [ # 发布窗口, 单位=当日第几分钟
(9 * 60 + 30, 11 * 60),
(14 * 60, 17 * 60),
(20 * 60, 22 * 60),
]

@dataclass
class MomentTask:
"""一条待发布的朋友圈。定向字段全部默认空 = 全员可见。"""
content: str
allow_tag_ids: list[str] = field(default_factory=list) # 允许谁看(标签)
allow_wxids: list[str] = field(default_factory=list) # 允许谁看(wxid白名单)
disable_tag_ids: list[str] = field(default_factory=list) # 不给谁看(标签)
at_wxids: list[str] = field(default_factory=list) # 提醒谁看
privacy: bool = False # True=仅自己可见

def send_moment(token: str, app_id: str, task: MomentTask) –> dict:
"""发布一条文字朋友圈, 返回接口 data 段 {id, userName, nickName, createTime}。

失败判定基于返回结构: 成功时 data 内必含朋友圈 id (文档 schema),
不猜 ret 成功码, 避免不同环境返回不一致导致误判。
"""
payload = {
"appId": app_id,
"content": task.content,
"privacy": task.privacy,
"allowTagIds": task.allow_tag_ids,
"allowWxIds": task.allow_wxids,
"disableTagIds": task.disable_tag_ids,
"atWxIds": task.at_wxids,
}
resp = requests.post(
SNS_TEXT_URL,
headers={"Content-Type": "application/json", "X-finder-TOKEN": token},
json=payload,
timeout=15,
)
data = resp.json()
result = data.get("data", {})
if "id" not in result:
raise RuntimeError(f"发布失败: ret={data.get('ret')} msg={data.get('msg')}")
return result

def send_moment_with_retry(token: str, app_id: str, task: MomentTask,
max_retry: int = 2) –> dict:
"""网络类错误指数退避重试; 业务类错误(RuntimeError)直接抛出不重试。"""
for attempt in range(max_retry + 1):
try:
return send_moment(token, app_id, task)
except RuntimeError:
raise # 参数/登录态问题, 重试无意义
except requests.RequestException as e:
if attempt == max_retry:
raise
wait = 2 ** attempt * 5
print(f"[重试] 第{attempt + 1}次网络失败({e}), {wait}s 后重试")
time_mod.sleep(wait)
raise AssertionError("unreachable")

def schedule_for(day: date, tasks: list[MomentTask]) –> list[tuple[datetime, MomentTask]]:
"""把任务列表铺进当日发布窗口, 窗口内随机取时刻。"""
if len(tasks) > DAILY_LIMIT:
raise ValueError(f"任务数 {len(tasks)} 超过单日上限 {DAILY_LIMIT}")
plan = []
for task in tasks:
start, end = random.choice(PUBLISH_WINDOWS)
fire = datetime.combine(day, datetime.min.time()) + timedelta(minutes=random.randint(start, end))
plan.append((fire, task))
plan.sort(key=lambda x: x[0])
return plan

if __name__ == "__main__":
TOKEN = "YOUR_TOKEN" # 控制台获取
APP_ID = "YOUR_APPID" # 扫码登录后生成的设备ID

# 第一步: 发一条仅小号可见的探针朋友圈, 验证链路和定向语义
probe = MomentTask(content="接口联调测试", allow_wxids=["wxid_你的小号"])
print("探针发布:", send_moment(TOKEN, APP_ID, probe))

# 第二步: 验证通过后, 按标签定向铺排当日任务
# daily = schedule_for(date.today(), [
# MomentTask(content="上午内容", allow_tag_ids=["标签ID_老客户"]),
# MomentTask(content="晚间活动", allow_tag_ids=["标签ID_老客户"],
# at_wxids=["wxid_重点客户"]),
# ])
# for fire_time, task in daily:
# while datetime.now() < fire_time:
# time_mod.sleep(30)
# print("发布:", send_moment_with_retry(TOKEN, APP_ID, task))

第一段探针跑通后，你就拿到了三样东西：链路是通的、定向语义是实测过的、返回结构里能拿到朋友圈 id（后续做“发了没有”的对账就靠它）。

上线前评审清单

按顺序过一遍，全部打勾再放量：

下一步

最后照例说明：朋友圈自动化建议用于企业自有账号的内容运营，遵守微信平台规则与相关法律法规，控制发布频次。

–