news 2026/10/21 22:49:25

微信公众号天气推送实战:从定时任务到模板消息的完整开发指南

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
微信公众号天气推送实战:从定时任务到模板消息的完整开发指南

简介:这份资源面向具备少量编程基础、希望为伴侣定制微信每日推送的开发者,围绕微信公众号定时发送天气与纪念日提醒这一场景,提供可落地的实现方案。资源包共5个文件,以py主程序、yml自动化工作流配置、txt配置与说明、md文档为主,压缩包仅5KB,轻量易读,适合直接部署到GitHub实现每日自动线上运行。内容涵盖最新完整版、简化版、星座单独版、星座加天气版以及PHP版本等多套成品与消息模板,并附有配置说明,读者可据此替换天行、和风等API密钥,快速搭建属于自己的推送服务。目前已有7695人学习下载,教程结合作者亲身经历讲解操作流程,成品可直接运行,也便于二次修改消息模板与推送时间,适合想用低成本方式制造惊喜、又希望理解自动化推送原理的读者参考实践。

1. 一个天气推送小项目,为什么值得动手拆一遍

有个朋友在群里吐槽,说每天早上女朋友都要问他“今天冷不冷、要不要带伞”,问得他头皮发麻。后来他花了一个周末,用微信公众号做了个每日天气推送,早上七点自动把当天的天气、温度区间、穿衣建议发到对方微信上。这事听起来像是个哄人开心的小玩具,但真动手做一遍,你会发现它把公众号开发里几个最核心的环节全串起来了:服务器验证、access_token 管理、模板消息、定时任务、第三方天气 API 调用。对于想入门微信公众号开发、又不想啃官方文档里那些枯燥示例的开发者来说,这是一个体量刚好、闭环完整的练手项目。它不需要你懂前端框架,也不需要买服务器集群,一台能跑 Python 的机器加上一个已经认证的公众号,就能把整条链路跑通。下面我就按自己拆项目的习惯,把这个天气推送系统从架构到代码、从部署到改造,一层层讲清楚。

2. 项目架构与目录结构:一条消息从定时触发到微信送达的完整链路

2.1 整体架构:四个角色各干什么活

这个项目本质上是一个“定时拉取 + 主动推送”的服务,不涉及用户交互,所以不需要处理复杂的消息回复逻辑。整条链路里只有四个角色在干活:

第一个是定时调度器,负责每天到点触发任务。常见做法是用 Linux 的 crontab,或者用 Python 的 APScheduler 在进程内起一个调度线程。crontab 更简单,适合部署在服务器上;APScheduler 更灵活,适合本地调试。

第二个是天气数据获取模块,负责调用第三方天气 API,拿到目标城市的实时天气和未来预报。国内常用的有和风天气、心知天气、高德开放平台,选哪个看你的预算和调用量。免费额度一般够个人用。

第三个是微信消息推送模块,负责把天气数据填充到模板消息里,调用微信的模板消息接口发送给指定用户。这里有两个关键点:access_token 的缓存与刷新,以及模板消息的 openid 获取。

第四个是配置与日志模块,负责管理 API Key、公众号 AppID、AppSecret、目标用户 openid、城市编码这些敏感信息,同时记录每次推送的结果,方便排查问题。

这四个角色之间的关系可以用一句话概括:调度器触发任务,任务先拿天气数据,再拿 access_token,最后组装模板消息发出去。整个流程没有数据库依赖,配置用 JSON 或 YAML 文件管理即可,轻量到可以塞进一个单文件脚本里。

2.2 目录结构:我习惯这样组织代码

虽然项目小,但目录结构不能乱。我一般会按下面这种方式组织,方便后面加城市、加推送渠道:

weather-push/ ├── config/ │ ├── settings.yaml # 公众号配置、天气 API Key、目标用户 openid │ └── cities.yaml # 城市名称与天气 API 城市编码的映射 ├── core/ │ ├── weather.py # 天气 API 调用与数据解析 │ ├── wechat.py # access_token 管理与模板消息发送 │ └── scheduler.py # 定时任务入口 ├── utils/ │ ├── logger.py # 日志配置 │ └── cache.py # access_token 本地缓存读写 ├── main.py # 单次推送入口,方便手动测试 └── requirements.txt

config/settings.yaml里放的是不常变的东西,比如公众号的 AppID 和 AppSecret、天气 API 的 Key、女朋友的 openid。cities.yaml单独拆出来是因为后面你可能想给不同的人推不同城市,或者一个人关注多个城市,拆开后扩展起来不用动主逻辑。

