soul-yongping/scripts/wechat_miniprogram_release.py

#!/usr/bin/env python3
"""
微信小程序发布辅助：上传（调开发者工具 CLI）+ 审核状态查询 + 尝试 API 提审。

重要说明（微信官方限制）：
- 代码上传：可用本机「微信开发者工具」CLI 或 miniprogram-ci。
- submit_audit：开放平台文档标明主要为「第三方平台代调用」；自有主体使用小程序 appid+secret
  换取的 access_token 调用时，常见返回 errcode=86000（仅允许第三方代调用），此时必须在
  mp 后台手动点「提交审核」。
- 「自动过审」不可能由开发者脚本保证：是否通过由微信审核决定。
"""
from __future__ import annotations

import argparse
import json
import os
import subprocess
import sys
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

# 登录 mp 后进入「版本管理 / 开发版本」列表（选体验版、提交审核均在此页操作；具体路由以微信后台为准）。
MP_VERSION_MANAGE_URL = (
    "https://mp.weixin.qq.com/wxopen/wacodepage?action=getcodepage&lang=zh_CN"
)


def _repo_root() -> Path:
    return Path(__file__).resolve().parent.parent


def _load_partial_env_file(path: Path, keys: tuple[str, ...]) -> None:
    if not path.is_file():
        return
    for raw in path.read_text(encoding="utf-8", errors="ignore").splitlines():
        line = raw.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        k, v = line.split("=", 1)
        k, v = k.strip(), v.strip().strip('"').strip("'")
        if k in keys and not os.environ.get(k):
            os.environ[k] = v


def ensure_wechat_env_from_soul_api() -> None:
    """若未 export 凭证，则从 soul-api/.env.production 或 .env 读取 WECHAT_APPID/SECRET（不写日志）。"""
    if os.environ.get("WECHAT_APPID") and os.environ.get("WECHAT_APPSECRET"):
        return
    root = _repo_root()
    for name in (".env.production", ".env"):
        _load_partial_env_file(
            root / "soul-api" / name, ("WECHAT_APPID", "WECHAT_APPSECRET")
        )
        if os.environ.get("WECHAT_APPID") and os.environ.get("WECHAT_APPSECRET"):
            return


def _get(url: str) -> dict:
    req = urllib.request.Request(url, method="GET")
    with urllib.request.urlopen(req, timeout=60) as resp:
        return json.loads(resp.read().decode())


def _post_json(url: str, body: dict) -> dict:
    data = json.dumps(body, ensure_ascii=False).encode("utf-8")
    req = urllib.request.Request(
        url,
        data=data,
        method="POST",
        headers={"Content-Type": "application/json; charset=utf-8"},
    )
    with urllib.request.urlopen(req, timeout=60) as resp:
        return json.loads(resp.read().decode())


def get_access_token(appid: str, secret: str) -> str:
    q = urllib.parse.urlencode(
        {"grant_type": "client_credential", "appid": appid, "secret": secret}
    )
    url = f"https://api.weixin.qq.com/cgi-bin/token?{q}"
    try:
        data = _get(url)
    except urllib.error.HTTPError as e:
        raise SystemExit(f"获取 access_token HTTP 错误: {e}") from e
    if data.get("errcode"):
        raise SystemExit(f"获取 access_token 失败: {data}")
    token = data.get("access_token")
    if not token:
        raise SystemExit(f"获取 access_token 无 access_token 字段: {data}")
    return token


def cmd_audit_status(appid: str, secret: str) -> None:
    token = get_access_token(appid, secret)
    url = f"https://api.weixin.qq.com/wxa/get_latest_auditstatus?access_token={urllib.parse.quote(token)}"
    try:
        data = _get(url)
    except urllib.error.HTTPError as e:
        raise SystemExit(f"get_latest_auditstatus HTTP 错误: {e}") from e
    print(json.dumps(data, ensure_ascii=False, indent=2))
    # 常见 status: 0 审核成功 1 审核被拒绝 2 审核中 3 已撤回 4 审核延后


def cmd_get_category(appid: str, secret: str) -> None:
    token = get_access_token(appid, secret)
    url = f"https://api.weixin.qq.com/wxa/get_category?access_token={urllib.parse.quote(token)}"
    try:
        data = _get(url)
    except urllib.error.HTTPError as e:
        raise SystemExit(f"get_category HTTP 错误: {e}") from e
    print(json.dumps(data, ensure_ascii=False, indent=2))


