☰
微信公众号天气提醒:模板消息推送与access_token缓存实践
2026/10/29 15:38:31 网站建设 项目流程

简介:面向具备少量编程基础、希望为伴侣制作微信公众号每日定时推送的读者,这份资源围绕天气通知与纪念日提醒场景,提供一套可参考的自动推送实现方案,重点解决消息模板配置、天气接口接入以及每日自动运行等问题。压缩包共5个文件,以Python主程序、GitHub Actions工作流配置、txt配置与说明、md文档为主,整体约5KB,体量轻巧但结构完整,便于理解推送逻辑与线上运行方式。目前已有7698人学习下载,说明该玩法受到不少读者关注。资源中包含消息模板、接口参数配置说明与定时运行配置,读者可据此了解天气数据获取、纪念日文案组织及公众号消息下发的基本链路,并按自身需求替换天气API与推送内容;同时提供简化版、星座方向等成品与源码参考,适合想快速复现或二次调整的读者。

1. 从一条早上七点二十的推送说起:公众号天气提醒到底难在哪

早上七点二十,手机在床头震一下,点开是一条天气通知:今天多云转小雨,18 到 24 度,出门带伞。很多人第一反应是“这不就是调两个接口的事”,真动手做微信公众号每日消息推送才发现,接口调用只占两成工作量,剩下八成全是 token 缓存、时区对齐、失败重试和模板变量对不上的问题。

它适合谁?想给身边人做每日问候的开发者,手上有台常开的小机器,或者会用免费的定时任务;也适合把公众号当个人消息通道,用来推运维提醒、日报摘要的人。不太合适的是想群发营销的——模板消息是点对点推给已关注用户,本质上是一条私有通道,不是广播喇叭。

下面按真实落地顺序拆:先把链路打通,再把天气变成一句人话,然后解决“每天准点”这件事,最后是那些会让你静默失败的坑。

2. 打通三段链路:测试号、模板消息与天气接口的关键参数

整个系统只有三段:拿数据、拼消息、发出去。三段各自有一个“卡住就全盘不动”的关键参数——appID/appSecret 与 openid、template_id 与变量名、access_token。先把这三个东西拿齐,后面写代码才有意义。

2.1 测试号路径:一次拿齐 appID、openid 和 template_id

正式服务号的模板消息有资质门槛,个人开发者最省事的路径是公众平台的测试号,不用认证,模板内容还能自己编排。进测试号管理页扫一下,页面上直接给出 appID 和 appSecret,同时有一个二维码,用要接收消息的微信号扫一次,用户列表里就会冒出一串 openid——这串字符是后面 touser 字段唯一合法的取值,抄错一位就是 40003。

紧接着建模板。测试号的好处是模板正文随便写,我一般写成这样的固定骨架:

{{date.DATA}} {{city.DATA}} 天气:{{weather.DATA}} 温度:{{temp.DATA}} {{tips.DATA}}

保存后会拿到一串 template_id。把模板理解成填空题:花括号里{{xxx.DATA}}的 xxx 是变量名,接口里 data 对象的键必须和它逐字对应,大小写、下划线、拼写都不能差。这一条后面还会反复出现,因为它是最高频的翻车点。

三步动作固定下来:记录 appID/appSecret 写进环境变量;扫码关注后复制 openid;建模板拿 template_id。三个值凑齐,链路的“身份”部分就完成了。

2.2 天气数据源选型:三个免费接口的字段对比

天气接口的选择比想象中重要,因为它决定了你后面要写多少清洗代码。常见做法是在几个免费额度够用的服务里挑一个,重点看三件事:返回字段够不够拼文案、字段是不是字符串、错误码走 HTTP 状态还是走 body。

数据源典型请求形态关键字段需要额外注意
和风天气GET /v7/weather/now,参数 location + keynow.temp、now.text、now.feelsLike、now.windScale业务码在 body 的 code 里,值为字符串 "200"
心知天气GET /v3/weather/now.json,参数 key + locationresults[0].now.temperature、now.text返回是数组嵌套,要先取 results[0]
高德开放平台GET /v3/weather/weatherInfo,参数 city + keylives[0].temperature、weather、windpower用 adcode 而不是城市名,字段全为字符串

我个人偏向第一种,理由是字段命名直白、更新时间字段齐全,拼“体感温度”这种文案时不用再算。城市定位用城市 ID 而不是中文名,避免“杭州”被解析成别的行政区——这种玄学问题排查起来非常费时间。

请求这一层没什么花活,但有两个参数必须显式写:

