用C语言手搓一个带文件存储的通讯录(动态扩容+增删改查全实现)
2026/6/7 4:03:27
在进行技术开发前,需先完成以下准备工作,确保后续开发顺畅:
AppID(应用唯一标识)、AppSecret(应用密钥,需保密,不可明文暴露在前端)。id、parent_mobile(家长手机号)、wechat_openid(微信唯一标识)、student_id(学生 ID)、create_time(绑定时间))。student_id、student_name(学生姓名)、photo_url(照片云存储地址)、photo_update_time(照片更新时间))。根据需求场景,优先推荐「模板消息推送」(支持批量精准推送,不限用户触发,适合定期发送学习照片),其次是「客服消息推送」(适用于用户主动查询场景),以下分别详细讲解技术实现。
access_token(所有微信接口的通用凭证);plaintext
【XX学校/机构】您的孩子{{studentName.DATA}}近期学习照片已更新,点击下方链接即可查看高清照片: {{photoUrl.DATA}} 温馨提示:照片仅用于家校沟通,请勿随意转发。template_id,后续接口调用需使用)。access_token微信所有接口调用均需携带access_token,有效期 2 小时(7200 秒),需实现缓存机制(避免频繁调用接口导致限流)。
https://api.weixin.qq.com/cgi-bin/token| 参数名 | 说明 |
|---|---|
| grant_type | 固定值:client_credential |
| appid | 服务号的 AppID |
| secret | 服务号的 AppSecret |
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}")https://api.weixin.qq.com/cgi-bin/message/template/sendContent-Type: application/jsonjson
{ "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
运行
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()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"))客服消息支持直接发送图片(单张≤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)media_id有效期为 3 天,若需长期使用,可上传到永久素材库(需符合微信永久素材规范)。access_token,提升接口调用效率。access_token接口、模板消息接口、客服消息接口、素材上传接口)完成消息推送。若你暂无技术团队,可直接使用微信第三方家校服务平台(如「校宝在线」「微校通」「腾讯智慧校园」),操作流程如下:
access_token、分批次调用微信模板消息接口;access_token缓存、分批次推送避限流、照片隐私验证;