core/weather.py只负责一件事:传入城市编码,返回一个结构化的天气字典。core/wechat.py负责 access_token 的获取、缓存、刷新,以及模板消息的组装和发送。core/scheduler.py是调度入口,决定什么时候调用前两个模块。

utils/cache.py用来存 access_token,因为微信的 access_token 有 7200 秒有效期,而且每天获取次数有限,不能每次推送都重新拿。常见做法是存到本地文件里,记录过期时间,过期了再刷新。

main.py是一个手动触发入口,调试的时候直接python main.py就能发一条,不用等定时任务。

2.3 技术选型:为什么用 Python 而不是 Node.js

这个项目用 Python 还是 Node.js 都能做,我选 Python 的原因很实际:天气 API 的返回是 JSON,Python 的requests+json处理起来很顺手;模板消息的组装就是字符串格式化,Python 的 f-string 写起来直观;定时任务用APScheduler或者直接 crontab 调 Python 脚本,部署成本低。Node.js 当然也能做,但如果你只是想在服务器上跑一个小脚本,Python 的依赖更少,调试也更直接。

依赖方面,核心就三个包:requests用来发 HTTP 请求,PyYAML用来读配置文件,APScheduler用来做进程内定时。如果你用 crontab,APScheduler都可以省掉。日志用 Python 自带的logging模块就够了,不需要额外装东西。

提示:公众号必须是认证的服务号或订阅号,个人订阅号没有模板消息接口权限。如果你用的是测试号,可以在微信公众平台申请测试号,测试号自带模板消息权限,适合开发阶段用。

3. 核心模块拆解:天气获取、access_token 管理与模板消息组装

3.1 天气数据获取:从 API 返回到结构化字典

天气 API 的返回通常是一大坨 JSON,直接塞进模板消息里会又长又乱。我一般会在weather.py里做一层转换,把原始数据压成模板消息需要的几个字段。下面以和风天气的免费接口为例,代码逻辑是这样的:

import requests from datetime import datetime def fetch_weather(city_code, api_key): """ 调用天气 API,返回结构化天气字典 :param city_code: 城市编码,如 "101010100" 表示北京 :param api_key: 天气 API 的 Key :return: dict,包含 date, temp_min, temp_max, text, tips """ url = "https://devapi.qweather.com/v7/weather/3d" params = { "location": city_code, "key": api_key, "lang": "zh" } resp = requests.get(url, params=params, timeout=10) data = resp.json() # 取今天的数据,索引 0 是今天,1 是明天 today = data["daily"][0] # 穿衣建议根据温度区间简单判断,也可以调 API 的 indices 接口 temp_min = int(today["tempMin"]) temp_max = int(today["tempMax"]) if temp_max <= 10: tips = "今天很冷,记得穿厚外套" elif temp_max <= 20: tips = "今天微凉,建议穿件薄外套" else: tips = "今天温度舒适,穿件长袖就行" return { "date": today["fxDate"], "temp_min": temp_min, "temp_max": temp_max, "text": today["textDay"], "tips": tips }

这段代码里,city_code是天气 API 定义的城市编码,不是城市名。和风天气的城市编码可以通过它的城市查询接口拿到,也可以在网上搜到常用城市的编码表。api_key放在配置文件里,不要硬编码在代码里。

timeout=10是必须加的,不然网络抖动的时候脚本会卡死。data["daily"][0]取的是今天的数据,如果你想把明天的天气也推过去,可以再取data["daily"][1]。

穿衣建议这里我用了一个简单的温度判断,没有调专门的指数接口。如果你想要更准确的建议,和风天气有indices接口,可以拿到穿衣、紫外线、洗车等指数,但那个接口需要额外调用一次,会增加一点复杂度。对于哄女朋友这个场景,简单的温度判断已经够用了。

3.2 access_token 管理:缓存与刷新的正确姿势

access_token 是调用微信接口的通行证,每次调用模板消息接口都要带上。微信官方给的 access_token 有效期是 7200 秒,而且获取次数有限制,所以不能每次推送都重新获取。我一般会在wechat.py里做一个简单的文件缓存:

