“天上掉下一只小日和”如果只看字面很容易以为是一句玩笑话。把它当成项目代号以后这句玩笑反而概括了整条技术链路“小日和”负责每天从天气服务里取数据“天上掉下来”对应的是定时任务和 HTTP 请求“送到群里”则是 webhook 通知。这篇文章就用这个代号从零搭建一个可运行的天气日报机器人每天按计划抓取指定城市天气按阈值判断是否需要提醒再通过钉钉机器人推送到群并且用 SQLite 做幂等去重避免同一天重复打扰。这类小工具很适合作为 Python 自动化项目的入门工程。它覆盖了配置管理、网络请求、数据解析、定时调度、持久化、通知发送和异常排查几乎每一环都是后端开发会反复遇到的基础问题。读完以后你不仅有了一个能每天自动播报天气的“小日和”还会知道如何把类似的定时任务改造成监控告警、数据日报、定时备份通知等场景。1. 先想清楚“小日和”要解决什么问题1.1 “日和”和“天上掉下来”分别对应什么“日和”来自日语常指天气、晴天。把“小日和”当作项目代号时它就是一个和天气相关的助手。“天上掉下一只小日和”这个标题可以做两层理解“天上掉下来”代表数据的来源。它每天从公网天气服务获取天气预报数据就像从云端把天气信息取下来。“小日和”代表提醒动作。它根据一定的判断规则决定今天要不要告诉群成员带伞、注意高温或低温并通过消息机器人把结果送到群里。确定这个定位后需求就很清楚了。这个工具的核心能力有四块每天定时执行一次任务。从天气 API 获取目标城市当天的天气数据。根据天气代码、降水概率、温度生成一段人类可读的提醒内容。通过钉钉机器人发送到群并保证同一天同一内容不会重复推送。这个范围刚好适合用一个小型 Python 项目来实现不需要引入消息队列、容器编排或微服务框架。越是简单清晰的项目越适合用来打磨定时任务、异常处理和部署细节这些基本功。1.2 技术选型与关键依赖围绕上述需求选型可以尽量轻量。下面是“小日和”使用的核心组件组件作用选择理由Open-Meteo天气数据源无需 API Key支持每日天气聚合数据返回标准 JSONrequests发起 HTTP 请求Python 生态最常用的 HTTP 客户端适合快速对接 REST APIAPScheduler定时任务调度支持 Cron 表达式、时区配置、任务合并和错过任务补偿SQLite本地去重与历史记录单文件数据库无需额外服务适合轻量场景钉钉机器人消息通知通过 webhook 发送 Markdown 消息配置简单适合群消息场景这套组合在个人项目和小型团队工具里很常见。不要一上来就上 Celery、Redis、Kafka 这类重型组件等确实出现多实例、任务堆积、消息丢失等问题时再演进也不迟。项目里还会用到python-dotenv来加载.env文件以及 Python 标准库logging输出日志。整体依赖非常少便于在普通服务器上安装和运行。2. 环境准备与项目结构2.1 准备 Python 环境和依赖文件“小日和”建议使用 Python 3.10 及以上版本。新项目最好用虚拟环境隔离依赖不要直接装到系统 Python 里。在项目根目录执行python3 -m venv venv source venv/bin/activate python -m pip install --upgrade pip然后创建requirements.txtrequests2.31.0 APScheduler3.10.4 python-dotenv1.0.1安装依赖pip install -r requirements.txt这三个库分别负责 HTTP 请求、定时调度和环境变量读取。版本号在实际安装前最好去 PyPI 确认最新稳定版尤其是后续使用新版本时API 可能有细微调整。2.2 项目目录结构建议按模块拆分文件而不是把所有逻辑写进一个main.py。拆分后每个模块职责单一后续加功能或排查问题会轻松很多。xiao_rihe/ ├── app/ │ ├── __init__.py │ ├── config.py │ ├── weather.py │ ├── message.py │ ├── notifier.py │ ├── storage.py │ └── scheduler.py ├── data/ ├── logs/ ├── .env.example ├── requirements.txt ├── run.py └── README.mdconfig.py负责读取环境变量并生成配置对象。weather.py负责调用天气 API、解析返回结果。message.py负责根据天气数据生成提醒文案。notifier.py负责发送钉钉消息。storage.py负责 SQLite 初始化和去重查询。scheduler.py负责把上述模块串成定时任务。run.py是命令行入口支持一次运行和定时运行。data目录存放 SQLite 数据库文件logs目录存放日志。这两个目录建议提前建好也可以在代码里用Path.mkdir(parentsTrue, exist_okTrue)自动创建。2.3 配置项速查配置通过环境变量注入本地开发时用.env文件管理。先复制模板cp .env.example .env.env.example内容如下CITY_NAME北京 LATITUDE39.9042 LONGITUDE116.4074 TIMEZONEAsia/Shanghai SCHEDULE_HOUR8 SCHEDULE_MINUTE30 WEBHOOK_URL WEBHOOK_SECRET RAIN_PROBABILITY_THRESHOLD40 COLD_TEMPERATURE_THRESHOLD10 HOT_TEMPERATURE_THRESHOLD33 DATA_DIR./data LOG_LEVELINFO各项参数含义环境变量默认值说明CITY_NAME北京城市名称仅用于消息文案展示LATITUDE39.9042维度Open-Meteo 使用 WGS84 坐标LONGITUDE116.4074经度TIMEZONEAsia/Shanghai天气数据和调度任务使用的时区SCHEDULE_HOUR8每天执行的小时SCHEDULE_MINUTE30每天执行的分钟WEBHOOK_URL空钉钉机器人 webhook 地址WEBHOOK_SECRET空钉钉机器人加签密钥未启用则留空RAIN_PROBABILITY_THRESHOLD40降水概率达到多少时提醒带伞COLD_TEMPERATURE_THRESHOLD10最低温度低于多少时提醒保暖HOT_TEMPERATURE_THRESHOLD33最高温度高于多少时提醒防暑DATA_DIR./dataSQLite 数据库存放目录LOG_LEVELINFO日志级别WEBHOOK_URL和WEBHOOK_SECRET不要提交到 Git。.env文件要加入.gitignore只保留.env.example作为配置模板。3. 实现核心模块3.1 配置加载为什么选择环境变量配置写在代码里最大的问题是改参数要改代码而且 API 密钥容易泄露。环境变量则可以在不修改源码的情况下调整运行参数也更容易在 systemd、Docker 等环境中统一注入。app/config.py使用dataclass定义配置对象并通过os.getenv读取环境变量import os from dataclasses import dataclass from pathlib import Path from dotenv import load_dotenv load_dotenv() dataclass class Config: city_name: str latitude: float longitude: float timezone: str schedule_hour: int schedule_minute: int webhook_url: str webhook_secret: str rain_probability_threshold: int cold_temperature_threshold: float hot_temperature_threshold: float data_dir: Path log_level: str property def db_path(self) - Path: return self.data_dir / xiao_rihe.db property def log_path(self) - Path: return Path(logs) / xiao_rihe.log def load_config() - Config: return Config( city_nameos.getenv(CITY_NAME, 北京), latitudefloat(os.getenv(LATITUDE, 39.9042)), longitudefloat(os.getenv(LONGITUDE, 116.4074)), timezoneos.getenv(TIMEZONE, Asia/Shanghai), schedule_hourint(os.getenv(SCHEDULE_HOUR, 8)), schedule_minuteint(os.getenv(SCHEDULE_MINUTE, 30)), webhook_urlos.getenv(WEBHOOK_URL, ), webhook_secretos.getenv(WEBHOOK_SECRET, ), rain_probability_thresholdint(os.getenv(RAIN_PROBABILITY_THRESHOLD, 40)), cold_temperature_thresholdfloat(os.getenv(COLD_TEMPERATURE_THRESHOLD, 10)), hot_temperature_thresholdfloat(os.getenv(HOT_TEMPERATURE_THRESHOLD, 33)), data_dirPath(os.getenv(DATA_DIR, ./data)), log_levelos.getenv(LOG_LEVEL, INFO), )这里有几个容易出错的地方load_dotenv()默认读取当前工作目录下的.env所以启动命令必须在项目根目录执行。纬度经度是浮点数环境变量里写错格式会在启动阶段直接抛ValueError这比运行一半才报错更友好。阈值参数用int或float视业务需要而定。降水概率通常是整数百分比温度一般有小数所以二者类型不同。3.2 天气数据获取与 WMO 天气码解析Open-Meteo 的免费接口适合学习和个人项目。请求地址是https://api.open-meteo.com/v1/forecast关键查询参数包括latitude、longitude、daily、timezone和forecast_days。其中daily可以一次性获取多个每日聚合字段。app/weather.py实现请求和解析import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry WMO_WEATHER_CODES { 0: 晴, 1: 大部晴朗, 2: 多云, 3: 阴, 45: 雾, 48: 雾凇, 51: 毛毛雨, 53: 小毛毛雨, 55: 大毛毛雨, 56: 冻毛毛雨, 57: 强冻毛毛雨, 61: 小雨, 63: 中雨, 65: 大雨, 66: 冻雨, 67: 强冻雨, 71: 小雪, 73: 中雪, 75: 大雪, 77: 雪粒, 80: 阵雨, 81: 强阵雨, 82: 猛烈阵雨, 85: 阵雪, 86: 强阵雪, 95: 雷暴, 96: 雷暴伴小冰雹, 99: 雷暴伴大冰雹, } def _build_session() - requests.Session: retries Retry( total3, backoff_factor1, status_forcelist[500, 502, 503, 504], allowed_methods[GET], ) session requests.Session() session.mount(https://, HTTPAdapter(max_retriesretries)) return session def fetch_daily_forecast(latitude: float, longitude: float, timezone: str) - dict: url https://api.open-meteo.com/v1/forecast params { latitude: latitude, longitude: longitude, daily: ( weather_code, temperature_2m_max, temperature_2m_min, precipitation_probability_max ), timezone: timezone, forecast_days: 1, } session _build_session() response session.get(url, paramsparams, timeout10) response.raise_for_status() data response.json() return parse_daily_forecast(data) def parse_daily_forecast(data: dict) - dict: daily data[daily] weather_code daily[weather_code][0] return { date: daily[time][0], weather_code: weather_code, weather_desc: WMO_WEATHER_CODES.get(weather_code, f未知天气代码 {weather_code}), temperature_max: daily[temperature_2m_max][0], temperature_min: daily[temperature_2m_min][0], precipitation_probability_max: daily[precipitation_probability_max][0], }为什么要配置Retry外部 API 不可能永远稳定偶尔会出现 500 或网络抖动。重试三次并采用指数退避可以明显减少偶发失败。但要注意重试只对 GET 请求安全如果是写操作盲目重试可能产生重复数据。timeout10也必须设置。如果不设置超时requests 在极端情况下会一直等下去导致定时任务卡住不再执行。3.3 生成提醒消息拿到天气数据后不要直接把 JSON 丢到群里而是生成一段易读的 Markdown。app/message.py负责把数据转成文案from app.config import Config def build_weather_message(forecast: dict, config: Config) - str: precipitation forecast[precipitation_probability_max] temp_min forecast[temperature_min] temp_max forecast[temperature_max] lines [ ### 小日和天气日报, f城市{config.city_name}, f日期{forecast[date]}, f天气{forecast[weather_desc]}, f气温{temp_min:.0f}°C ~ {temp_max:.0f}°C, f降水概率{precipitation}%, ] tips [] if precipitation config.rain_probability_threshold: tips.append(出门记得带伞) if temp_min config.cold_temperature_threshold: tips.append(早晚偏冷注意保暖) if temp_max config.hot_temperature_threshold: tips.append(午后较热注意防暑) if tips: lines.append() lines.append(提醒 .join(tips)) return \n\n.join(lines)温度格式化用:.0f是为了避免输出30.0°C这种看起来不够自然的文案。如果城市不在中国单位可以根据需要改成华氏度Open-Meteo 也支持temperature_unitfahrenheit参数。这里的设计判断是提醒规则要可配置而不是写死在代码里。降水概率阈值、高温阈值、低温阈值都来自Config以后调整规则只需要改环境变量不需要重新部署代码。3.4 通知模块钉钉机器人 webhook 与加签钉钉群机器人通过 webhook 接收消息。最简单的调试方式是不启用任何安全设置只填WEBHOOK_URL。如果群有外部成员或要求更严格通常使用“加签”方式。app/notifier.py实现发送逻辑import base64 import hashlib import hmac import time import urllib.parse import requests def _build_signed_url(webhook_url: str, secret: str) - str: timestamp str(round(time.time() * 1000)) string_to_sign f{timestamp}\n{secret} hmac_code hmac.new( secret.encode(utf-8), string_to_sign.encode(utf-8), digestmodhashlib.sha256, ).digest() sign urllib.parse.quote_plus(base64.b64encode(hmac_code)) separator if ? in webhook_url else ? return f{webhook_url}{separator}timestamp{timestamp}sign{sign} def send_dingtalk_message(webhook_url: str, secret: str, title: str, text: str) - None: if not webhook_url: raise ValueError(webhook_url 不能为空) url _build_signed_url(webhook_url, secret) payload { msgtype: markdown, markdown: { title: title, text: text, }, } response requests.post(url, jsonpayload, timeout10) response.raise_for_status() result response.json() if result.get(errcode) ! 0: raise RuntimeError(f钉钉返回错误{result.get(errmsg)})钉钉的加签流程是使用当前毫秒时间戳和 secret 拼接再做 HMAC-SHA256 签名然后 URL 编码后拼到 webhook 地址后面。如果只配了WEBHOOK_URL而没配WEBHOOK_SECRET_build_signed_url会原样返回地址兼容无加签的机器人。这里的错误处理很关键。钉钉接口即使 HTTP 返回 200业务上仍可能返回错误码例如errcode310000表示关键字不匹配或签名错误。所以不能只看raise_for_status()还要检查响应体里的errcode。3.5 SQLite 去重与记录定时任务重复执行的可能性很高手动运行一次自动调度再跑一次或者网络超时后重试一次。如果每次都发消息群成员会收到一堆重复内容。SQLite 在这里的作用是幂等去重。app/storage.py实现存储和去重import sqlite3 from pathlib import Path class WeatherStore: def __init__(self, db_path: Path): self.db_path db_path self.db_path.parent.mkdir(parentsTrue, exist_okTrue) self._init_db() def _connect(self) - sqlite3.Connection: conn sqlite3.connect(self.db_path) return conn def _init_db(self) - None: with self._connect() as conn: conn.execute( CREATE TABLE IF NOT EXISTS notify_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, send_date TEXT NOT NULL, message_hash TEXT NOT NULL, city TEXT, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, UNIQUE (send_date, message_hash) ) ) def is_duplicate(self, send_date: str, message_hash: str) - bool: with self._connect() as conn: row conn.execute( SELECT 1 FROM notify_log WHERE send_date ? AND message_hash ?, (send_date, message_hash), ).fetchone() return row is not None def save(self, send_date: str, message_hash: str, city: str) - None: with self._connect() as conn: conn.execute( INSERT OR IGNORE INTO notify_log(send_date, message_hash, city) VALUES (?, ?, ?) , (send_date, message_hash, city), )为什么使用(send_date, message_hash)联合唯一键“send_date”保证跨天可以再次推送“message_hash” 保证同一天内容变化后可以重新发送。如果同一天下午天气突变提醒内容变了就能通过新的哈希正常推送而不是永远被“今天已经发过”拦掉。message_hash可以直接用 Python 内置hashlib.md5生成import hashlib def compute_message_hash(text: str) - str: return hashlib.md5(text.encode(utf-8)).hexdigest()在最后组装任务的函数里先查询是否重复未发送过才调用通知并写入记录。这样即使多个调度器误跑数据库也会挡住重复消息。3.6 定时调度与启动入口APScheduler 的BlockingScheduler适合这种单进程、单任务的场景。app/scheduler.py把天气获取、消息生成、通知、存储串成完整任务import hashlib import logging from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger from app.config import Config from app.message import build_weather_message from app.notifier import send_dingtalk_message from app.storage import WeatherStore from app.weather import fetch_daily_forecast logger logging.getLogger(__name__) def run_once(config: Config, dry_run: bool False) - None: store WeatherStore(config.db_path) forecast fetch_daily_forecast( latitudeconfig.latitude, longitudeconfig.longitude, timezoneconfig.timezone, ) message build_weather_message(forecast, config) logger.info(天气数据获取成功%s, message.replace(\n, | )) message_hash hashlib.md5(message.encode(utf-8)).hexdigest() send_date forecast[date] if store.is_duplicate(send_date, message_hash): logger.info(同一天同一内容已推送跳过发送) return if dry_run: logger.info(dry-run 模式不执行发送) print(message) return send_dingtalk_message( webhook_urlconfig.webhook_url, secretconfig.webhook_secret, title小日和天气日报, textmessage, ) store.save(send_date, message_hash, config.city_name) logger.info(消息发送成功) def build_scheduler(config: Config): scheduler BlockingScheduler(timezoneconfig.timezone) trigger CronTrigger( day_of_week*, hourconfig.schedule_hour, minuteconfig.schedule_minute, timezoneconfig.timezone, ) scheduler.add_job( run_once, triggertrigger, args[config], iddaily_weather_job, misfire_grace_time3600, coalesceTrue, max_instances1, ) return scheduler这里几个 APScheduler 参数需要理解misfire_grace_time3600任务预期执行时间错过 3600 秒内仍允许补跑适合服务器休眠或短暂停机恢复的场景。coalesceTrue如果进程长时间离线导致任务多次错过只合并执行最后一次避免连续触发一堆无用任务。max_instances1同一任务未结束时不允许再次触发防止上一次天气请求卡住导致下一次并发执行。run.py是入口脚本支持手动执行一次和启动调度器import argparse import logging from app.config import load_config from app.scheduler import build_scheduler, run_once def setup_logging(level: str) - None: logging.basicConfig( levellevel.upper(), format%(asctime)s [%(levelname)s] %(name)s: %(message)s, ) def main() - None: parser argparse.ArgumentParser(description小日和天气日报) parser.add_argument(--once, actionstore_true, help只执行一次) parser.add_argument(--dry-run, actionstore_true, help只打印消息不发送) args parser.parse_args() config load_config() setup_logging(config.log_level) if args.once: run_once(config, dry_runargs.dry_run) return scheduler build_scheduler(config) logger logging.getLogger(__name__) logger.info(调度器启动时区 %s首次任务由 Cron 触发, config.timezone) scheduler.start() if __name__ __main__: main()--dry-run是最有价值的调试参数。它在不发送任何真实消息的情况下把最终会推到群里的内容打印出来方便调整文案和阈值。4. 运行与验证4.1 先用 dry-run 验证消息内容配置好.env后先不急着发真实消息。执行python run.py --once --dry-run预期输出类似### 小日和天气日报 城市北京 日期2025-05-20 天气多云 气温18°C ~ 27°C 降水概率30% 提醒午后较热注意防暑这个步骤验证的是整条链路中不依赖外部通知渠道的部分环境变量是否正确加载、天气 API 是否能请求成功、JSON 解析是否正确、消息模板是否合理、阈值规则是否生效。如果--dry-run里出现未知天气代码说明 WMO 映射表缺项如果城市名不对说明.env没加载成功或路径不对。4.2 用真实 webhook 发送一次确认文案无误后先手动测试钉钉 webhook 是否可用。用 curl 发送一条最简单的文本curl -X POST https://oapi.dingtalk.com/robot/send?access_token你的token \ -H Content-Type: application/json \ -d {msgtype:text,text:{content:小日和天气测试}}如果使用的是自定义关键词安全设置文本中必须包含关键词比如“天气”。如果返回{errcode:0,errmsg:ok}说明 webhook 可用。然后关闭--dry-run执行一次真实发送python run.py --once日志中应当出现2025-05-20 08:30:01 [INFO] app.scheduler: 天气数据获取成功... | 北京 | ... 2025-05-20 08:30:02 [INFO] app.scheduler: 消息发送成功群里会收到一条 Markdown 格式的天气日报。注意钉钉机器人发送的 Markdown 支持有限复杂表格不一定展示完整所以这里尽量用标题、列表、加粗这些基础语法。4.3 验证数据库和日志发送成功后查询 SQLite 数据库sqlite3 data/xiao_rihe.db SELECT id, send_date, city, created_at FROM notify_log;预期结果1|2025-05-20|北京|2025-05-20 08:30:02再执行一次python run.py --once日志应该显示同一天同一内容已推送跳过发送这说明去重逻辑生效。如果修改了温度阈值导致消息文案变化message_hash会改变同一天仍会发送一条新内容。这个行为是合理的因为规则变了才需要重新提醒。验证完成后可以把data/xiao_rihe.db和logs/加入.gitignore避免本地运行产生的数据污染仓库。注意不要只验证程序能启动还要验证输入、输出、异常分支和日志是否符合预期。特别是定时任务场景必须确认“该任务今天已经发过”时能被正确拦下。5. 常见问题排查5.1 天气请求失败或超时现象日志中出现HTTPConnectionPool(hostapi.open-meteo.com, port443): Max retries exceeded或Timeout异常。可能原因服务器没有外网访问权限。DNS 解析失败。Open-Meteo 服务端返回 5xx。经纬度参数格式错误。检查方式先用 curl 直接请求接口curl https://api.open-meteo.com/v1/forecast?latitude39.9042longitude116.4074dailyweather_code,temperature_2m_maxforecast_days1timezoneAsia/Shanghai看到 JSON 说明网络链路没问题问题大概率在代码参数或重试策略。如果 curl 也失败先检查服务器出网策略、DNS 和防火墙。处理建议保持代码中的Retry和timeout配置。在日志里打印完整的请求 URL方便复现。如果 API 偶尔超时可以增加backoff_factor让重试间隔更长。如果经常失败可以考虑在目标地区部署一个中转服务把天气数据缓存下来。5.2 钉钉 webhook 返回错误码现象HTTP 返回 200但日志抛出钉钉返回错误...。群里没有收到消息。常见错误码和排查方向错误码常见原因处理方式300001token 不存在或已被重置到钉钉机器人管理页面重新复制 webhook310000关键字不匹配或签名错误检查自定义关键词是否在消息文本中检查加签 URL 拼接40035参数格式错误检查msgtype和markdown字段结构41001access_token 缺失确认 webhook 地址完整如果启用了自定义关键词“小日和天气日报”里的“天气”就是关键词所以默认模板通常能通过。如果启用了加签建议先输出_build_signed_url()生成的结果确认其中包含timestamp和sign参数。处理建议先用钉钉机器人管理页的“测试”按钮确认 webhook 本身正常。把response.json()完整记录到日志不要只记录errmsg。不要在日志中打印完整 webhook 地址避免 token 泄露。可以只打印access_token的前 6 位和后 4 位。5.3 定时任务不触发或触发时间不对现象手动执行python run.py --once正常。调度器启动后到点没有消息。可能原因进程没有保持运行终端关闭后任务就停止了。调度器时区与预期不一致。服务器系统休眠或者有容器回收机制杀掉了进程。Cron 表达式与配置参数不一致。检查方式在job.next_run_time输出到日志确认下一次执行时间job scheduler.get_job(daily_weather_job) logger.info(下次执行时间%s, job.next_run_time)在服务器上执行date确认系统时区。处理建议本地开发时用python run.py保持前台运行。生产环境优先使用 systemd timer 或系统 crontab而不是只靠应用内调度器。如果任务偶尔没有执行把misfire_grace_time调大并开启coalesceTrue。5.4 同一天收到重复消息现象群消息出现多条相同内容。数据库notify_log中没有对应记录。可能原因DATA_DIR指向了不同目录导致每次运行都使用新的数据库。多实例同时运行例如既用调度器又用 systemd timer两个进程同时执行。消息内容每毫秒不同导致message_hash变化。检查方式打印数据库路径确认进程实际使用哪个文件。查询数据库中相同日期的记录sqlite3 data/xiao_rihe.db SELECT send_date, message_hash, city, created_at FROM notify_log ORDER BY id DESC LIMIT 10;处理建议单实例运行避免多种调度方式叠加。数据库路径使用绝对路径。生产环境可以在 systemdEnvironmentFile中显式配置DATA_DIR/opt/xiao_rihe/data。关键可以展示send_date和message_hash用联合唯一键兜底。5.5 WMO 天气代码不识别现象消息显示“未知天气代码 1001”。数据解析本身没有报错。原因Open-Meteo 返回了映射表中不存在的天气代码。API 字段拼写错误取到了错误数组。检查方式先在parse_daily_forecast里直接打印daily[weather_code]的原始值。对照 WMO 官方天气代码表完善字典。处理建议字典使用get(code, f未知天气代码 {code})兜底避免程序崩溃。单独写一个单元测试覆盖常见代码def test_weather_code_mapping(): assert WMO_WEATHER_CODES[0] 晴 assert WMO_WEATHER_CODES[95] 雷暴注意外部 API 的返回结构可能升级解析代码要保留原始 JSON 的日志记录。断点调试时打印完整data比只打印解析后字段更管用。6. 生产环境部署、最佳实践与扩展方向6.1 用 systemd 定时任务代替应用内调度应用内调度适合个人电脑和临时脚本但生产服务器上更推荐用 systemd timer 或 crontab 来触发。原因是系统级任务管理器更可靠进程崩溃后能自动拉起日志也更好管理。如果使用 systemd服务文件/etc/systemd/system/xiao-rihe.service如下[Unit] DescriptionXiao Rihe Weather Reminder Afternetwork-online.target [Service] Typesimple WorkingDirectory/opt/xiao_rihe EnvironmentFile/etc/xiao_rihe.env ExecStart/opt/xiao_rihe/venv/bin/python run.py --once Restarton-failure RestartSec10 Userrihe定时器文件/etc/systemd/system/xiao-rihe.timer[Unit] DescriptionRun Xiao Rihe every day at 08:30 [Timer] OnCalendar*-*-* 08:30:00 Persistenttrue [Install] WantedBytimers.target启用定时器sudo systemctl daemon-reload sudo systemctl enable xiao-rihe.timer sudo systemctl start xiao-rihe.timerPersistenttrue的作用是如果系统在计划执行时间处于关机状态下次开机后补跑一次。这样就不需要依赖应用内的misfire_grace_time。这种方案下应用内调度器不再启动run.py只执行--once分支。两种调度方式只能选一种否则会出现重复消息。6.2 生产环境需要补的保障学习环境跑通不代表生产环境安全。下面这些保障项可以直接整理进发布检查清单检查项具体要求配置外置webhook、经纬度、阈值都通过环境变量注入不写死在代码密钥安全.env和 webhook 地址不进 Git钉钉机器人定期重置密钥日志持久化配置TimedRotatingFileHandler或交给 systemd journal 管理