256 lines
11 KiB
Python
256 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""把成稿同步到微信公众号草稿箱（只写草稿，绝不群发）。
|
||
|
||
前置条件：
|
||
1. 已认证的公众号（未认证账号调用草稿接口会返回 48001）。
|
||
2. 环境变量 WECHAT_APP_ID、WECHAT_APP_SECRET（在 mp.weixin.qq.com 设置与开发→基本配置获取）。
|
||
3. 调用方公网 IP 已加入公众号后台的 IP 白名单（家用宽带 IP 变化后需回后台更新）。
|
||
|
||
用法：
|
||
python3 publish_draft.py <文章目录> [--title 标题] [--digest 摘要] [--cover 封面图] \
|
||
[--check] [--dry-run] [--author 作者]
|
||
--check 只验证凭证与 IP 白名单（调 token 接口），不写任何内容
|
||
--dry-run 不联网，打印将上传的图片与将提交的草稿参数
|
||
退出码：0 成功；1 失败（错误码已翻译成人话）；2 凭证缺失或参数不合法。
|
||
"""
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import struct
|
||
import sys
|
||
import urllib.parse
|
||
import urllib.request
|
||
|
||
API = "https://api.weixin.qq.com"
|
||
ERR = {
|
||
40001: "AppSecret 错误或 access_token 已过期——核对 WECHAT_APP_SECRET，或稍后重试",
|
||
40013: "AppID 无效——核对 WECHAT_APP_ID",
|
||
40164: "本机公网 IP 不在白名单——把提示里的 IP 加到公众号后台「基本配置→IP白名单」后重试",
|
||
40125: "AppSecret 无效或已重置——到后台重置 Secret 后更新环境变量",
|
||
45009: "接口当日调用额度用完——明天再试",
|
||
48001: "该账号无此 API 权限——草稿箱接口需要已认证公众号；未认证个人订阅号用不了",
|
||
40113: "图片格式被素材接口拒绝（部分 PNG 编码会被挑剔）——转成 jpg 再试："
|
||
"sips -s format jpeg cover.png --out cover.jpg（macOS 自带），"
|
||
"然后用 --cover cover.jpg 重跑",
|
||
53401: "封面图校验失败——换一张（建议 900×383 以上的 2.35:1 图）",
|
||
53402: "封面 media_id 无效或不是图片素材",
|
||
}
|
||
|
||
|
||
def die(msg, code=1):
|
||
print(f"❌ {msg}", file=sys.stderr)
|
||
sys.exit(code)
|
||
|
||
|
||
def wx_err(r):
|
||
"""把微信返回体翻译成人话；成功返回 None。"""
|
||
code = r.get("errcode", 0)
|
||
if not code:
|
||
return None
|
||
hint = ERR.get(code, r.get("errmsg", "未知错误"))
|
||
return f"errcode {code}：{hint}（{r.get('errmsg', '')}）"
|
||
|
||
|
||
def http_json(url, data=None, timeout=30):
|
||
"""GET（data=None）或 JSON POST，返回解析后的 dict。"""
|
||
req = urllib.request.Request(url, method="POST" if data is not None else "GET")
|
||
body = None
|
||
if data is not None:
|
||
body = json.dumps(data, ensure_ascii=False).encode("utf-8")
|
||
req.add_header("Content-Type", "application/json; charset=utf-8")
|
||
with urllib.request.urlopen(req, body, timeout=timeout) as resp:
|
||
return json.loads(resp.read().decode("utf-8"))
|
||
|
||
|
||
def multipart(field, filename, content):
|
||
b = "----LeWeChatBoundary7MA4YWxk"
|
||
parts = [
|
||
f"--{b}\r\nContent-Disposition: form-data; name=\"{field}\"; "
|
||
f"filename=\"{filename}\"\r\nContent-Type: application/octet-stream\r\n\r\n"
|
||
]
|
||
body = parts[0].encode("utf-8") + content + f"\r\n--{b}--\r\n".encode("utf-8")
|
||
return body, b
|
||
|
||
|
||
def post_file(url, path):
|
||
with open(path, "rb") as f:
|
||
content = f.read()
|
||
body, boundary = multipart("media", os.path.basename(path), content)
|
||
req = urllib.request.Request(url, method="POST", data=body)
|
||
req.add_header("Content-Type", f"multipart/form-data; boundary={boundary}")
|
||
with urllib.request.urlopen(req, timeout=60) as resp:
|
||
return json.loads(resp.read().decode("utf-8"))
|
||
|
||
|
||
def img_size(path):
|
||
"""读图片像素尺寸：PNG（IHDR）/ JPEG（SOF 帧），标准库解析。"""
|
||
with open(path, "rb") as f:
|
||
head = f.read(32)
|
||
f.seek(0)
|
||
data = f.read()
|
||
if head.startswith(b"\x89PNG"):
|
||
w, h = struct.unpack(">II", data[16:24])
|
||
return w, h
|
||
if head.startswith(b"\xff\xd8"): # JPEG：找第一帧 SOF0/2
|
||
i = 2
|
||
while i < len(data) - 9:
|
||
if data[i] != 0xFF:
|
||
i += 1
|
||
continue
|
||
marker = data[i + 1]
|
||
if marker in (0xC0, 0xC2):
|
||
h, w = struct.unpack(">HH", data[i + 5:i + 9])
|
||
return w, h
|
||
if marker in (0xD8, 0x01) or 0xD0 <= marker <= 0xD7:
|
||
i += 2
|
||
continue
|
||
seg_len = struct.unpack(">H", data[i + 2:i + 4])[0]
|
||
i += 2 + seg_len
|
||
raise ValueError(f"无法识别图片尺寸（支持 PNG/JPEG）：{path}")
|
||
|
||
|
||
def crop_fractions(w, h, tw, th):
|
||
"""最大居中裁剪框 → 微信 pic_crop 格式 x1_y1_x2_y2（0–1 百分比）。"""
|
||
if w / h >= tw / th:
|
||
ch, cw = h, h * tw / th
|
||
else:
|
||
cw, ch = w, w * th / tw
|
||
x1, y1 = (w - cw) / 2 / w, (h - ch) / 2 / h
|
||
x2, y2 = x1 + cw / w, y1 + ch / h
|
||
return f"{x1:.4f}_{y1:.4f}_{x2:.4f}_{y2:.4f}"
|
||
|
||
|
||
def get_token(app_id, secret):
|
||
q = urllib.parse.urlencode({"grant_type": "client_credential",
|
||
"appid": app_id, "secret": secret})
|
||
try:
|
||
r = http_json(f"{API}/cgi-bin/token?{q}")
|
||
except urllib.error.URLError as e:
|
||
die(f"网络请求失败：{e}")
|
||
err = wx_err(r)
|
||
if err:
|
||
die(f"获取 access_token 失败——{err}")
|
||
return r["access_token"]
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser(description="同步成稿到公众号草稿箱（不群发）")
|
||
ap.add_argument("dir", help="文章目录（含 article.html）")
|
||
ap.add_argument("--title", required=not any(a in sys.argv for a in ("--check", "--dry-run")))
|
||
ap.add_argument("--digest", default="")
|
||
ap.add_argument("--author", default="")
|
||
ap.add_argument("--cover", help="封面图路径（默认在文章目录找 cover.png/jpg/jpeg/webp）")
|
||
ap.add_argument("--check", action="store_true", help="只验证凭证与 IP 白名单")
|
||
ap.add_argument("--no-crop", action="store_true", help="不提交封面裁剪框参数")
|
||
ap.add_argument("--dry-run", action="store_true", help="不联网，打印执行计划")
|
||
args = ap.parse_args()
|
||
|
||
app_id = os.environ.get("WECHAT_APP_ID", "")
|
||
secret = os.environ.get("WECHAT_APP_SECRET", "")
|
||
|
||
# --check 只验凭证与 IP，不需要文章产物
|
||
if args.check:
|
||
if not (app_id and secret):
|
||
print("❌ 缺少环境变量 WECHAT_APP_ID / WECHAT_APP_SECRET\n"
|
||
" 获取路径：mp.weixin.qq.com → 设置与开发 → 基本配置\n"
|
||
" 设置方式：export WECHAT_APP_ID=xxx WECHAT_APP_SECRET=yyy（建议写进 ~/.zshrc）\n"
|
||
" 别把密钥写进任何 git 仓库。", file=sys.stderr)
|
||
sys.exit(2)
|
||
get_token(app_id, secret) # 失败时内部已 die 并翻译错误码
|
||
print("✔ 凭证与 IP 白名单验证通过，可以同步草稿箱")
|
||
return
|
||
|
||
# ---- 收集产物（dry-run 也走这段，但不联网） ----
|
||
html_path = os.path.join(args.dir, "article.html")
|
||
if not os.path.isfile(html_path):
|
||
die(f"找不到 {html_path}——先跑 md2wechat.py 生成 HTML", 2)
|
||
html_text = open(html_path, encoding="utf-8").read()
|
||
|
||
if args.title and len(args.title) > 64:
|
||
die(f"标题 {len(args.title)} 字，微信上限 64 字", 2)
|
||
if len(args.digest) > 120:
|
||
die(f"摘要 {len(args.digest)} 字，微信上限 120 字", 2)
|
||
|
||
cover = args.cover or next(
|
||
(os.path.join(args.dir, f"cover{ext}")
|
||
for ext in (".jpg", ".jpeg", ".png", ".webp")
|
||
if os.path.isfile(os.path.join(args.dir, f"cover{ext}"))), None)
|
||
|
||
# 正文里的本地图片 → 逐个经 uploadimg 换成微信域名 URL
|
||
local_imgs = re.findall(r'src="([^"]+)"', html_text)
|
||
local_imgs = [s for s in local_imgs if not s.startswith(("http://", "https://", "data:"))]
|
||
|
||
if args.dry_run:
|
||
print("—— DRY RUN（不联网，不需要凭证）——")
|
||
print(f"标题：{args.title or '（未提供）'}")
|
||
print(f"摘要：{args.digest or '（空）'}（{len(args.digest)}/120 字）")
|
||
print(f"封面：{cover or '❌ 未找到——用 --cover 指定或放 cover.png 进文章目录'}")
|
||
print(f"正文本地图片 {len(local_imgs)} 张：")
|
||
for s in local_imgs:
|
||
print(f" - {s}")
|
||
crop_note = "跳过（--no-crop）" if args.no_crop else f"2.35:1={crop_fractions(*img_size(cover), 2.35, 1)} · 1:1={crop_fractions(*img_size(cover), 1, 1)}"
|
||
print(f"裁剪框：{crop_note}")
|
||
print("执行动作：上传封面拿 thumb_media_id → 逐张上传正文图并替换 src → 提交 draft/add")
|
||
return
|
||
|
||
if not (app_id and secret):
|
||
print("❌ 缺少环境变量 WECHAT_APP_ID / WECHAT_APP_SECRET\n"
|
||
" 获取路径：mp.weixin.qq.com → 设置与开发 → 基本配置\n"
|
||
" 设置方式：export WECHAT_APP_ID=xxx WECHAT_APP_SECRET=yyy（建议写进 ~/.zshrc）\n"
|
||
" 别把密钥写进任何 git 仓库。", file=sys.stderr)
|
||
sys.exit(2)
|
||
|
||
if not args.title:
|
||
die("正式同步必须提供 --title", 2)
|
||
if not cover:
|
||
die("找不到封面图——先按封面 brief 生成 cover.png 放进文章目录，或用 --cover 指定", 2)
|
||
|
||
token = get_token(app_id, secret)
|
||
|
||
print("上传封面…")
|
||
r = post_file(f"{API}/cgi-bin/material/add_material?type=image&access_token={token}", cover)
|
||
err = wx_err(r)
|
||
if err:
|
||
die(f"封面上传失败——{err}")
|
||
thumb_media_id = r["media_id"]
|
||
print(f" ✔ thumb_media_id={thumb_media_id}")
|
||
|
||
for s in local_imgs:
|
||
p = os.path.join(args.dir, s)
|
||
if not os.path.isfile(p):
|
||
die(f"正文图片不存在：{p}")
|
||
print(f"上传正文图 {s}…")
|
||
r = post_file(f"{API}/cgi-bin/media/uploadimg?access_token={token}", p)
|
||
err = wx_err(r)
|
||
if err:
|
||
die(f"正文图上传失败（{s}）——{err}")
|
||
html_text = html_text.replace(f'src="{s}"', f'src="{r["url"]}"')
|
||
print(f" ✔ {s} → {r['url']}")
|
||
|
||
print("提交草稿…")
|
||
article = {"title": args.title, "content": html_text,
|
||
"thumb_media_id": thumb_media_id,
|
||
"need_open_comment": 0, "only_fans_can_comment": 0}
|
||
if not args.no_crop:
|
||
w, h = img_size(cover)
|
||
article["pic_crop_235_1"] = crop_fractions(w, h, 2.35, 1)
|
||
article["pic_crop_1_1"] = crop_fractions(w, h, 1, 1)
|
||
print(f" 裁剪框：2.35:1={article['pic_crop_235_1']} · 1:1={article['pic_crop_1_1']}")
|
||
if args.digest:
|
||
article["digest"] = args.digest
|
||
if args.author:
|
||
article["author"] = args.author
|
||
r = http_json(f"{API}/cgi-bin/draft/add?access_token={token}", {"articles": [article]})
|
||
err = wx_err(r)
|
||
if err:
|
||
die(f"草稿提交失败——{err}")
|
||
print(f"✔ 已进入草稿箱（media_id={r['media_id']}）\n"
|
||
f" 下一步：mp.weixin.qq.com → 内容与互动 → 发表 → 草稿，人工预览确认后再发表。\n"
|
||
f" 本 skill 到草稿箱为止，不执行群发。")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|