为3000+家长发送孩子学习照片的通知具体实现流程和方案

一、技术实现核心前置条件

在进行技术开发前，需先完成以下准备工作，确保后续开发顺畅：

1. 已拥有认证通过的微信服务号，获取到核心凭证：AppID（应用唯一标识）、AppSecret（应用密钥，需保密，不可明文暴露在前端）。
2. 已搭建基础技术架构：后端服务（Java/SpringBoot、Python/Flask/Django、PHP 等均可）、数据库（MySQL 推荐，用于存储绑定关系）、云存储（阿里云 OSS / 腾讯云 COS，用于存放孩子照片，避免本地存储容量不足）。
3. 已完成「家长手机号 ↔ 公众号 OpenID」的绑定（关联关系存入 MySQL 数据库，核心字段：id、parent_mobile（家长手机号）、wechat_openid（微信唯一标识）、student_id（学生 ID）、create_time（绑定时间））。
4. 已完成「学生信息 ↔ 照片信息」的关联（数据库存储字段：student_id、student_name（学生姓名）、photo_url（照片云存储地址）、photo_update_time（照片更新时间））。

二、核心技术方案选型（2 种主流实现）

根据需求场景，优先推荐「模板消息推送」（支持批量精准推送，不限用户触发，适合定期发送学习照片），其次是「客服消息推送」（适用于用户主动查询场景），以下分别详细讲解技术实现。

方案一：模板消息推送（推荐，批量定向发送核心方案）

1. 实现流程概述

1. 申请微信模板消息权限及符合需求的消息模板；
2. 获取微信接口调用凭证access_token（所有微信接口的通用凭证）；
3. 从数据库批量查询家长 OpenID、对应学生照片 URL 等信息；
4. 调用微信模板消息接口，分批次推送消息给家长；
5. 记录推送结果，便于后续统计排查。

2. 关键步骤技术实现

步骤 1：申请模板消息模板

• 登录微信公众平台 → 「功能」→ 「模板消息」→ 「模板库」→ 申请符合需求的模板，示例模板内容：

  plaintext

  【XX学校/机构】您的孩子{{studentName.DATA}}近期学习照片已更新，点击下方链接即可查看高清照片： {{photoUrl.DATA}} 温馨提示：照片仅用于家校沟通，请勿随意转发。

• 申请通过后，获取模板 ID（template_id，后续接口调用需使用）。

步骤 2：获取微信接口凭证access_token

微信所有接口调用均需携带access_token，有效期 2 小时（7200 秒），需实现缓存机制（避免频繁调用接口导致限流）。

接口信息

• 接口地址：https://api.weixin.qq.com/cgi-bin/token
• 请求方式：GET
• 请求参数：

  参数名	说明
  grant_type	固定值：client_credential
  appid	服务号的 AppID
  secret	服务号的 AppSecret

代码示例（Python/Flask）

python

运行

import requests import redis import json from datetime import datetime, timedelta # 初始化Redis（用于缓存access_token，无Redis可使用本地缓存/数据库缓存） redis_client = redis.Redis(host="127.0.0.1", port=6379, db=0, password="你的Redis密码") # 配置信息 APPID = "你的服务号AppID" APPSECRET = "你的服务号AppSecret" ACCESS_TOKEN_KEY = "wechat_access_token" def get_wechat_access_token(): # 先从Redis获取缓存的access_token cached_token = redis_client.get(ACCESS_TOKEN_KEY) if cached_token: return cached_token.decode("utf-8") # 缓存不存在，调用接口获取 url = f"https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid={APPID}&secret={APPSECRET}" response = requests.get(url) result = json.loads(response.text) if "access_token" in result: access_token = result["access_token"] expires_in = result["expires_in"] # 7200秒 # 缓存到Redis，过期时间比接口返回少300秒，避免过期 redis_client.setex(ACCESS_TOKEN_KEY, expires_in - 300, access_token) return access_token else: # 抛出异常或记录错误（如AppID/AppSecret错误） raise Exception(f"获取access_token失败：{result}")

步骤 3：批量调用模板消息接口推送通知

接口信息