# fetch_weather.py import os import requests from dotenv import load_dotenv load_dotenv() QWEATHER_KEY = os.environ["QWEATHER_KEY"] CITY_ID = os.environ.get("CITY_ID", "101210101") # 城市 ID,换城市只改这一个值 def fetch_now(city_id: str = CITY_ID) -> dict: url = "https://devapi.qweather.com/v7/weather/now" params = { "location": city_id, "key": QWEATHER_KEY, "gzip": "y", # 开启压缩,返回体小一半 } resp = requests.get(url, params=params, timeout=8) # 不设 timeout 会把定时任务挂死 resp.raise_for_status() data = resp.json() if data.get("code") != "200": # 注意是字符串,不是整数 200 raise RuntimeError(f"qweather code={data.get('code')}") return data["now"]

这段的逻辑是“先判传输、再判业务”。raise_for_status()管 HTTP 层,code != "200"管业务层,两层分开的好处是日志里能立刻看出是网络断了还是 key 失效了。timeout=8不是可选项,定时任务里一次无响应的请求会把整个进程吊在那里,第二天的推送就跟着一起没了。

2.3 access_token:两小时有效期,还带每日获取上限

access_token 是调用模板消息接口的通行证,有效期约两小时,且每天能获取的次数有限。最糟糕的写法是每次推送前现取一个——一天一次看着没问题,但你调试的时候会连着跑十几次,很容易撞上频率限制,报出来的错误码看着像“权限不足”,实际是取太勤了。

正确姿势是本地缓存,提前过期。我一般的做法是存一个 JSON 文件,记录 token 和过期时间戳,读取时留 600 秒余量:

# token_store.py import json, os, time import requests TOKEN_FILE = "/opt/weather/.token.json" # 放在持久化目录,别放 /tmp APPID = os.environ["WX_APPID"] SECRET = os.environ["WX_SECRET"] def get_token(force: bool = False) -> str: if not force and os.path.exists(TOKEN_FILE): with open(TOKEN_FILE) as f: cache = json.load(f) if cache["expire_at"] - 600 > time.time(): # 提前 10 分钟视为过期 return cache["token"] url = "https://api.weixin.qq.com/cgi-bin/token" r = requests.get(url, params={ "grant_type": "client_credential", "appid": APPID, "secret": SECRET, }, timeout=8).json() if "access_token" not in r: raise RuntimeError(f"token error: {r}") # errcode/errmsg 原样抛,别吞 cache = {"token": r["access_token"], "expire_at": time.time() + r["expires_in"]} with open(TOKEN_FILE, "w") as f: json.dump(cache, f) os.chmod(TOKEN_FILE, 0o600) # 权限收紧,里面是明文凭证 return cache["token"]

参数说明:expires_in是服务端给的剩余秒数,直接拿来做基准比写死 7200 稳;force=True留给 40001 错误时强制刷新用;文件权限 600 是基本习惯,凭证落盘不加权限等于把钥匙放在门口。多进程并发的情况要再加文件锁,单机单任务一般不用。

三段链路到这里就通了:身份有了、数据有了、通行证有了。接下来才是真正决定“推出来好不好看”的部分。

3. 把天气 JSON 变成一句人话:字段取舍、文案规则与模板拼装

接口返回的 JSON 有十几个字段,全塞进模板里没人看。真正有价值的是“今天要不要带伞、穿几件、风大不大”,所以这一章的重点是取舍和映射,而不是把所有字段都搬上去。

3.1 只取五个字段,剩下的都是噪音

一个每日推送的模板长度有限,我一般只保留:日期与星期、城市、天气现象、温度区间、一句建议。对应的原始字段是obsTime、text、temp、feelsLike、windScale、humidity。其中湿度和风力不进正文,只参与“建议”的生成判断。

有一个细节值得单独说:接口给的是当前时刻的温度,不是当天的最低温最高温。如果只调 now 接口,你拼出来的“18 到 24 度”其实是编的。想给区间,就得再调一次逐日预报接口(daily),从返回里取tempMax和tempMin。这一步很多人省掉,结果推了一周才发现温度永远是单值,看着像坏了一样。

另外obsTime是观测时间,不一定等于你推送的时间。凌晨接口更新滞后时,你会把昨天的数据推出去。稳妥做法是拿obsTime和当前时间比一下,差距超过三小时就在日志里记一条警告,别硬推。

3.2 温度区间映射穿衣建议:一张规则表替掉一坨 if

新手常写一长串 if-elif 判断温度,改一次规则要动一遍代码。我习惯把规则抽成一张表,代码只负责查表,这样调文案不用改逻辑:

最低温区间建议文案附加判断
≥ 26℃短袖短裤,注意防晒补水湿度 > 80 时追加“闷热”
20 ~ 25℃一件长袖刚好,早晚加薄外套温差 > 8℃ 时强调“早晚凉”
12 ~ 19℃外套加内搭,别只穿单件风力 ≥ 5 级追加“防风”
5 ~ 11℃厚外套或薄羽绒,围巾备上有降水时追加“路滑”
< 5℃羽绒加保暖内衣,手套别忘体感温度低于实际时按体感算
# compose.py RULES = [ (26, "短袖短裤,注意防晒补水"), (20, "一件长袖刚好,早晚加件薄外套"), (12, "外套加内搭,别只穿单件"), (5, "厚外套或薄羽绒,围巾备上"), (-99, "羽绒加保暖内衣,手套别忘"), ] def pick_advice(t_min: int, diff: int, wind_scale: int, humidity: int) -> str: base = next(text for floor, text in RULES if t_min >= floor) extra = [] if diff >= 8: extra.append("早晚温差大") if wind_scale >= 5: extra.append("风有点冲") if humidity >= 80 and t_min >= 26: extra.append("体感闷") return ",".join([base] + extra)

RULES是降序排列的,next()拿到第一个满足条件的就停,顺序错了文案就会串。diff是最高温减最低温,wind_scale是风力等级,两个都从接口直接取,不要自己估算。这段的价值在于:以后想改文案,只动元组里的字符串,不需要碰判断逻辑。

3.3 拼装模板消息 body:变量对齐是唯一硬约束

模板消息的请求体结构很固定,难点全在 data 的键名要和模板里的变量名一模一样:

# push.py import requests from token_store import get_token def send_template(openid: str, template_id: str, data: dict, retry: bool = True) -> dict: token = get_token() body = { "touser": openid, "template_id": template_id, # 模板里 {{key.DATA}} 的 key 必须与这里逐字一致 "data": {k: {"value": str(v)} for k, v in data.items()}, } resp = requests.post( "https://api.weixin.qq.com/cgi-bin/message/template/send", params={"access_token": token}, json=body, timeout=8, ).json() # 40001 = token 无效/过期,清缓存重发一次就够,不要循环重试 if retry and resp.get("errcode") == 40001: get_token(force=True) return send_template(openid, template_id, data, retry=False) return resp

参数说明:touser是 openid,不是微信号也不是昵称;data里每个值都要包一层{"value": ...},直接传字符串接口会报参数错误;str(v)是保险,温度这种字段有时是数字有时是字符串,统一转掉省事。retry=False防止 token 一直失效时打成死循环。

拼装完成后,判断成功不能只看有没有抛异常,要看返回体的errcode。为 0 才算真推出去,这一点和第 5 章的排查直接相关。

4. 定时投递:cron 时区、重试与日志的三处落地细节

代码能跑通只是第一步,“每天七点二十准时到”是另一件事。定时这一层出问题最隐蔽,因为脚本本身没报错,只是没在正确的时间跑。

4.1 crontab 写法与容器时区

服务器上的 crontab 是最省事的方案,一行搞定:

# 每天 07:20 推送,日志按天切分 TZ=Asia/Shanghai 20 7 * * * cd /opt/weather && /usr/bin/python3 push.py >> /var/log/weather/push-$(date +\%F).log 2>&1

两个参数值得说明。TZ=Asia/Shanghai写在 crontab 顶部,是因为多数云主机默认 UTC,不写这一行你的七点二十会变成下午三点二十——这种“时区错位”是定时任务里最经典的翻车现场。日志文件名里的\%必须转义,crontab 里%有特殊含义(表示换行),不转义整行任务会被截断。

如果跑在容器里,时区要从两头设:镜像里装 tzdata,运行时传TZ=Asia/Shanghai环境变量,再挂载/etc/localtime。只设一头的情况我见过不止一次,容器里date是对的,Python 的datetime.now()却还是 UTC,因为解释器读的是另一套配置。

4.2 不用 cron 的自循环方案

没有服务器权限、或者只想在本地挂着的场景,可以让脚本自己循环。好处是时区由代码控制,坏处是进程挂了就没了,得配合进程守护:

# daemon.py import time, datetime, traceback import pytz from push import run_once # 封装好的单次推送 TZ = pytz.timezone("Asia/Shanghai") while True: now = datetime.datetime.now(TZ) target = now.replace(hour=7, minute=20, second=0, microsecond=0) if now >= target: target += datetime.timedelta(days=1) # 今天已过,等明天 wait = (target - now).total_seconds() time.sleep(wait) try: run_once() except Exception: traceback.print_exc() # 单次失败不退出循环,明天继续

关键点是try/except把单次推送的异常包住。不包的话,某天接口抽风抛了个异常,整个进程直接退出,第二天的推送就永远不会来了。pytz换成zoneinfo也可以,Python 3.9 之后标准库就够用。

4.3 失败重试与日志字段:给明天的自己留证据

推送失败太正常了,网络抖动、接口限流、token 过期,都会碰到。策略是“有限重试 + 结构化日志”,重试两次、间隔递增即可,别做无上限重试,那只会把限流撞得更狠。

日志我要求至少包含这几个字段,缺一个后面排查就得靠猜:

字段用途
时间戳(带时区)判断是不是时区错位
城市 ID确认拉的是哪个城市的数据
天气业务码区分是接口问题还是自己代码问题
token 是否命中缓存判断是不是取 token 太频繁
errcode / errmsg接口层返回的原始错误
耗时毫秒异常慢的时候能看出是网络问题
import logging, time logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", ) def run_once(): t0 = time.time() now = fetch_now() logging.info("weather ok city=%s code=200 temp=%s", CITY_ID, now["temp"]) result = send_template(OPENID, TEMPLATE_ID, build_data(now)) logging.info("push done errcode=%s errmsg=%s cost=%dms", result.get("errcode"), result.get("errmsg"), int((time.time() - t0) * 1000)) if result.get("errcode") != 0: raise RuntimeError(f"push failed: {result}")

errcode != 0主动抛异常,是为了让上层循环或监控能感知到失败;如果只打日志不抛,脚本每天“成功”退出,你永远不知道消息没发出去。

5. 避坑与排查:五个让推送静默失败的原因

这一节全是我自己踩过的。共同特征是不报错、不崩、日志看着还挺正常,但手机就是不响。

现象一:本地跑得好好的,一上服务器就没动静。原因基本是环境变量没带过去——.env文件没跟着部署,或者 cron 的环境变量和登录 shell 的完全不是一套。解决:在 cron 里显式cd到项目目录,并且用绝对路径调 Python;启动时先打印一行os.environ.get("WX_APPID")是否存在,缺了就在日志里大喊一句,别让它静悄悄地跑。

现象二:日志显示推送成功,手机上一条没收到。先查 errcode,多数是 43004(接收者未关注)或 openid 抄错。测试号换微信号扫过码之后,openid 是会变的,旧值不会失效但也不指向新用户。解决:把 openid 和关注列表一起维护,扫码后立刻更新配置;排查时先手动调一次发送接口,看返回体原文。

现象三:某天开始连续报 40001,重试也没用。原因通常是别处也在取同一个 appID 的 token,新 token 一生成,你缓存的那个就作废了。解决:同一个 appID 只保留一个取 token 的入口,多台机器共享时把缓存放进同一个存储,或者干脆只让一台机器负责推送。这类问题最像玄学,其实是凭证被顶掉了。

现象四:消息在半夜两点推出来。时区。服务器 UTC、容器 UTC、cron 没写 TZ,三者任一都能造成这个结果。解决:判据是日志时间戳和实际推送时间对不上;统一在 crontab 顶部写TZ=Asia/Shanghai,容器额外传环境变量,Python 侧用带时区的 datetime 打印日志,一眼就能看出偏差。

现象五:模板改了一个字,之后推出来全是空白。{{temp.DATA}}被改成了{{temperature.DATA}},你代码里还在传temp,接口不报错,那个位置就是空。解决:把模板变量名和代码里的 data 键抽成一份常量表,改模板时同步改表;发送前做一次键集合比对,缺键直接拦下来不推。

6. 从能跑到敢托付:token 缓存、错误码与一次回归验证

跑到第三个月,我给自己加了一条规矩:任何改动上线前,必须先手动跑一次“干推”——把消息发给自己,确认收到、确认内容对。这个习惯救过我两次,一次是模板变量改名,一次是城市 ID 被改成了别的区。

进阶方向其实就三块。一是 token 缓存从文件升级成带锁的共享存储,多实例部署时不会互相顶掉;二是错误码分层处理,40001 刷新重试、45009 降频等待、43004 直接告警不再重试,别把可恢复和不可恢复的错误混在一个except里;三是加一条“发送前自检”,把模板键、openid 非空、温度数值范围一起校验,任何一项不过就中止并记录,宁可当天不发,也别发一条格式错乱的消息。

一个最小的自检函数长这样:

REQUIRED_KEYS = {"date", "city", "weather", "temp", "tips"} def precheck(data: dict, openid: str) -> None: if not openid or len(openid) < 20: raise ValueError("openid 为空或格式异常") missing = REQUIRED_KEYS - set(data.keys()) if missing: raise ValueError(f"模板变量缺失: {missing}") if not data.get("temp"): raise ValueError("温度为���,接口可能返回空数据")

REQUIRED_KEYS要和模板里的变量名一一对应,模板一动就同步改这里。看着是重复劳动,但它把“推出去才发现是空白”变成了“推之前就拦下来”,出问题的成本差一个量级。

验证方法也很朴素:连续跑七天,每天看日志里的 errcode 和实际到达时间,允许的偏差不超过两分钟;中间故意把 key 改错一次,确认告警能触达你本人。一套流程走完,这个推送才算是能托付的。

从那以后我每次改模板或者换城市,都会先跑一遍干推、核对一遍键集合,再让它进定时任务——这个动作花不了两分钟,但能省掉一整个早上的排查。希望帮到你。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询