import json import time import os import requests TOKEN_CACHE_FILE = "token_cache.json" def get_access_token(app_id, app_secret): """ 获取 access_token,优先从本地缓存读取 :param app_id: 公众号 AppID :param app_secret: 公众号 AppSecret :return: str,access_token """ # 先读缓存 if os.path.exists(TOKEN_CACHE_FILE): with open(TOKEN_CACHE_FILE, "r") as f: cache = json.load(f) # 提前 300 秒刷新,避免边界过期 if cache["expires_at"] > time.time() + 300: return cache["access_token"] # 缓存过期或不存在,重新获取 url = "https://api.weixin.qq.com/cgi-bin/token" params = { "grant_type": "client_credential", "appid": app_id, "secret": app_secret } resp = requests.get(url, params=params, timeout=10) data = resp.json() if "access_token" not in data: raise Exception(f"获取 access_token 失败: {data}") token = data["access_token"] expires_at = time.time() + data["expires_in"] # 写回缓存 with open(TOKEN_CACHE_FILE, "w") as f: json.dump({ "access_token": token, "expires_at": expires_at }, f) return token

这里有几个细节值得说。第一,缓存文件里存了expires_at,也就是过期时间戳,而不是存“获取时间 + 7200 秒”,这样读的时候直接比较时间戳就行,不用再算一次。第二,我提前 300 秒刷新,也就是有效期还剩 5 分钟的时候就重新获取,避免刚好在过期边界上调用失败。第三,如果获取失败,直接把微信返回的错误信息抛出来,方便排查是 AppID 错了还是 IP 白名单没配。

注意:微信公众平台的后台需要配置 IP 白名单,只有白名单里的 IP 才能调用 access_token 接口。如果你把脚本部署在服务器上,记得把服务器的公网 IP 加进去。本地调试的时候,把本机公网 IP 加进去,或者用测试号,测试号不需要配 IP 白名单。

3.3 模板消息组装与发送:字段映射与跳转链接

模板消息的发送需要三个东西:接收者的 openid、模板 ID、模板数据。模板数据是一个嵌套的 JSON,每个字段对应模板里定义的关键词。下面是一个发送函数的实现:

def send_weather_message(access_token, openid, template_id, weather): """ 发送天气模板消息 :param access_token: 微信 access_token :param openid: 接收者的 openid :param template_id: 模板消息 ID :param weather: 天气字典,来自 fetch_weather """ url = f"https://api.weixin.qq.com/cgi-bin/message/template/send?access_token={access_token}" # 模板数据,key 对应模板里的关键词,value 是具体内容 data = { "touser": openid, "template_id": template_id, "url": "https://www.example.com/weather", # 点击消息跳转的链接,没有就留空 "data": { "date": { "value": weather["date"], "color": "#173177" }, "city": { "value": "北京", "color": "#173177" }, "temp": { "value": f"{weather['temp_min']}°C ~ {weather['temp_max']}°C", "color": "#FF0000" }, "weather": { "value": weather["text"], "color": "#173177" }, "tips": { "value": weather["tips"], "color": "#173177" } } } resp = requests.post(url, json=data, timeout=10) result = resp.json() if result.get("errcode") != 0: raise Exception(f"发送模板消息失败: {result}") return result

模板消息的data里,每个 key 必须和你在公众号后台创建的模板里的关键词一一对应。比如你的模板里写的是{{date.DATA}},那代码里就要有date这个 key。color字段可以控制文字颜色,天气温度我用红色突出一下,其他用默认的深蓝色。

url字段是点击消息后跳转的链接,如果你没有自己的网页,可以留空或者填一个天气网页的链接。注意这个链接必须是已经备案的域名,不然微信会拦截。

发送失败的时候,微信会返回errcode和errmsg,常见的错误有40001(access_token 无效)、40003(openid 错误)、40037(模板 ID 无效)。把这些错误信息打到日志里,排查起来会快很多。

3.4 定时调度:crontab 与 APScheduler 怎么选

定时调度有两种做法。如果你部署在 Linux 服务器上,直接用 crontab 最省事:

# 每天早上 7 点执行一次推送脚本 0 7 * * * /usr/bin/python3 /home/user/weather-push/main.py >> /home/user/weather-push/push.log 2>&1

这行 crontab 的意思是每天 7:00 执行main.py,标准输出和错误输出都追加到push.log里。/usr/bin/python3要换成你服务器上 Python 的实际路径,可以用which python3查。

如果你在本地调试,或者不想依赖 crontab,可以用 APScheduler 在进程内起一个调度器:

from apscheduler.schedulers.blocking import BlockingScheduler from main import push_weather if __name__ == "__main__": scheduler = BlockingScheduler() # 每天 7:00 执行 push_weather scheduler.add_job(push_weather, "cron", hour=7, minute=0) scheduler.start()

BlockingScheduler会阻塞当前进程,适合单独跑一个调度脚本。add_job里的cron表达式和 crontab 类似,hour=7, minute=0就是每天 7 点整。如果你想让脚本在后台跑,可以用nohup python scheduler.py &或者写成 systemd 服务。