def _first_item_from_category(data: dict) -> dict | None:
    lst = data.get("category_list")
    if not lst or not isinstance(lst, list):
        return None
    c = lst[0]
    if not isinstance(c, dict):
        return None
    first_class = c.get("first_class") or ""
    second_class = c.get("second_class") or ""
    first_id = c.get("first_id")
    second_id = c.get("second_id")
    if first_id is None or second_id is None:
        return None
    item: dict = {
        "address": "pages/index/index",
        "tag": "阅读 创业",
        "first_class": first_class,
        "second_class": second_class,
        "first_id": int(first_id),
        "second_id": int(second_id),
        "title": "首页",
    }
    third_class = c.get("third_class")
    third_id = c.get("third_id")
    if third_class and third_id:
        item["third_class"] = third_class
        item["third_id"] = int(third_id)
    return item


def cmd_submit_audit(
    appid: str,
    secret: str,
    version_desc: str,
    item_json: Path | None,
    privacy_api_not_use: bool | None,
) -> dict:
    token = get_access_token(appid, secret)
    if item_json and item_json.is_file():
        payload = json.loads(item_json.read_text(encoding="utf-8"))
        item_list = payload.get("item_list")
        if not item_list:
            raise SystemExit("item_json 中缺少 item_list")
    else:
        cat_url = f"https://api.weixin.qq.com/wxa/get_category?access_token={urllib.parse.quote(token)}"
        cat = _get(cat_url)
        if cat.get("errcode"):
            raise SystemExit(f"get_category 失败: {cat}")
        one = _first_item_from_category(cat)
        if not one:
            raise SystemExit(
                "无法从 get_category 构造审核项；请在公众平台配置服务类目，"
                "或使用 --item-json 指定完整 item_list（见 scripts/miniprogram_audit_item.example.json）。"
            )
        item_list = [one]

    body: dict = {
        "item_list": item_list,
        "version_desc": version_desc[:400] if version_desc else "版本更新",
    }
    if privacy_api_not_use is True:
        body["privacy_api_not_use"] = True
    elif privacy_api_not_use is False:
        body["privacy_api_not_use"] = False

    submit_url = f"https://api.weixin.qq.com/wxa/submit_audit?access_token={urllib.parse.quote(token)}"
    try:
        data = _post_json(submit_url, body)
    except urllib.error.HTTPError as e:
        raise SystemExit(f"submit_audit HTTP 错误: {e}") from e
    print(json.dumps(data, ensure_ascii=False, indent=2))
    if data.get("errcode") == 86000:
        print(
            "\n说明 errcode=86000：该接口仅支持「第三方平台」代小程序调用。"
            "自有主体请在浏览器打开公众平台 → 管理 → 版本管理 → 提交审核。\n"
            "可先运行: python3 scripts/wechat_miniprogram_release.py open-mp",
            file=sys.stderr,
        )
    elif data.get("errcode") == 61039:
        print(
            "\n说明 errcode=61039：上传后隐私/代码检测任务未完成，请等待数分钟后再提交审核。",
            file=sys.stderr,
        )
    return data


def cmd_upload(version: str, desc: str) -> None:
    root = Path(__file__).resolve().parent.parent
    sh = root / "scripts" / "miniprogram_upload.sh"
    if not sh.is_file():
        raise SystemExit(f"未找到 {sh}")
    r = subprocess.run([str(sh), version, desc], check=False)
    if r.returncode != 0:
        raise SystemExit(r.returncode)


def cmd_open_mp() -> None:
    url = "https://mp.weixin.qq.com/"
    try:
        subprocess.run(["open", url], check=False)
    except FileNotFoundError:
        print(url)


def cmd_open_mp_version() -> None:
    """打开版本管理（开发版本列表，可设体验版、提交审核）。"""
    try:
        subprocess.run(["open", MP_VERSION_MANAGE_URL], check=False)
    except FileNotFoundError:
        print(MP_VERSION_MANAGE_URL)


def cmd_upload_open(version: str, desc: str) -> None:
    cmd_upload(version, desc)
    print(
        "\n下一步（微信未开放「一键设为体验版」API，需在网页上点两次）：\n"
        "1）在「开发版本」列表找到刚上传的版本号；\n"
        "2）点击「选为体验版」；\n"
        "3）同一列表对该版本点击「提交审核」（或先体验再提审）。\n",
        file=sys.stderr,
    )
    cmd_open_mp_version()


