Wechatpay Payment Integration

{
  "LAST_CHECK_TIME": "2026-06-12T12:06:24+08:00",
  "LAST_UPDATE_TIME": "2026-06-12T12:06:28+08:00",
  "REMOTE_LAST_MODIFIED": "thu, 11 jun 2026 21:31:18 gmt",
  "REMOTE_ETAG": "\"60ff836e1d9eec113bcf8a3726e22ec5\""
}

#!/bin/bash
#
# 微信支付 APIv3 - P12 证书信息提取与签名工具
#
# 用法（推荐，签名时刻生成 TIMESTAMP / NONCE_STR）:
#   bash extract_and_sign.sh --filePath <P12> [--password <密码>] \
#     --method GET --url '/v3/pay/transactions/id/xxx?mchid=yyy'
#
# 用法（可选，自行提供时间戳与随机串）:
#   bash extract_and_sign.sh --filePath <P12> --method GET --url '...' \
#     --timestamp <秒级时间戳> --nonce_str <32位随机串>
#
# 用法（兼容，传入完整待签名串）:
#   bash extract_and_sign.sh --filePath <P12> --signString 'GET\n/...\n...\n\n'

set -euo pipefail

P12_FILE=""
P12_PASSWORD=""
HTTP_METHOD=""
REQUEST_URL=""
TIMESTAMP_IN=""
NONCE_IN=""
SIGN_STRING=""

usage() {
    echo "用法: bash extract_and_sign.sh --filePath <P12文件路径> [--password <P12密码>] \\"
    echo "        (--method <HTTP方法> --url '<请求URL路径+query>') | --signString '<待签名串>'"
    echo ""
    echo "参数说明:"
    echo "  --filePath      apiclient_cert.p12 文件路径（支持 file:///... 或 @/path/...）"
    echo "  --password      P12 密码（可选；未传时可能从 URL 中的 mchid/sp_mchid 尝试）"
    echo "  --method        HTTP 方法，如 GET（与 --url 搭配使用）"
    echo "  --url           请求 URL（含 path 与 query，以 / 开头）"
    echo "  --timestamp     秒级时间戳（可选；未传则在签名时自动生成）"
    echo "  --nonce_str     随机串（可选；未传则在签名时自动生成）"
    echo "  --signString    完整待签名串（兼容旧用法；含 \\\\n 转义换行）"
    exit 1
}

while [[ $# -gt 0 ]]; do
  case "$1" in
    --filePath)
      P12_FILE="${2:-}"
      shift 2
      ;;
    --password)
      P12_PASSWORD="${2:-}"
      shift 2
      ;;
    --method)
      HTTP_METHOD="${2:-}"
      shift 2
      ;;
    --url)
      REQUEST_URL="${2:-}"
      shift 2
      ;;
    --timestamp)
      TIMESTAMP_IN="${2:-}"
      shift 2
      ;;
    --nonce_str)
      NONCE_IN="${2:-}"
      shift 2
      ;;
    --signString)
      SIGN_STRING="${2:-}"
      shift 2
      ;;
    -h|--help)
      usage
      ;;
    *)
      echo "错误: 未知参数: $1"
      usage
      ;;
  esac
done

if [ -z "$P12_FILE" ]; then
    usage
fi