两种方式各有优劣:crontab 更稳定,服务器重启后任务还在;APScheduler 更灵活,可以在代码里动态加任务,但进程挂了任务就没了。我一般生产环境用 crontab,本地开发用 APScheduler。

4. 避坑与排查:那些让我半夜爬起来改代码的坑

4.1 模板消息发送成功但用户没收到

现象是接口返回errcode: 0,日志里显示发送成功,但微信里就是没有消息。原因通常是用户没有关注公众号,或者用户把公众号的消息免打扰打开了。模板消息只能发给已经关注公众号的用户,如果对方取关了,接口会返回43004(require subscribe)。解决方法是先确认对方是否还在关注,如果取关了,需要重新关注才能收到。另外,如果对方开了消息免打扰,消息会进“服务通知”而不是聊天列表,不是没收到,是没弹窗。

4.2 access_token 频繁刷新导致接口调用失败

现象是脚本跑了一段时间后,突然开始报40001或42001(access_token 过期)。原因可能是缓存文件被多个进程同时读写,或者服务器时间不准导致过期判断出错。解决方法是确保只有一个进程在刷新 token,如果用了多进程或分布式部署,需要把 token 缓存放到 Redis 里,用分布式锁控制刷新。另外,服务器时间要同步 NTP,不然time.time()算出来的过期时间会偏。

4.3 天气 API 返回的城市编码不对

现象是天气数据一直是同一个城市的,或者返回204(无数据)。原因是城市编码填错了,或者用了城市名而不是编码。和风天气的城市编码是101010100这种格式,不是beijing也不是北京。解决方法是先用城市查询接口确认编码,再填到配置文件里。如果你想让用户自己选城市,可以在公众号里加一个回复城市名的逻辑,但那就需要处理消息回复了,复杂度会上去。

4.4 定时任务在服务器上不执行

现象是本地跑python main.py没问题,但 crontab 到点了没反应。原因通常是 crontab 的环境变量和登录 shell 不一样,python3的路径找不到,或者工作目录不对。解决方法是在 crontab 里用绝对路径,并且在脚本开头cd到项目目录。另外,crontab 的输出默认会发邮件,如果服务器没配邮件,日志就丢了,所以一定要重定向到文件里。

4.5 模板消息里的换行和特殊字符被吞了

现象是模板消息里想换行,用了\n但发出来是一行。原因是微信模板消息的value字段不支持\n换行,需要用\n的转义形式或者用多个字段拼接。解决方法是把长文本拆成多个模板字段,或者用空格代替换行。如果你非要在一条消息里换行,可以试试在value里写\n,但不同版本的微信客户端表现不一样,不太稳定。

5. 二次开发与改造:从单城市推送到多用户多城市

5.1 扩展方向:多用户、多城市、多推送渠道

这个项目跑通之后,最自然的改造方向是支持多用户和多城市。现在的代码是单用户单城市,配置写死在settings.yaml里。如果你想给多个人推,可以把用户信息存到 SQLite 或 MySQL 里,每个用户关联一个城市编码和一个 openid。推送的时候遍历用户列表,逐个发送。

另一个方向是加推送渠道。除了微信公众号,还可以加邮件、钉钉机器人、飞书机器人。做法是把发送逻辑抽象成一个接口,微信、邮件、钉钉各自实现一个发送类,调度器根据配置决定用哪个渠道。这样改完之后,你的天气推送就不只服务于微信了。

还有一个方向是加交互。现在的项目是单向推送,用户不能主动查询。你可以在公众号后台配置服务器地址,接收用户消息,当用户回复“天气”时,返回当前天气。这需要处理微信的消息加解密和签名验证,复杂度比单向推送高不少,但做出来之后体验会好很多。

5.2 改造示例:用 SQLite 存用户配置

下面是一个简单的用户表设计和查询逻辑,把原来的单用户配置改成多用户:

import sqlite3 def init_db(): """初始化用户表""" conn = sqlite3.connect("users.db") cursor = conn.cursor() cursor.execute(""" CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, openid TEXT NOT NULL UNIQUE, city_code TEXT NOT NULL, city_name TEXT NOT NULL, enabled INTEGER DEFAULT 1 ) """) conn.commit() return conn def get_active_users(conn): """获取所有启用的用户""" cursor = conn.cursor() cursor.execute("SELECT openid, city_code, city_name FROM users WHERE enabled = 1") return cursor.fetchall()

users表里存了 openid、城市编码、城市名称和一个启用开关。推送的时候先get_active_users拿到所有用户,然后循环调用fetch_weather和send_weather_message。这样加用户只需要往表里插一条记录,不用改代码。