def main() -> None:
    p = argparse.ArgumentParser(description="微信小程序上传与审核辅助")
    sub = p.add_subparsers(dest="cmd", required=True)

    p_up = sub.add_parser("upload", help="调用微信开发者工具 CLI 上传（需本机已登录）")
    p_up.add_argument(
        "--version",
        "-v",
        default=os.environ.get("MINIPROGRAM_DEFAULT_VERSION", "1.7.1"),
        help="版本号，默认 1.7.1 或环境变量 MINIPROGRAM_DEFAULT_VERSION",
    )
    p_up.add_argument(
        "--desc",
        "-d",
        default="",
        help="版本说明，默认同「版本 v<版本号>」",
    )

    p_st = sub.add_parser("audit-status", help="查询最近一次审核状态（需 WECHAT_APPID/SECRET）")
    p_cat = sub.add_parser("get-category", help="拉取已配置服务类目 JSON（构造提审项用）")
    p_sa = sub.add_parser(
        "submit-audit",
        help="尝试调用 submit_audit（自有主体常会 86000，需改 mp 后台手动提审）",
    )
    p_sa.add_argument("--version-desc", default="版本更新", help="审核说明 version_desc")
    p_sa.add_argument(
        "--item-json",
        type=Path,
        help="自定义 item_list 的 JSON 文件（含 item_list 数组）",
    )
    p_sa.add_argument(
        "--privacy-api-not-use",
        action=argparse.BooleanOptionalAction,
        default=None,
        help="是否声明未使用检测到的隐私接口（与微信报错 61040 相关时再用）",
    )

    sub.add_parser("open-mp", help="在浏览器打开 mp 首页")
    sub.add_parser(
        "open-version",
        help="打开版本管理页（开发版本：选体验版、提交审核；需已登录 mp）",
    )

    p_uo = sub.add_parser(
        "upload-open",
        help="上传代码并打开版本管理页（设体验版、提审在浏览器完成）",
    )
    p_uo.add_argument(
        "--version",
        "-v",
        default=os.environ.get("MINIPROGRAM_DEFAULT_VERSION", "1.7.1"),
    )
    p_uo.add_argument("--desc", "-d", default="", help="默认：版本 v<版本号>")

    p_rel = sub.add_parser("release", help="先 upload 再 submit-audit（提审失败仍可到后台操作）")
    p_rel.add_argument(
        "--version",
        "-v",
        default=os.environ.get("MINIPROGRAM_DEFAULT_VERSION", "1.7.1"),
    )
    p_rel.add_argument("--desc", "-d", default="", help="上传说明，默认：版本 v<版本号>")
    p_rel.add_argument("--version-desc", default="", help="提交审核说明，默认同上传说明")
    p_rel.add_argument("--item-json", type=Path)
    p_rel.add_argument(
        "--privacy-api-not-use",
        action=argparse.BooleanOptionalAction,
        default=None,
    )

    args = p.parse_args()

    if args.cmd in (
        "audit-status",
        "get-category",
        "submit-audit",
        "release",
    ):
        ensure_wechat_env_from_soul_api()

    appid = os.environ.get("WECHAT_APPID", "").strip()
    secret = os.environ.get("WECHAT_APPSECRET", "").strip()

    if args.cmd == "upload":
        desc = args.desc.strip() or f"版本 v{args.version}"
        cmd_upload(args.version, desc)
        return
    if args.cmd == "open-mp":
        cmd_open_mp()
        return
    if args.cmd == "open-version":
        cmd_open_mp_version()
        return
    if args.cmd == "upload-open":
        d = args.desc.strip() or f"版本 v{args.version}"
        cmd_upload_open(args.version, d)
        return
    if args.cmd == "release":
        d = args.desc.strip() or f"版本 v{args.version}"
        cmd_upload(args.version, d)
        if not appid or not secret:
            print(
                "未设置 WECHAT_APPID / WECHAT_APPSECRET，跳过 submit-audit。",
                file=sys.stderr,
            )
            cmd_open_mp_version()
            return
        vd = (args.version_desc or "").strip() or d
        res = cmd_submit_audit(
            appid, secret, vd, args.item_json, args.privacy_api_not_use
        )
        if res.get("errcode") == 86000:
            cmd_open_mp_version()
        return

    if not appid or not secret:
        raise SystemExit("请设置环境变量 WECHAT_APPID、WECHAT_APPSECRET（与 soul-api 一致即可）")

    if args.cmd == "audit-status":
        cmd_audit_status(appid, secret)
    elif args.cmd == "get-category":
        cmd_get_category(appid, secret)
    elif args.cmd == "submit-audit":
        cmd_submit_audit(
            appid, secret, args.version_desc, args.item_json, args.privacy_api_not_use
        )


if __name__ == "__main__":
    main()