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

简介这份资源面向具备少量编程基础、希望为伴侣定制微信每日推送的开发者围绕微信公众号定时发送天气与纪念日提醒这一场景提供可落地的实现方案。资源包共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.txtconfig/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 的返回是 JSONPython 的requestsjson处理起来很顺手模板消息的组装就是字符串格式化Python 的 f-string 写起来直观定时任务用APScheduler或者直接 crontab 调 Python 脚本部署成本低。Node.js 当然也能做但如果你只是想在服务器上跑一个小脚本Python 的依赖更少调试也更直接。依赖方面核心就三个包requests用来发 HTTP 请求PyYAML用来读配置文件APScheduler用来做进程内定时。如果你用 crontabAPScheduler都可以省掉。日志用 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, paramsparams, timeout10) 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放在配置文件里不要硬编码在代码里。timeout10是必须加的不然网络抖动的时候脚本会卡死。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: straccess_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, paramsparams, timeout10) 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 fhttps://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, jsondata, timeout10) result resp.json() if result.get(errcode) ! 0: raise Exception(f发送模板消息失败: {result}) return result模板消息的data里每个 key 必须和你在公众号后台创建的模板里的关键词一一对应。比如你的模板里写的是{{date.DATA}}那代码里就要有date这个 key。color字段可以控制文字颜色天气温度我用红色突出一下其他用默认的深蓝色。url字段是点击消息后跳转的链接如果你没有自己的网页可以留空或者填一个天气网页的链接。注意这个链接必须是已经备案的域名不然微信会拦截。发送失败的时候微信会返回errcode和errmsg常见的错误有40001access_token 无效、40003openid 错误、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 21这行 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, hour7, minute0) scheduler.start()BlockingScheduler会阻塞当前进程适合单独跑一个调度脚本。add_job里的cron表达式和 crontab 类似hour7, minute0就是每天 7 点整。如果你想让脚本在后台跑可以用nohup python scheduler.py 或者写成 systemd 服务。两种方式各有优劣crontab 更稳定服务器重启后任务还在APScheduler 更灵活可以在代码里动态加任务但进程挂了任务就没了。我一般生产环境用 crontab本地开发用 APScheduler。4. 避坑与排查那些让我半夜爬起来改代码的坑4.1 模板消息发送成功但用户没收到现象是接口返回errcode: 0日志里显示发送成功但微信里就是没有消息。原因通常是用户没有关注公众号或者用户把公众号的消息免打扰打开了。模板消息只能发给已经关注公众号的用户如果对方取关了接口会返回43004require subscribe。解决方法是先确认对方是否还在关注如果取关了需要重新关注才能收到。另外如果对方开了消息免打扰消息会进“服务通知”而不是聊天列表不是没收到是没弹窗。4.2 access_token 频繁刷新导致接口调用失败现象是脚本跑了一段时间后突然开始报40001或42001access_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 过期时间”方便确认是不是缓存生效了。希望这个习惯也能帮到你。本文还有配套的精品资源点击获取