5.3 验证方法:怎么确认改造没把原来的功能搞坏

改完之后,我一般会按这个顺序验证:先手动跑main.py,确认单用户推送正常;再往数据库里插一个测试用户,跑一次批量推送,确认两个用户都收到了;然后把其中一个用户的enabled改成 0,再跑一次,确认只有启用的用户收到。最后把 crontab 的时间改成 5 分钟后,等一次自动触发,确认定时任务也没问题。

日志里要记录每个用户的发送结果,成功和失败都记,失败的要带上errcode和errmsg。这样出问题的时候,直接看日志就知道是哪个用户、哪个环节挂了。

5.4 一个我踩过的坑:别在循环里反复获取 access_token

批量推送的时候,最容易犯的错误是在用户循环里调用get_access_token。虽然缓存能挡住大部分请求,但如果缓存刚好过期,第一个用户触发了刷新,第二个用户又触发一次,就会导致 token 被刷新两次,前一个 token 失效。正确的做法是在循环外面获取一次 access_token,然后传给每个用户的发送函数。这个坑我踩过一次,当时日志里一堆40001,排查了半天才发现是循环里重复获取 token 导致的。

从那以后我每次写批量推送,都强制把 access_token 的获取提到循环外面,并且在日志里打一行“token 获取时间”和“token 过期时间”,方便确认是不是缓存生效了。希望这个习惯也能帮到你。

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

版权声明: 本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若内容造成侵权/违法违规/事实不符,请联系邮箱:809451989@qq.com进行投诉反馈,一经查实,立即删除!
网站建设 2026/10/21 22:47:20

考拉兹猜想 - 3

再看一遍&#xff0c;在 为偶数时&#xff0c;在 为奇数时&#xff0c;两端减去 &#xff0c;两边减去1&#xff0c;左右交换&#xff0c;此时右边为虚数单位 &#xff0c;等价关系说明左边也一样&#xff0c;两边同时乘以 &#xff0c;此处&#xff0c;正如&#xff0c;左右两…

作者头像 李华
网站建设 2026/10/21 22:47:17

FreeSWITCH ICE 协商慢?RTP 地址获取延迟排查与参数调优实战

1. 问题背景与核心症结拆解FreeSWITCH 作为一款成熟的软交换平台&#xff0c;在语音通信领域承担着信令控制与媒体转发的核心角色。当它运行在 NAT 环境或者复杂网络拓扑中时&#xff0c;媒体协商环节会引入 ICE&#xff08;Interactive Connectivity Establishment&#xff09…

作者头像 李华
网站建设 2026/10/21 22:47:17

Unity LeanTween 常用 API 速查与实战避坑指南

1. 为什么一个动画库的API值得单独拎出来聊 做Unity客户端开发的人&#xff0c;几乎都绕不开补间动画这件事。UI弹窗要缩放、按钮要回弹、血条要平滑过渡、道具要飘动、相机要跟随&#xff0c;这些需求如果全部手写协程或者Update里插值&#xff0c;代码很快就会变成一锅粥。Le…

作者头像 李华
网站建设 2026/10/21 22:46:56

Java网络编程进阶指南:从BIO、NIO到Netty实战聊天室

1. 为什么网络编程是Java进阶路上绕不开的一道坎刚入行那会儿&#xff0c;我对网络编程的理解基本停留在“Socket就是用来聊天的”这种程度。直到有一次需要做一个设备数据采集网关&#xff0c;要求同时对接几十台终端设备&#xff0c;每台设备通过TCP长连接上报心跳和数据&…

作者头像 李华
网站建设 2026/10/21 22:46:15

Hive SQL巧解连续登录:日期减行号实现连续区间分组

做过用户增长或数据运营的朋友&#xff0c;应该都接到过这类需求&#xff1a;报表里要统计每个用户最长连续登录了多少天&#xff1b;或者用户断签几天后又回来&#xff0c;后续连续登录的序号得重新从1算起。这俩需求在Hive SQL里其实是同一类问题&#xff0c;都属于“连续区间…

作者头像 李华
网站建设 2026/10/21 22:46:14

知识工作流插件体系:从捕获到输出的分层搭建与避坑实践

你打开浏览器&#xff0c;看到收藏夹里躺着几十个"效率神器"&#xff0c;编辑器里装着上百个插件&#xff0c;某笔记软件里七八个主题脚本来回切换&#xff0c;可一周下来真正用到的可能还是那几个。我维护一套叫"knowledge-work-plugins"的插件清单已经有…

作者头像 李华