前言
近期基于某场景需要，注册并认证了微信开放平台，进一步需要根据已有的openid数据记录获取其unionid并建立关联，其中自然而然的方法是待其再次调用 oidc能力的时候补充存储，但这样在我的业务场景下会降低启用后未经历oidc的用户的使用体验。本着试一试的态度上网浏览还真得到了些许启发。
原理
微信公众平台提供基于Stable Access token的/cgi-bin/user/info接口能力，具体实现如下：
1. 获取Stable Access token
HTTPS 调用
POST https://api.weixin.qq.com/cgi-bin/stable_token 请求参数：
| 属性 | 类型 | 必填 | 说明 |
|---|---|---|---|
| grant_type | string | 是 | 填写 client_credential |
| appid | string | 是 | 账号唯一凭证，即 AppID，可在「微信公众平台 - 设置 - 开发设置」页中获得。（需要已经成为开发者，且账号没有异常状态） |
| secret | string | 是 | 账号唯一凭证密钥，即 AppSecret，获取方式同 appid |
| force_refresh | boolean | 否 | 默认使用 false。1. force_refresh = false 时为普通调用模式，access_token 有效期内重复调用该接口不会更新 access_token；2. 当force_refresh = true 时为强制刷新模式，会导致上次获取的 access_token 失效，并返回新的 access_token |
返回参数：
| 属性 | 类型 | 说明 |
|---|---|---|
| access_token | string | 获取到的凭证 |
| expires_in | number | 凭证有效时间，单位：秒。目前是7200秒之内的值。 |
请求体：
{
"grant_type": "client_credential",
"appid": "APPID",
"secret": "APPSECRET"
}2. 获取用户基本信息（UnionID机制）
接口调用请求说明 http请求方式: GET https://api.weixin.qq.com/cgi-bin/user/info?access_token=ACCESS_TOKEN&openid=OPENID&lang=zh_CN
参数说明：
| 参数 | 是否必须 | 说明 |
|---|---|---|
| access_token | 是 | 调用接口凭证 |
| openid | 是 | 普通用户的标识，对当前公众号唯一 |
| lang | 否 | 返回国家地区语言版本，zh_CN 简体，zh_TW 繁体，en 英语 |
返回说明：
正常情况下，微信会返回下述JSON数据包给公众号：
{
"subscribe": 1,
"openid": "o6_bmjrPTlm6_2sgVt7hMZOPfL2M",
"language": "zh_CN",
"subscribe_time": 1382694957,
"unionid": " o6_bmasdasdsad6_2sgVt7hMZOPfL",
"remark": "",
"groupid": 0,
"tagid_list":[128,2],
"subscribe_scene": "ADD_SCENE_QR_CODE",
"qr_scene": 98765,
"qr_scene_str": ""
}参数说明：
| 参数 | 说明 |
|---|---|
| subscribe | 用户是否订阅该公众号标识，值为0时，代表此用户没有关注该公众号，拉取不到其余信息。 |
| openid | 用户的标识，对当前公众号唯一 |
| language | 用户的语言，简体中文为zh_CN |
| subscribe_time | 用户关注时间，为时间戳。如果用户曾多次关注，则取最后关注时间 |
| unionid | 只有在用户将公众号绑定到微信开放平台账号后，才会出现该字段。 |
| remark | 公众号运营者对粉丝的备注，公众号运营者可在微信公众平台用户管理界面对粉丝添加备注 |
| groupid | 用户所在的分组ID（兼容旧的用户分组接口） |
| tagid_list | 用户被打上的标签ID列表 |
| subscribe_scene | 返回用户关注的渠道来源，ADD_SCENE_SEARCH 公众号搜索，ADD_SCENE_ACCOUNT_MIGRATION 公众号迁移，ADD_SCENE_PROFILE_CARD 名片分享，ADD_SCENE_QR_CODE 扫描二维码，ADD_SCENE_PROFILE_LINK 图文页内名称点击，ADD_SCENE_PROFILE_ITEM 图文页右上角菜单，ADD_SCENE_PAID 支付后关注，ADD_SCENE_WECHAT_ADVERTISEMENT 微信广告，ADD_SCENE_REPRINT 他人转载，ADD_SCENE_LIVESTREAM 视频号直播，ADD_SCENE_CHANNELS 视频号，ADD_SCENE_WXA 小程序关注，ADD_SCENE_OTHERS 其他 |
| qr_scene | 二维码扫码场景（开发者自定义） |
| qr_scene_str | 二维码扫码场景描述（开发者自定义） |
至此，成功获取openid和unionid的关联
自动化
将openid存到文件openid.csv：
user_wx_openid
xxxxxxx
yyyyyyy
zzzzzzz手动填写access_token（要是记录太多以至于wx默认的7200秒不够用，也可以修改程序使之自动化获取），运行open2Union.py：
import csv
import requests
import concurrent.futures
from tqdm import tqdm # 引入tqdm
from time import sleep
# 读取CSV文件，返回所有的openid列表
def read_openid_from_csv(filename):
openids = []
with open(filename, mode='r', encoding='utf-8') as file:
reader = csv.reader(file)
next(reader) # 如果csv文件有表头，跳过表头
for row in reader:
openids.append(row[0]) # 假设user_wx_openid在第一列
return openids
# 获取用户信息的函数
def fetch_user_info(openid, access_token):
url = f"https://api.weixin.qq.com/cgi-bin/user/info?access_token={access_token}&openid={openid}&lang=zh_CN"
try:
response = requests.get(url)
response.raise_for_status() # 确保请求成功
data = response.json()
if 'unionid' in data:
return openid, data['unionid'] # 返回openid和unionid
else:
print(f"没有返回unionid，openid: {openid}")
return openid, None
except requests.RequestException as e:
print(f"请求失败，openid: {openid}, 错误: {e}")
return openid, None
# 批量请求，使用并发执行，并引入tqdm进度条
def fetch_all_user_info(openids, access_token, max_workers=10):
results = []
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
# 使用tqdm显示进度条
futures = {executor.submit(fetch_user_info, openid, access_token): openid for openid in openids}
# tqdm进度条，total设置为openids的数量
for future in tqdm(concurrent.futures.as_completed(futures), total=len(futures), desc="Fetching User Info"):
openid, unionid = future.result()
if unionid:
results.append((openid, unionid))
else:
results.append((openid, '未找到unionid'))
return results
# 保存结果到文件
def save_results(results, output_file):
with open(output_file, mode='w', encoding='utf-8', newline='') as file:
writer = csv.writer(file)
writer.writerow(['openid', 'unionid'])
for result in results:
writer.writerow(result)
# 主函数
def main():
access_token = 'xxxxxxxxxxx' # 这里替换为你自己的access_token
input_file = 'openid.csv' # 输入CSV文件名
output_file = 'user_unionid.csv' # 输出CSV文件名
# 读取openid列表
openids = read_openid_from_csv(input_file)
# 获取所有用户的unionid信息
results = fetch_all_user_info(openids, access_token)
# 保存结果到CSV文件
save_results(results, output_file)
print(f"结果已保存到{output_file}")
if __name__ == '__main__':
main()
输出结果user_unionid.csv：
user_wx_openid,unionid
xxxxxxx,aaaaaaa
yyyyyyy,bbbbbbb
zzzzzzz,ccccccc其中，如果用户已经取关了公众号，会在unionid列呈现：未找到unionid
最后
附上这位网友大开的脑洞：
参考资料：
如何通过unionId获取公众好号的openId? | 微信开放社区
通过微信用户的openid获取用户的头像,昵称,性别等信息!_根据openid获取用户信息-CSDN博客
如何解决invalid credential, access_token is invalid ? | 微信开放社区
developers.weixin.qq.com/doc/offiaccount/User_Management/Get_users_basic_information_UnionID.html
微信小程序与公众号关联（同一主体），获取unionId并关联公众号openid_通过unionid获取公众号openid-CSDN博客
版权属于：soarli
本文链接：https://blog.soarli.top/archives/745.html
转载时须注明出处及本声明。