• 接口地址：https://api.weixin.qq.com/cgi-bin/message/template/send
• 请求方式：POST
• 请求头：Content-Type: application/json
• 请求参数（JSON 格式）：

  json

  { "touser": "家长的OpenID", // 单个家长的OpenID "template_id": "你的模板ID", // 申请到的模板消息ID "url": "可选，点击模板消息跳转的H5页面地址（照片查看页面）", "topcolor": "#FF0000", // 模板顶部颜色，可选 "data": { "studentName": { "value": "张三", // 学生姓名 "color": "#173177" }, "photoUrl": { "value": "https://xxx.oss-cn-beijing.aliyuncs.com/photos/zhangsan.jpg", // 照片URL "color": "#173177" } } }

代码示例（Python/Flask，批量分批次推送）

python

运行

import pymysql import time # 数据库配置 DB_CONFIG = { "host": "127.0.0.1", "user": "数据库用户名", "password": "数据库密码", "database": "你的数据库名", "charset": "utf8mb4" } # 分批次配置（避免接口限流，每批次500人，间隔15分钟） BATCH_SIZE = 500 BATCH_INTERVAL = 15 * 60 # 秒 def get_parent_student_photo_data(): """从数据库查询家长OpenID、学生姓名、照片URL关联数据""" conn = pymysql.connect(**DB_CONFIG) cursor = conn.cursor(pymysql.cursors.DictCursor) # 关联家长表和学生照片表，查询有效数据 sql = """ SELECT p.wechat_openid, s.student_name, s.photo_url FROM parent_wechat_bind p LEFT JOIN student_photo s ON p.student_id = s.student_id WHERE p.wechat_openid IS NOT NULL AND s.photo_url IS NOT NULL """ cursor.execute(sql) data_list = cursor.fetchall() cursor.close() conn.close() return data_list def send_template_message(access_token, openid, template_id, student_name, photo_url): """发送单个模板消息""" url = f"https://api.weixin.qq.com/cgi-bin/message/template/send?access_token={access_token}" payload = { "touser": openid, "template_id": template_id, "url": photo_url, # 点击跳转至照片页面 "topcolor": "#333333", "data": { "studentName": { "value": student_name, "color": "#173177" }, "photoUrl": { "value": f"点击查看：{photo_url}", "color": "#0088ff" } } } try: response = requests.post(url, json=payload) result = json.loads(response.text) if result.get("errcode") == 0: print(f"推送成功：OpenID={openid}，学生={student_name}") return True else: print(f"推送失败：OpenID={openid}，错误信息={result}") return False except Exception as e: print(f"推送异常：OpenID={openid}，异常信息={str(e)}") return False def batch_send_template_messages(): """批量分批次发送模板消息""" # 1. 获取access_token try: access_token = get_wechat_access_token() except Exception as e: print(f"获取access_token失败：{str(e)}") return # 2. 获取待推送数据 data_list = get_parent_student_photo_data() if not data_list: print("暂无待推送的家长数据") return # 3. 分批次推送 total_count = len(data_list) template_id = "你的模板ID" # 替换为实际模板ID for i in range(0, total_count, BATCH_SIZE): batch_data = data_list[i:i+BATCH_SIZE] batch_num = (i // BATCH_SIZE) + 1 print(f"开始推送第{batch_num}批次，共{len(batch_data)}条数据") for item in batch_data: openid = item["wechat_openid"] student_name = item["student_name"] photo_url = item["photo_url"] send_template_message(access_token, openid, template_id, student_name, photo_url) time.sleep(0.5) # 单个推送间隔0.5秒，避免瞬时请求过多 # 最后一批次无需等待 if i + BATCH_SIZE < total_count: print(f"第{batch_num}批次推送完成，等待{BATCH_INTERVAL/60}分钟后推送下一批次") time.sleep(BATCH_INTERVAL) print(f"所有批次推送完成，总计推送{total_count}条数据") # 执行批量推送 if __name__ == "__main__": batch_send_template_messages()

步骤 4：关键注意事项

• 分批次推送：3000 位家长建议分 6 批次（每批次 500 人），间隔 10-30 分钟，避免触发微信接口限流（微信默认单个公众号模板消息接口日调用限额为 10 万次，单批次请求不宜过多）。
• 照片权限控制：跳转的 H5 照片页面需增加验证（如手机号验证码、OpenID 校验），避免非对应家长查看照片，保障隐私。
• 异常处理：对推送失败的家长（如 OpenID 无效、用户已取消关注），需记录日志，后续手动排查补推。

方案二：客服消息推送（用户主动触发场景）

1. 实现流程概述

1. 配置公众号服务器回调（接收用户消息）；
2. 用户在公众号对话框发送关键词（如 “我的孩子照片”）；
3. 后端接收回调消息，通过 OpenID 查询对应学生照片；
4. 调用客服消息接口，给用户发送照片（或照片链接）。

2. 关键步骤技术实现

步骤 1：配置公众号消息推送回调

• 登录微信公众平台 → 「设置与开发」→ 「基本配置」→ 「服务器配置」→ 开启并填写以下信息：
  • 服务器地址（URL）：你的后端接口地址（需为 HTTPS 协议，微信要求），用于接收微信推送的用户消息；
  • Token：自定义字符串（如 “wechat_token_123”，后端需用该 Token 验证消息合法性）；
  • EncodingAESKey：可选，用于消息加密，建议填写并使用安全模式。

步骤 2：后端接收用户消息并验证合法性

代码示例（Python/Flask，接收并验证微信消息）

python

运行

from flask import Flask, request, abort import hashlib import xml.etree.ElementTree as ET app = Flask(__name__) # 公众号配置的Token WECHAT_TOKEN = "你的微信Token" def validate_wechat_signature(signature, timestamp, nonce): """验证微信消息签名，确保消息来自微信服务器""" # 1. 将token、timestamp、nonce按字典序排序 params = [WECHAT_TOKEN, timestamp, nonce] params.sort() # 2. 拼接为字符串并进行sha1加密 sign_str = "".join(params).encode("utf-8") sha1_sign = hashlib.sha1(sign_str).hexdigest() # 3. 对比加密结果与签名是否一致 return sha1_sign == signature @app.route("/wechat/callback", methods=["GET", "POST"]) def wechat_callback(): # GET请求：微信验证服务器有效性 if request.method == "GET": signature = request.args.get("signature") timestamp = request.args.get("timestamp") nonce = request.args.get("nonce") echostr = request.args.get("echostr") if validate_wechat_signature(signature, timestamp, nonce): return echostr else: abort(403) # POST请求：接收用户发送的消息 elif request.method == "POST": # 解析XML格式消息 xml_data = request.data root = ET.fromstring(xml_data) # 提取消息核心字段 to_user_name = root.find("ToUserName").text # 公众号ID from_user_name = root.find("FromUserName").text # 用户OpenID msg_type = root.find("MsgType").text # 消息类型（text=文本消息） content = root.find("Content").text if msg_type == "text" else "" # 用户发送的文本内容 # 处理用户查询照片请求 if msg_type == "text" and content in ["照片", "我的孩子照片", "查询照片"]: # 异步处理消息推送（避免同步请求超时） from threading import Thread Thread(target=send_custom_photo_message, args=(from_user_name,)).start() # 微信要求返回空XML，否则会持续推送消息 return """<xml> <ToUserName><![CDATA[{from_user}]]></ToUserName> <FromUserName><![CDATA[{to_user}]]></FromUserName> <CreateTime>{time}</CreateTime> <MsgType><![CDATA[text]]></MsgType> <Content><![CDATA[正在为您查询孩子照片，请稍候...]]></Content> </xml>""".format( from_user=from_user_name, to_user=to_user_name, time=int(time.time()) ) if __name__ == "__main__": app.run(host="0.0.0.0", port=443, ssl_context=("你的证书文件.pem", "你的私钥文件.key"))

步骤 3：调用客服消息接口发送照片

客服消息支持直接发送图片（单张≤5M）或图片链接，以下是发送图片的代码示例：

python

运行

def send_custom_photo_message(openid): """给用户发送客服消息（图片）""" # 1. 获取access_token try: access_token = get_wechat_access_token() except Exception as e: print(f"获取access_token失败：{str(e)}") return # 2. 从数据库查询对应学生照片（先获取学生ID，再查询照片） conn = pymysql.connect(**DB_CONFIG) cursor = conn.cursor(pymysql.cursors.DictCursor) # 查询学生ID sql = "SELECT student_id FROM parent_wechat_bind WHERE wechat_openid = %s" cursor.execute(sql, (openid,)) parent_data = cursor.fetchone() if not parent_data: print(f"未查询到该用户绑定信息：OpenID={openid}") cursor.close() conn.close() return student_id = parent_data["student_id"] # 查询照片信息（优先获取最新照片） sql = "SELECT photo_url FROM student_photo WHERE student_id = %s ORDER BY photo_update_time DESC LIMIT 1" cursor.execute(sql, (student_id,)) photo_data = cursor.fetchone() cursor.close() conn.close() if not photo_data: # 发送无照片提示 send_custom_text_message(openid, access_token, "暂未查询到您孩子的学习照片，请稍后再试~") return photo_url = photo_data["photo_url"] # 3. 调用客服消息接口发送图片（需先将图片上传到微信临时素材库，获取media_id） media_id = upload_photo_to_wechat(access_token, photo_url) if not media_id: send_custom_text_message(openid, access_token, "照片上传失败，请稍后再试~") return # 客服消息接口地址 url = f"https://api.weixin.qq.com/cgi-bin/message/custom/send?access_token={access_token}" payload = { "touser": openid, "msgtype": "image", "image": { "media_id": media_id } } try: response = requests.post(url, json=payload) result = json.loads(response.text) if result.get("errcode") == 0: print(f"客服消息（图片）推送成功：OpenID={openid}") else: print(f"客服消息推送失败：OpenID={openid}，错误信息={result}") except Exception as e: print(f"客服消息推送异常：OpenID={openid}，异常信息={str(e)}") def upload_photo_to_wechat(access_token, photo_url): """将照片上传到微信临时素材库，获取media_id（有效期3天）""" # 先下载照片到本地（或直接从云存储读取二进制流） try: photo_response = requests.get(photo_url) photo_content = photo_response.content except Exception as e: print(f"下载照片失败：{photo_url}，异常={str(e)}") return None # 上传到微信临时素材库 url = f"https://api.weixin.qq.com/cgi-bin/media/upload?access_token={access_token}&type=image" files = { "media": ("photo.jpg", photo_content, "image/jpeg") } try: response = requests.post(url, files=files) result = json.loads(response.text) if "media_id" in result: return result["media_id"] else: print(f"照片上传到微信素材库失败：{result}") return None except Exception as e: print(f"上传素材异常：{str(e)}") return None def send_custom_text_message(openid, access_token, content): """发送文本类型客服消息""" url = f"https://api.weixin.qq.com/cgi-bin/message/custom/send?access_token={access_token}" payload = { "touser": openid, "msgtype": "text", "text": { "content": content } } requests.post(url, json=payload)

步骤 4：关键注意事项

• 触发限制：仅在用户主动发送消息后 48 小时内可推送客服消息，超出时间无法主动推送。
• 素材有效期：临时素材库的media_id有效期为 3 天，若需长期使用，可上传到永久素材库（需符合微信永久素材规范）。
• 图片大小：单张图片不超过 5M，支持 JPG/PNG 格式，建议压缩后再上传。

三、核心技术架构总结

1. 数据层：MySQL 存储「家长 - 微信 OpenID - 学生」绑定关系、学生照片信息；Redis 缓存access_token，提升接口调用效率。
2. 存储层：阿里云 OSS / 腾讯云 COS 存储孩子学习照片，提供稳定的 HTTPS 访问链接。
3. 应用层：后端服务（Python/Java/PHP）实现凭证获取、消息推送、数据库操作、回调处理等核心逻辑。
4. 接口层：调用微信开放平台接口（access_token接口、模板消息接口、客服消息接口、素材上传接口）完成消息推送。

四、无开发能力的替代方案

若你暂无技术团队，可直接使用微信第三方家校服务平台（如「校宝在线」「微校通」「腾讯智慧校园」），操作流程如下：

1. 将认证后的服务号授权给第三方平台；
2. 批量导入家长手机号、学生信息（提前完成微信绑定）；
3. 上传孩子照片并关联对应学生；
4. 使用平台现成的 “家校通知” 功能，批量推送照片通知，无需自行编写代码。

总结

1. 优先选择「模板消息推送」实现批量精准通知，核心是获取access_token、分批次调用微信模板消息接口；
2. 关键技术点：OpenID 与手机号绑定、access_token缓存、分批次推送避限流、照片隐私验证；
3. 客服消息仅适用于用户主动查询场景，需依赖消息回调和素材上传；
4. 无开发能力可通过第三方家校平台快速落地，降低技术门槛。