if [ -z "$SIGN_STRING" ]; then
    if [ -z "$HTTP_METHOD" ] || [ -z "$REQUEST_URL" ]; then
        echo "错误: 请提供 --method 与 --url，或提供 --signString"
        usage
    fi
    if [[ "$REQUEST_URL" != /* ]]; then
        echo "错误: --url 必须以 / 开头（path + query）"
        exit 1
    fi
    if [ -z "$TIMESTAMP_IN" ]; then
        TIMESTAMP_IN="$(date +%s)"
    fi
    if [ -z "$NONCE_IN" ]; then
        NONCE_IN="$(openssl rand -hex 16 | tr '[:lower:]' '[:upper:]')"
    fi
    SIGN_STRING="${HTTP_METHOD}"$'\n'"${REQUEST_URL}"$'\n'"${TIMESTAMP_IN}"$'\n'"${NONCE_IN}"$'\n\n'
elif [ -z "$TIMESTAMP_IN" ] || [ -z "$NONCE_IN" ]; then
  # 从完整待签名串解析时间戳与随机串（第 3、4 行）
  [ -z "$TIMESTAMP_IN" ] && TIMESTAMP_IN=$(printf '%b' "$SIGN_STRING" | sed -n '3p')
  [ -z "$NONCE_IN" ] && NONCE_IN=$(printf '%b' "$SIGN_STRING" | sed -n '4p')
fi

# 兼容 file:/// file: @ 前缀
if [[ "$P12_FILE" == file://* ]]; then
    P12_FILE="${P12_FILE#file://}"
    if [[ "$P12_FILE" != /* ]]; then
        P12_FILE="/$P12_FILE"
    fi
elif [[ "$P12_FILE" == file:* ]]; then
    P12_FILE="${P12_FILE#file:}"
fi
if [[ "$P12_FILE" == @* ]]; then
    P12_FILE="${P12_FILE#@}"
fi

if [[ "$P12_FILE" == "/path/to/"* ]] || [[ "$P12_FILE" == *"\\path\\to\\"* ]] || [[ "$P12_FILE" == *":\\path\\to\\"* ]]; then
    echo "错误: 请将 --filePath 参数替换为你本地 P12 证书的真实路径"
    exit 1
fi

if [ ! -f "$P12_FILE" ]; then
    echo "错误: P12 文件不存在: $P12_FILE"
    exit 1
fi

_is_pem_like_file() {
    local file="$1"
    local ext first
    ext="${file##*.}"
    ext=$(printf '%s' "$ext" | tr '[:upper:]' '[:lower:]')
    case "$ext" in
        pem|crt|cer|key) return 0 ;;
    esac
    first=$(head -n 1 "$file" 2>/dev/null || true)
    [[ "$first" == -----BEGIN* ]]
}

_fail_wrong_cert_format() {
    local file="$1"
    echo "错误: --filePath 指向的是 PEM/证书文件（$(basename "$file")），本脚本仅支持 PKCS#12 格式的 apiclient_cert.p12"
    echo "提示: 请将 --filePath 改为你本地的 apiclient_cert.p12 路径（证书压缩包内通常同时提供 .p12 与 .pem，请选用 .p12）"
    exit 1
}

if _is_pem_like_file "$P12_FILE"; then
    _fail_wrong_cert_format "$P12_FILE"
fi

_guess_p12_password() {
    local hint="$1"
    local pwd
    pwd=$(printf '%s' "$hint" | sed -n 's/.*[?&]mchid=\([^&]*\).*/\1/p' | head -n 1)
    if [ -z "$pwd" ]; then
        pwd=$(printf '%s' "$hint" | sed -n 's/.*[?&]sp_mchid=\([^&]*\).*/\1/p' | head -n 1)
    fi
    printf '%s' "$pwd"
}

# 密码候选：优先从 URL 提取 mchid / sp_mchid
PWD_HINT="${REQUEST_URL:-$SIGN_STRING}"

LEGACY_FLAG=""
PASSIN="pass:$P12_PASSWORD"

if openssl pkcs12 -in "$P12_FILE" -clcerts -nokeys -passin "$PASSIN" -legacy >/dev/null 2>&1; then
    LEGACY_FLAG="-legacy"
elif ! openssl pkcs12 -in "$P12_FILE" -clcerts -nokeys -passin "$PASSIN" >/dev/null 2>&1; then
    if [ -z "$P12_PASSWORD" ]; then
        CAND_PWD="$(_guess_p12_password "$PWD_HINT")"
        if [ -n "$CAND_PWD" ]; then
            P12_PASSWORD="$CAND_PWD"
            PASSIN="pass:$P12_PASSWORD"
            if openssl pkcs12 -in "$P12_FILE" -clcerts -nokeys -passin "$PASSIN" -legacy >/dev/null 2>&1; then
                LEGACY_FLAG="-legacy"
            elif ! openssl pkcs12 -in "$P12_FILE" -clcerts -nokeys -passin "$PASSIN" >/dev/null 2>&1; then
                echo "错误: 无法读取 P12 文件。可能需要 P12 密码，请在命令中追加: --password \"<P12密码>\"（常见为商户号）"
                exit 1
            fi
        else
            echo "错误: 无法读取 P12 文件。可能需要 P12 密码，请在命令中追加: --password \"<P12密码>\"（常见为商户号）"
            exit 1
        fi
    else
        if _is_pem_like_file "$P12_FILE"; then
            _fail_wrong_cert_format "$P12_FILE"
        fi
        echo "错误: 无法读取 P12 文件，请检查 --password 是否为 P12 密码（常见为商户号）"
        exit 1
    fi
fi

CERT_PEM=$(openssl pkcs12 -in "$P12_FILE" -clcerts -nokeys -passin "$PASSIN" $LEGACY_FLAG 2>/dev/null)

SERIAL=$(echo "$CERT_PEM" | openssl x509 -serial -noout 2>/dev/null | sed 's/serial=//')
if [ -z "$SERIAL" ]; then
    echo "错误: 无法提取证书序列号"
    exit 1
fi

SUBJECT=$(echo "$CERT_PEM" | openssl x509 -subject -noout -nameopt RFC2253 2>/dev/null)
MCHID=$(echo "$SUBJECT" | sed -n 's/.*CN=\([^,]*\).*/\1/p')
if [ -z "$MCHID" ]; then
    MCHID="（无法从证书 CN 字段提取，请手动确认）"
fi

PRIVKEY=$(openssl pkcs12 -in "$P12_FILE" -nocerts -nodes -passin "$PASSIN" $LEGACY_FLAG 2>/dev/null)
if [ -z "$PRIVKEY" ]; then
    echo "错误: 无法提取私钥"
    exit 1
fi

SIGNATURE=$(printf "%b" "$SIGN_STRING" | \
    openssl dgst -sha256 -sign <(echo "$PRIVKEY") 2>/dev/null | \
    openssl base64 -A)

if [ -z "$SIGNATURE" ]; then
    echo "错误: 签名失败"
    exit 1
fi

echo "---------- 签名结果开始 ----------"
echo "API 证书序列号（serial_no）: $SERIAL"
echo "时间戳（timestamp）: $TIMESTAMP_IN"
echo "随机串（nonce_str）: $NONCE_IN"
echo "API 证书中的商户号: $MCHID"
echo "签名值（signature）: $SIGNATURE"
echo "---------- 签名结果结束 ----------"

#
# 微信支付 APIv3 - P12 证书信息提取与签名工具 (Windows PowerShell 版)
#
# 推荐用法（签名时刻生成 TIMESTAMP / NONCE_STR）:
#   powershell -ExecutionPolicy Bypass -File extract_and_sign.ps1 `
#     -FilePath "apiclient_cert.p12" -Method GET -Url "/v3/pay/transactions/id/xxx?mchid=yyy"
#

param(
    [Parameter(Mandatory=$true, HelpMessage="Path to apiclient_cert.p12")]
    [string]$FilePath,

    [Parameter(Mandatory=$false, HelpMessage="P12 password (optional)")]
    [string]$Password = "",

    [Parameter(Mandatory=$false, HelpMessage="HTTP method, e.g. GET")]
    [string]$Method = "",

    [Parameter(Mandatory=$false, HelpMessage="Request URL path+query, starts with /")]
    [string]$Url = "",

    [Parameter(Mandatory=$false, HelpMessage="Unix timestamp seconds (optional)")]
    [string]$Timestamp = "",

    [Parameter(Mandatory=$false, HelpMessage="Nonce string (optional)")]
    [string]$NonceStr = "",

    [Parameter(Mandatory=$false, HelpMessage="Full sign string; use \\n for newlines")]
    [string]$SignString = ""
)

$ErrorActionPreference = "Stop"

function Get-MchidFromHint([string]$hint) {
    if ($hint -match '[\?&]mchid=([^&\\n]+)') { return $Matches[1] }
    if ($hint -match '[\?&]sp_mchid=([^&\\n]+)') { return $Matches[1] }
    return $null
}

if ([string]::IsNullOrEmpty($SignString)) {
    if ([string]::IsNullOrEmpty($Method) -or [string]::IsNullOrEmpty($Url)) {
        Write-Host "错误: 请提供 -Method 与 -Url，或提供 -SignString" -ForegroundColor Red
        exit 1
    }
    if (-not $Url.StartsWith("/")) {
        Write-Host "错误: -Url 必须以 / 开头（path + query）" -ForegroundColor Red
        exit 1
    }
    if ([string]::IsNullOrEmpty($Timestamp)) {
        $Timestamp = [DateTimeOffset]::UtcNow.ToUnixTimeSeconds().ToString()
    }
    if ([string]::IsNullOrEmpty($NonceStr)) {
        $bytes = New-Object byte[] 16
        [System.Security.Cryptography.RandomNumberGenerator]::Create().GetBytes($bytes)
        $NonceStr = ([BitConverter]::ToString($bytes) -replace '-', '')
    }
    $SignString = "$Method`n$Url`n$Timestamp`n$NonceStr`n`n"
} elseif ([string]::IsNullOrEmpty($Timestamp) -or [string]::IsNullOrEmpty($NonceStr)) {
    $lines = ($SignString -replace '\\n', "`n") -split "`n"
    if ($lines.Count -ge 4) {
        if ([string]::IsNullOrEmpty($Timestamp)) { $Timestamp = $lines[2] }
        if ([string]::IsNullOrEmpty($NonceStr)) { $NonceStr = $lines[3] }
    }
}

if ($FilePath.StartsWith("file://")) {
    $FilePath = $FilePath.Substring(7)
    if (-not $FilePath.StartsWith("/")) { $FilePath = "/" + $FilePath }
} elseif ($FilePath.StartsWith("file:")) {
    $FilePath = $FilePath.Substring(5)
}
if ($FilePath.StartsWith("@")) { $FilePath = $FilePath.Substring(1) }

if ($FilePath -like "/path/to/*" -or $FilePath -like "*\path\to\*" -or $FilePath -like "*:\path\to\*") {
    Write-Host "错误: 请将 -FilePath 参数替换为你本地 P12 证书的真实路径" -ForegroundColor Red
    exit 1
}

if (-not (Test-Path $FilePath)) {
    Write-Host "错误: P12 文件不存在: $FilePath" -ForegroundColor Red
    exit 1
}

function Test-PemLikeFile([string]$path) {
    $ext = [System.IO.Path]::GetExtension($path).ToLowerInvariant()
    if ($ext -in '.pem', '.crt', '.cer', '.key') { return $true }
    $first = Get-Content -Path $path -TotalCount 1 -ErrorAction SilentlyContinue
    return ($first -like '-----BEGIN*')
}

function Write-WrongCertFormatError([string]$path) {
    $name = [System.IO.Path]::GetFileName($path)
    Write-Host "错误: -FilePath 指向的是 PEM/证书文件（$name），本脚本仅支持 PKCS#12 格式的 apiclient_cert.p12" -ForegroundColor Red
    Write-Host "提示: 请将 -FilePath 改为你本地的 apiclient_cert.p12 路径（证书压缩包内通常同时提供 .p12 与 .pem，请选用 .p12）" -ForegroundColor Yellow
    exit 1
}

if (Test-PemLikeFile $FilePath) {
    Write-WrongCertFormatError $FilePath
}

$P12FullPath = (Resolve-Path $FilePath).Path
$pwdHint = "$Url$SignString"

try {
    $cert = New-Object System.Security.Cryptography.X509Certificates.X509Certificate2(
        $P12FullPath, $Password,
        [System.Security.Cryptography.X509Certificates.X509KeyStorageFlags]::Exportable
    )
} catch {
    if ([string]::IsNullOrEmpty($Password)) {
        $cand = Get-MchidFromHint $pwdHint
        if ($cand) {
            try {
                $cert = New-Object System.Security.Cryptography.X509Certificates.X509Certificate2(
                    $P12FullPath, $cand,
                    [System.Security.Cryptography.X509Certificates.X509KeyStorageFlags]::Exportable
                )
                $Password = $cand
            } catch {
                Write-Host '错误: 无法加载 P12 文件。可能需要 P12 密码，请在命令中追加: -Password "你的P12密码"（常见为商户号）' -ForegroundColor Red
                exit 1
            }
        } else {
            Write-Host '错误: 无法加载 P12 文件。可能需要 P12 密码，请在命令中追加: -Password "你的P12密码"（常见为商户号）' -ForegroundColor Red
            exit 1
        }
    } else {
        if (Test-PemLikeFile $FilePath) {
            Write-WrongCertFormatError $FilePath
        }
        Write-Host "错误: 无法加载 P12 文件，请检查 -Password 是否为 P12 密码（常见为商户号）" -ForegroundColor Red
        exit 1
    }
}

$serial = $cert.SerialNumber
if ([string]::IsNullOrEmpty($serial)) {
    Write-Host "错误: 无法提取证书序列号" -ForegroundColor Red
    exit 1
}

$subject = $cert.Subject
if ($subject -match 'CN=([^,]+)') { $mchid = $Matches[1].Trim() }
else { $mchid = "（无法从证书 CN 字段提取，请手动确认）" }

$signContent = $SignString -replace '\\n', "`n"
$bytes = [System.Text.Encoding]::UTF8.GetBytes($signContent)
$signatureBytes = $null

# Windows PowerShell 5.1 / .NET Framework：使用 PrivateKey + SignHash
$legacyRsa = $cert.PrivateKey
if ($null -ne $legacyRsa) {
    try {
        $sha256 = New-Object System.Security.Cryptography.SHA256CryptoServiceProvider
        $hash = $sha256.ComputeHash($bytes)
        $oid = [System.Security.Cryptography.CryptoConfig]::MapNameToOID("SHA256")
        $signatureBytes = $legacyRsa.SignHash($hash, $oid)
    } catch {
        $signatureBytes = $null
    } finally {
        if ($null -ne $sha256) { $sha256.Dispose() }
    }
}

# PowerShell 7+ / .NET 4.6+：使用 RSACertificateExtensions::GetRSAPrivateKey
if ($null -eq $signatureBytes) {
    try {
        $rsaType = [System.Security.Cryptography.X509Certificates.RSACertificateExtensions]
        $rsa = $rsaType::GetRSAPrivateKey($cert)
        if ($null -ne $rsa) {
            $signatureBytes = $rsa.SignData($bytes,
                [System.Security.Cryptography.HashAlgorithmName]::SHA256,
                [System.Security.Cryptography.RSASignaturePadding]::Pkcs1)
        }
    } catch {
        $signatureBytes = $null
    }
}

if ($null -eq $signatureBytes) {
    Write-Host "错误: 无法提取私钥或签名失败，请确认 P12 文件包含私钥" -ForegroundColor Red
    exit 1
}

try {
    $signature = [Convert]::ToBase64String($signatureBytes)
} catch {
    Write-Host "错误: 签名失败" -ForegroundColor Red
    exit 1
}

Write-Host "---------- 签名结果开始 ----------" -ForegroundColor Green
Write-Host "API 证书序列号（serial_no）: $serial"
Write-Host "时间戳（timestamp）: $Timestamp"
Write-Host "随机串（nonce_str）: $NonceStr"
Write-Host "API 证书中的商户号: $mchid"
Write-Host "签名值（signature）: $signature"
Write-Host "---------- 签名结果结束 ----------" -ForegroundColor Green

"""
wechatpay-docs-sync.py — 微信支付知识库远程同步工具

用法:
    python3 wechatpay-docs-sync.py update  # 检查远程并同步（含首次安装）
    python3 wechatpay-docs-sync.py --encoding utf-8 update  # 显式锁定 UTF-8（推荐 Windows）
"""

from __future__ import annotations

import json
import locale
import os
import shutil
import stat
import sys
import tarfile
import tempfile
import zipfile
from datetime import datetime, timedelta, timezone
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

VERSION = "1.0"

# ==== 脚本配置 ====

SCRIPT_DIR = Path(__file__).resolve().parent  # scripts directory/
SKILL_DIR = SCRIPT_DIR.parent  # skill directory/

DOCS_URL = "https://wx.gtimg.com/resource/wechatpay_api/wechatpay-docs.zip"
DOCS_TARGET_DIR = SKILL_DIR / "assets"
STATE_FILE = SCRIPT_DIR / ".wechatpay-docs-sync-state.json"
CHECK_INTERVAL_HOURS = 12

_TZ = timezone(timedelta(hours=8))

# ==== HTTP header（统一小写） ====

# 代理/网关可能会改写 Header 字段名大小写；字段名本身大小写不敏感。
# 这里统一将“字段名”转为大写进行匹配，避免因大小写变化导致取值失败。
H_LAST_MODIFIED = "LAST-MODIFIED"
H_ETAG = "ETAG"
H_CONTENT_LENGTH = "CONTENT-LENGTH"

# ==== 状态文件 key ====

S_LAST_CHECK = "LAST_CHECK_TIME"
S_LAST_UPDATE = "LAST_UPDATE_TIME"
S_REMOTE_MODIFIED = "REMOTE_LAST_MODIFIED"
S_REMOTE_ETAG = "REMOTE_ETAG"


USER_AGENT = f"{Path(__file__).stem}/{VERSION}"
IGNORED_FILES = {".DS_Store", "Thumbs.db", "__MACOSX"}
DOCS_FILE_GLOB = "*.md"
ZIP_FALLBACK_ENCODING = "cp437"
DEFAULT_IO_ENCODING = "utf-8"


# ==== 编码 ====


def _set_windows_console_utf8() -> None:
    if sys.platform != "win32":
        return
    try:
        import ctypes

        kernel32 = ctypes.windll.kernel32  # type: ignore[attr-defined]
        kernel32.SetConsoleOutputCP(65001)
        kernel32.SetConsoleCP(65001)
    except (OSError, AttributeError):
        pass


def _bootstrap_encoding(encoding: str = DEFAULT_IO_ENCODING) -> None:
    """锁定脚本 I/O 编码，降低系统区域/终端代码页对中文路径与输出的干扰。"""
    normalized = (encoding or DEFAULT_IO_ENCODING).strip().lower()
    os.environ["PYTHONIOENCODING"] = f"{normalized}:replace"

    for stream in (sys.stdout, sys.stderr):
        if hasattr(stream, "reconfigure"):
            try:
                stream.reconfigure(encoding=normalized, errors="replace")
            except (OSError, ValueError, AttributeError):
                pass

    for name in ("en_US.UTF-8", "C.UTF-8", "UTF-8"):
        try:
            locale.setlocale(locale.LC_ALL, name)
            break
        except locale.Error:
            continue

    if sys.platform == "win32":
        _set_windows_console_utf8()


def _parse_cli_args(argv: list[str]) -> tuple[str, str]:
    """解析可选 --encoding / -E，返回 (command, encoding)。"""
    encoding = DEFAULT_IO_ENCODING
    args = list(argv)
    i = 0
    while i < len(args):
        token = args[i]
        if token in ("--encoding", "-E") and i + 1 < len(args):
            encoding = args[i + 1]
            del args[i : i + 2]
            continue
        if token.startswith("--encoding="):
            encoding = token.split("=", 1)[1]
            del args[i]
            continue
        i += 1
    return (args[0] if args else "", encoding)


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


def _now_iso() -> str:
    """返回当前时间的 ISO 8601 字符串（东八区）。"""
    return datetime.now(_TZ).isoformat(timespec="seconds")


def _load_state() -> dict:
    """从 STATE_FILE 读取同步状态，文件不存在则返回空 dict。"""
    if STATE_FILE.exists():
        state = json.loads(STATE_FILE.read_text(encoding="utf-8"))
        # 兼容：部分环境可能会改写 header value 的大小写；本地 state 统一按小写存取。
        if isinstance(state.get(S_REMOTE_ETAG), str):
            state[S_REMOTE_ETAG] = state[S_REMOTE_ETAG].lower()
        if isinstance(state.get(S_REMOTE_MODIFIED), str):
            state[S_REMOTE_MODIFIED] = state[S_REMOTE_MODIFIED].lower()
        return state
    return {}


def _save_state(state: dict) -> None:
    """将同步状态写入 STATE_FILE。"""
    STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
    STATE_FILE.write_text(
        json.dumps(state, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
    )


# ==== 远程资源 ====


def _head_remote() -> dict:
    """HEAD 请求，返回 last_modified / etag / content_length 或 error。"""
    req = Request(DOCS_URL, method="HEAD")
    req.add_header("User-Agent", USER_AGENT)
    try:
        with urlopen(req, timeout=15) as resp:
            headers = {k.upper(): v for k, v in resp.headers.items()}
            # value 统一小写用于比较/落盘，避免被代理改写大小写导致误判
            lm = headers.get(H_LAST_MODIFIED)
            etag = headers.get(H_ETAG)
            return {
                H_LAST_MODIFIED: lm.lower() if isinstance(lm, str) else lm,
                H_ETAG: etag.lower() if isinstance(etag, str) else etag,
                H_CONTENT_LENGTH: headers.get(H_CONTENT_LENGTH),
            }
    except (URLError, HTTPError) as exc:
        return {"error": str(exc)}


def _within_interval(state: dict) -> bool:
    """判断距上次检查是否不足 CHECK_INTERVAL_HOURS 小时。"""
    ts = state.get(S_LAST_CHECK)
    if not ts:
        return False
    elapsed = datetime.now(_TZ) - datetime.fromisoformat(ts)
    return elapsed.total_seconds() < CHECK_INTERVAL_HOURS * 3600


def _remote_changed(state: dict, remote: dict) -> bool:
    """比较远程 ETag / Last-Modified 与本地记录，判断是否有变化。"""
    r_etag = remote.get(H_ETAG)
    if r_etag and state.get(S_REMOTE_ETAG):
        return str(r_etag).lower() != str(state[S_REMOTE_ETAG]).lower()
    r_lm = remote.get(H_LAST_MODIFIED)
    if r_lm and state.get(S_REMOTE_MODIFIED):
        return str(r_lm).lower() != str(state[S_REMOTE_MODIFIED]).lower()
    return True  # 无法判断时视为有变化


# ==== 下载与解压 ====


def _download(dest: Path) -> None:
    """从 DOCS_URL 下载压缩包到 dest，显示下载进度。"""
    req = Request(DOCS_URL)
    req.add_header("User-Agent", USER_AGENT)
    with urlopen(req, timeout=300) as resp:
        headers = {k.upper(): v for k, v in resp.headers.items()}
        total = headers.get(H_CONTENT_LENGTH)
        total = int(total) if total else None
        done = 0
        last_pct = -1
        last_mb = -1
        is_tty = sys.stdout.isatty()
        with open(dest, "wb") as fp:
            while True:
                chunk = resp.read(65536)
                if not chunk:
                    break
                fp.write(chunk)
                done += len(chunk)
                if total:
                    pct = done * 100 // total
                    if pct == last_pct:
                        continue
                    last_pct = pct
                    # 非 TTY（IDE 输出区、管道等）下 \r 无法覆写同行，按里程碑换行
                    if not is_tty and pct % 10 != 0 and pct < 100:
                        continue
                    msg = (
                        f"  下载: {done / 1048576:.1f} MB / "
                        f"{total / 1048576:.1f} MB ({pct}%)"
                    )
                    if is_tty:
                        print(f"\r{msg}", end="", flush=True)
                    else:
                        print(msg, flush=True)
                else:
                    mb = int(done / 1048576)
                    if is_tty:
                        print(f"\r  下载: {done / 1048576:.1f} MB", end="", flush=True)
                    elif mb > last_mb:
                        last_mb = mb
                        print(f"  下载: {done / 1048576:.1f} MB", flush=True)
        print()


def _cjk_char_count(text: str) -> int:
    return sum(1 for ch in text if "\u4e00" <= ch <= "\u9fff")


def _try_recover_mojibake(name: str) -> str:
    """UTF-8 文件名被误按 Latin-1/CP1252 解码时，尝试 latin-1→utf-8 还原。"""
    try:
        recovered = name.encode("latin-1").decode("utf-8")
    except (UnicodeDecodeError, UnicodeEncodeError):
        return name
    if _cjk_char_count(recovered) > _cjk_char_count(name):
        return recovered
    return name


def _decode_zip_filename(info: zipfile.ZipInfo) -> str:
    """跨平台还原 zip 内文件名（Windows 上 zf.extract 偶发中文乱码，须先修正）。"""
    name = info.filename.replace("\\", "/")
    if info.flag_bits & 0x800:
        return _try_recover_mojibake(name)
    for encoding in ("utf-8", "gbk", "gb18030"):
        try:
            decoded = name.encode(ZIP_FALLBACK_ENCODING).decode(encoding)
            if _cjk_char_count(decoded) >= _cjk_char_count(name):
                return decoded
        except (UnicodeDecodeError, UnicodeEncodeError):
            continue
    return _try_recover_mojibake(name)


def _win_long_path(path: Path) -> str:
    """Windows 长路径前缀，绕过 MAX_PATH(260) 限制。"""
    resolved = str(path.resolve())
    if resolved.startswith("\\\\?\\"):
        return resolved
    if resolved.startswith("\\\\"):
        return "\\\\?\\UNC\\" + resolved[2:]
    return "\\\\?\\" + resolved


def _mkdir_parents(path: Path) -> None:
    try:
        path.mkdir(parents=True, exist_ok=True)
    except OSError:
        if sys.platform != "win32":
            raise
        os.makedirs(_win_long_path(path), exist_ok=True)


def _write_bytes(path: Path, data: bytes) -> None:
    _mkdir_parents(path.parent)
    try:
        path.write_bytes(data)
    except OSError:
        if sys.platform != "win32":
            raise
        with open(_win_long_path(path), "wb") as fp:
            fp.write(data)


def _extract_zip(archive: Path, dest: Path) -> None:
    """手动解压 zip，避免 Windows 上 ZipFile.extract 写出乱码中文路径。"""
    dest.mkdir(parents=True, exist_ok=True)
    zip_kwargs: dict = {}
    if sys.version_info >= (3, 11):
        zip_kwargs["metadata_encoding"] = "utf-8"
    with zipfile.ZipFile(archive, **zip_kwargs) as zf:
        for info in zf.infolist():
            name = _decode_zip_filename(info)
            if not name or name.startswith("__MACOSX"):
                continue
            parts = name.split("/")
            if any(p in IGNORED_FILES for p in parts):
                continue
            target = dest.joinpath(*parts)
            if name.endswith("/") or info.is_dir():
                _mkdir_parents(target)
                continue
            with zf.open(info) as src:
                _write_bytes(target, src.read())


def _extract(archive: Path, dest: Path) -> None:
    """解压 zip 或 tar.gz 到 dest。"""
    dest.mkdir(parents=True, exist_ok=True)
    if zipfile.is_zipfile(archive):
        _extract_zip(archive, dest)
        return
    try:
        with tarfile.open(archive) as tf:
            tf.extractall(dest, filter="data")
        return
    except (tarfile.TarError, TypeError):
        pass
    raise RuntimeError(
        f"无法识别压缩格式，请确认下载链接是否为 zip 或 tar.gz 文件: {archive.name}"
    )


def _find_content_root(extract_dir: Path) -> Path:
    """若解压后只有单一顶层目录，则进入该目录作为实际内容根。"""
    items = [p for p in extract_dir.iterdir() if p.name != "__MACOSX"]
    if len(items) == 1 and items[0].is_dir():
        return items[0]
    return extract_dir


def _chmod_writable(path: Path) -> None:
    try:
        os.chmod(path, stat.S_IWRITE)
    except OSError:
        pass


def _on_rm_error(func, path, _exc_info) -> None:
    """Windows 上只读文件/目录删除失败时，先改权限再重试。"""
    _chmod_writable(Path(path))
    func(path)


def _unlink(path: Path) -> None:
    """删除单个文件或符号链接。"""
    _chmod_writable(path)
    try:
        path.unlink()
        return
    except OSError:
        if sys.platform != "win32":
            raise
    os.remove(_win_long_path(path))


def _rmdir(path: Path) -> None:
    """删除空目录。"""
    _chmod_writable(path)
    try:
        path.rmdir()
        return
    except OSError:
        if sys.platform != "win32":
            raise
    os.rmdir(_win_long_path(path))


def _remove_tree(root: Path) -> None:
    """删除目录树；Windows 下自底向上并使用长路径，避免 rmtree 在深层目录失败。"""
    if not root.exists():
        return
    if sys.platform == "win32":
        walk_root = _win_long_path(root.resolve())
        for dirpath, dirnames, filenames in os.walk(walk_root, topdown=False):
            for name in filenames:
                fp = os.path.join(dirpath, name)
                try:
                    _chmod_writable(Path(fp))
                    os.remove(fp)
                except OSError:
                    try:
                        os.remove(_win_long_path(Path(fp)))
                    except OSError as exc:
                        raise OSError(f"无法删除文件: {fp}") from exc
            for name in dirnames:
                dp = os.path.join(dirpath, name)
                try:
                    _chmod_writable(Path(dp))
                    os.rmdir(dp)
                except OSError:
                    try:
                        os.rmdir(_win_long_path(Path(dp)))
                    except OSError as exc:
                        raise OSError(f"无法删除目录: {dp}") from exc
        try:
            _rmdir(root)
        except OSError as exc:
            raise OSError(f"无法删除目录: {root}") from exc
        return
    shutil.rmtree(root, onerror=_on_rm_error)


def _copy_file(src: Path, dst: Path) -> None:
    """复制单个文件；Windows 下自动尝试长路径。"""
    dst.parent.mkdir(parents=True, exist_ok=True)
    try:
        shutil.copy2(src, dst)
        return
    except OSError:
        if sys.platform != "win32":
            raise
    shutil.copy2(_win_long_path(src), _win_long_path(dst))


def _copy_tree(src: Path, dst: Path) -> list[tuple[Path, str]]:
    """逐文件复制目录树，返回 (相对路径, 错误信息) 列表。"""
    dst.mkdir(parents=True, exist_ok=True)
    errors: list[tuple[Path, str]] = []
    for item in sorted(src.rglob("*")):
        if item.name in IGNORED_FILES or item.name == "__MACOSX":
            continue
        rel = item.relative_to(src)
        target = dst / rel
        if item.is_dir():
            try:
                target.mkdir(parents=True, exist_ok=True)
            except OSError as exc:
                if sys.platform == "win32":
                    try:
                        os.mkdir(_win_long_path(target), exist_ok=True)
                    except OSError as exc2:
                        errors.append((rel, str(exc2)))
                else:
                    errors.append((rel, str(exc)))
            continue
        try:
            _copy_file(item, target)
        except OSError as exc:
            errors.append((rel, str(exc)))
    return errors


def _clear_docs_dir() -> None:
    """删除 DOCS_TARGET_DIR 下的所有文件与子目录（保留 assets 目录本身）。"""
    if not DOCS_TARGET_DIR.exists():
        return
    resolved = DOCS_TARGET_DIR.resolve()
    skill_root = SKILL_DIR.resolve()
    if not str(resolved).startswith(str(skill_root)):
        raise RuntimeError(f"目标目录不在 skill 范围内，拒绝清空: {DOCS_TARGET_DIR}")
    for child in DOCS_TARGET_DIR.iterdir():
        if child.is_symlink():
            _unlink(child)
        elif child.is_dir():
            _remove_tree(child)
        else:
            _unlink(child)


# ==== 命令 ====


def _is_installed() -> bool:
    """判断本地知识库目录是否存在且非空。"""
    return DOCS_TARGET_DIR.exists() and any(DOCS_TARGET_DIR.iterdir())


def cmd_update() -> None:
    """检查远程版本；有更新或首次安装时下载、清空 assets 并写入新版本。"""
    state = _load_state()
    installed = _is_installed()

    if installed and _within_interval(state):
        last = state.get(S_LAST_CHECK, "未知")
        print(
            f"知识库已是最新（上次检查: {last}，{CHECK_INTERVAL_HOURS}H 内无需重复检查）。"
        )
        return

    print("正在检查远程文档版本…")
    remote = _head_remote()
    if "error" in remote:
        print(
            f"无法连接远程服务器，请检查网络后重试。\n  错误详情: {remote['error']}",
            file=sys.stderr,
        )
        sys.exit(1)

    state[S_LAST_CHECK] = _now_iso()

    if installed and not _remote_changed(state, remote):
        _save_state(state)
        print("远程文档未发生变化，当前已是最新，无需更新。")
        return

    if not installed:
        print("本地尚未安装知识库，开始首次下载…")
    else:
        print(
            f"检测到远程文档已更新（{remote.get(H_LAST_MODIFIED, '时间未知')}），开始下载…"
        )

    with tempfile.TemporaryDirectory() as tmp:
        tmp = Path(tmp)
        suffix = Path(DOCS_URL.split("?")[0]).suffix or ".zip"
        archive = tmp / f"docs{suffix}"

        _download(archive)

        print("下载完成，正在解压…")
        extract_dir = tmp / "out"
        _extract(archive, extract_dir)
        new_root = _find_content_root(extract_dir)

        print("正在写入知识库...")
        DOCS_TARGET_DIR.parent.mkdir(parents=True, exist_ok=True)
        _clear_docs_dir()
        copy_errors = _copy_tree(new_root, DOCS_TARGET_DIR)
        if copy_errors:
            print(
                f"写入知识库时出现 {len(copy_errors)} 个文件错误（常见于 Windows 长路径或权限问题）：",
                file=sys.stderr,
            )
            for rel, msg in copy_errors[:10]:
                print(f"  - {rel}: {msg}", file=sys.stderr)
            if len(copy_errors) > 10:
                print(f"  … 另有 {len(copy_errors) - 10} 个错误未列出", file=sys.stderr)
            sys.exit(1)

    state[S_LAST_UPDATE] = _now_iso()
    # 为了抵抗代理/网关对 header value 的大小写改写，这里落盘时统一做小写。
    state[S_REMOTE_MODIFIED] = str(remote.get(H_LAST_MODIFIED, "") or "").lower()
    state[S_REMOTE_ETAG] = str(remote.get(H_ETAG, "") or "").lower()
    _save_state(state)

    count = sum(1 for _ in DOCS_TARGET_DIR.rglob(DOCS_FILE_GLOB))
    print(f"更新完成，当前共 {count} 篇文档。")


# ==== 入口 ====

_USAGE = """\
用法: python3 wechatpay-docs-sync.py [--encoding utf-8] update

  update              检查远程是否有更新；有变化时下载并全量替换本地知识库（含首次安装）
                      默认 12 小时内不重复检查远程
  --encoding, -E ENC  锁定脚本 stdout/stderr 编码（默认 utf-8；Windows 推荐显式指定）

示例（Windows）:
  python wechatpay-docs-sync.py --encoding utf-8 update
  set PYTHONUTF8=1 && python wechatpay-docs-sync.py update"""


def main(argv: list[str] | None = None) -> None:
    raw = list(argv if argv is not None else sys.argv[1:])
    if not raw or raw[0] in ("-h", "--help"):
        print(_USAGE)
        sys.exit(0)

    cmd, encoding = _parse_cli_args(raw)
    _bootstrap_encoding(encoding)

    if cmd != "update":
        print(f"未识别到有效命令: {cmd or '(空)'}\n")
        print(_USAGE)
        sys.exit(1)

    cmd_update()


if __name__ == "__main__":
    main()