很多时候一个看起来很简单的念头往往藏着完整的工程需求。比如“每天都能给在意的人发一句祝福”听起来只是复制粘贴的小事但真正落地时你会面对定时任务怎么跑、消息怎么送、祝福语怎么管、同一天不能重复发、失败之后怎么排查等一系列问题。这篇文章就把这个念头变成一个可以运行的 Python 小项目我把它叫做“幸福提醒助手”。它的设计初衷就是标题里那句话你的幸福就是我最大的幸福。文章会从需求拆解开始逐步讲解环境准备、模块设计、完整代码、运行验证、常见报错和工程建议。无论你是 Python 初学者还是想做一个能用于真实场景的自动化小工具都可以跟着这篇文章从零搭一套定时推送服务。代码结构上不只写单文件 Demo而是按配置、内容、发送渠道、调度、存储拆成模块方便以后扩展和维护。1. 背景与项目愿景1.1 一句话如何变成一个技术项目“你的幸福就是我最大的幸福”是一个很有温度的表达但放到开发语境里它需要被翻译成一条条可执行的需求定时希望每天固定时间触发比如早上 09:30。内容不能每天重复同一句话最好能随机轮换遇到纪念日或节日还能发送特定祝福。渠道能自动把消息发给对方常见方案包括邮件、钉钉机器人、企业微信机器人等。状态记录同一天不能重复发送程序重启后也知道今天已经发过了。可观测每次发送成功或失败要有日志方便排查问题。把这些需求组装起来就是一个典型的自动化任务项目。它的价值不只是“发祝福”本身而是把定时调度、第三方接口调用、数据持久化、异常处理这几个后端高频技能串在了一起。学会这个项目以后做每日天气推送、定时报表通知、监控告警提醒本质上都是同一套思路。1.2 这个工具解决什么问题第一个问题是解决“靠记忆坚持”的不可靠性。人总有忙忘的时候而定时任务不会忘。第二个问题是解决“消息内容管理”的麻烦。如果只是手动发送很难坚持每天换不同文案如果把文案放到 CSV 文件里就可以提前批量准备还能按日期命中特殊文案。第三个问题是让“发送这件事”可回滚、可追溯。每一条消息是否发送成功都有记录即便漏发或重复也能根据日志快速定位。从更宽的角度看这类工具在日常生活中可以延伸出很多变体每天给项目组发站会提醒、每天给服务器发送健康检查日报、每周末给家庭群推送出行天气。所以不要觉得“幸福提醒助手”只是一个玩具它的架构完全可以复用到实际工作场景。1.3 技术选型说明这个项目我选择使用 Python 3而不是写一个手机 App 或网页应用。原因很直接Python 标准库自带smtplib发送邮件不需要额外安装复杂依赖。requests库调用钉钉机器人 Webhook 非常方便。APScheduler是成熟的 Python 定时任务库支持 cron 表达式、时区配置、任务持久化比手写while True sleep要可靠得多。sqlite3标准库足以保存发送记录不需要额外部署数据库服务。整个项目依赖极少运行成本低适合部署在长期开机的家用电脑、小型服务器或云主机上。模块技术方案作用定时调度APScheduler按 cron 规则触发每日任务邮件发送smtplib MIMEText通过 SMTP 发送文本邮件群机器人推送requests Webhook推送消息到钉钉群发送记录SQLite记录当天发送状态防止重复配置管理configparser 环境变量读取基础配置保护敏感信息2. 环境准备与项目结构2.1 环境要求本项目代码使用 Python 3 编写建议使用 Python 3.8 及以上版本。操作系统方面Windows、Linux、macOS 都可以运行。如果你需要在服务器上长期运行推荐使用 Linux 系统配合 nohup 或 systemd 守护进程。需要提前安装以下 Python 依赖APScheduler3.10,4.0 requests2.28,3.0安装命令pip install APScheduler requests这里对版本做一个说明APScheduler 3.x 和 4.x 的 API 变化较大本文示例基于 3.10 系列所以你安装依赖时尽量锁定大版本。如果使用最新版本请以官方文档为准调整导入路径和触发器写法。2.2 项目目录结构我建议把项目按功能拆成多个模块这样后续维护、测试、扩展都更方便。整体目录结构如下happiness-reminder/ ├── config.ini.example # 配置模板不包含真实密钥 ├── requirements.txt # Python 依赖 ├── send_history.db # 运行时自动生成SQLite 数据库 ├── happiness_reminder/ │ ├── __init__.py │ ├── config.py # 配置读取 │ ├── messages.py # 祝福语管理 │ ├── sender.py # 邮件/钉钉发送渠道 │ ├── storage.py # SQLite 存储与幂等判断 │ ├── scheduler.py # APScheduler 定时任务 │ └── main.py # 程序入口 └── messages/ └── blessings.csv # 祝福语数据这个结构不算复杂但每一层职责清晰。实际项目中如果还要加企业微信机器人只需在sender.py里增加一个类上层调用基本不用改动。2.3 编写依赖文件requirements.txt内容如下APScheduler3.10,4.0 requests2.28,3.0在项目根目录执行以下命令安装pip install -r requirements.txt如果你使用的是虚拟环境请先创建并激活虚拟环境再安装依赖。Windows 下使用venv的命令是python -m venv venv venv\Scripts\activateLinux/macOS 下则是python3 -m venv venv source venv/bin/activate3. 核心模块设计与实现3.1 配置管理为什么敏感信息要走环境变量项目里需要配置 SMTP 服务器、邮箱账号、接收者邮箱、钉钉机器人地址、发送时间等。一个常见错误是直接把授权码和 Token 写进代码仓库一旦代码被提交到公共仓库敏感信息就会泄露。这里的经验是非敏感配置如发送时间、时区、SMTP 主机地址可以放在config.ini文件中。敏感配置如 SMTP 授权码、钉钉机器人 Token优先通过环境变量注入。config.py模块负责把所有配置读取到统一的AppConfig对象中。这样可以避免在业务代码里到处读配置也方便在测试时替换配置来源。# 文件路径happiness_reminder/config.py import configparser import os from pathlib import Path BASE_DIR Path(__file__).resolve().parent.parent class AppConfig: def __init__(self, config_fileNone): self.config_file config_file or (BASE_DIR / config.ini) self._parser configparser.ConfigParser() if Path(self.config_file).exists(): self._parser.read(self.config_file, encodingutf-8) # SMTP 配置 self.smtp_host self._get(smtp, smtp_host, smtp.qq.com) self.smtp_port self._getint(smtp, smtp_port, 465) self.sender_email self._get(smtp, sender_email, ) self.receiver_email self._get(smtp, receiver_email, ) self.smtp_auth_code os.getenv(SMTP_AUTH_CODE, ) # 钉钉配置 self.webhook_url os.getenv(DINGTALK_WEBHOOK_URL, ) if not self.webhook_url: self.webhook_url self._get(dingtalk, webhook_url, ) # 调度配置 self.send_time self._get(schedule, send_time, 09:30) self.timezone self._get(schedule, timezone, Asia/Shanghai) # 消息配置 self.csv_path self._get(message, csv_path, messages/blessings.csv) if not os.path.isabs(self.csv_path): self.csv_path str(BASE_DIR / self.csv_path) def _get(self, section, key, default): try: return self._parser.get(section, key) except (configparser.NoSectionError, configparser.NoOptionError): return default def _getint(self, section, key, default0): try: return self._parser.getint(section, key) except (configparser.NoSectionError, configparser.NoOptionError, ValueError): return default对应的配置模板config.ini.example是[smtp] smtp_host smtp.qq.com smtp_port 465 sender_email your_emailqq.com receiver_email friendexample.com # 授权码不要写在这里建议通过环境变量 SMTP_AUTH_CODE 传入 [dingtalk] webhook_url https://oapi.dingtalk.com/robot/send?access_tokenyour_access_token [schedule] send_time 09:30 timezone Asia/Shanghai [message] csv_path messages/blessings.csv注意这里的config.ini.example只是模板你需要把它复制为config.ini再按实际环境修改。复制命令cp config.ini.example config.iniWindows CMD 下则是copy config.ini.example config.ini3.2 祝福语管理模块祝福语模块要解决两个问题一是内容从哪里来二是如何按日期匹配特殊文案。我采用“内置默认文案 CSV 外部数据”的方式。CSV 文件每一行有两个字段date和blessing。date为空表示这条是普通祝福语date填写MM-DD格式表示每年这个日期会优先发送这条祝福。# 文件路径happiness_reminder/messages.py import csv import random from datetime import datetime from pathlib import Path DEFAULT_BLESSINGS [ 愿你今天也有好心情记得按时吃饭。, 幸福大概就是你刚好需要我刚好出现。, 今天的阳光很好希望你也在被温柔以待。, 你要照顾好自己你的幸福是我最重要的事。, 无论今天遇到什么都有人在默默支持你。, ] def load_blessings_from_csv(csv_path): 读取 CSV 祝福语返回 (无日期祝福列表, 有日期祝福列表)。 with_date [] without_date [] path Path(csv_path) if not path.exists(): return without_date, with_date with open(path, r, encodingutf-8-sig) as f: reader csv.DictReader(f) for row in reader: date_str (row.get(date) or ).strip() blessing (row.get(blessing) or ).strip() if not blessing: continue if date_str: with_date.append((date_str, blessing)) else: without_date.append(blessing) return without_date, with_date def get_today_blessing(csv_path, extra_blessingsNone): 获取今天的祝福语优先命中日期匹配的文案。 without_date, with_date load_blessings_from_csv(csv_path) today datetime.now().strftime(%m-%d) for date_str, blessing in with_date: if date_str today: return blessing pool without_date or DEFAULT_BLESSINGS if extra_blessings: pool pool list(extra_blessings) return random.choice(pool) if pool else 这里有一个细节CSV 文件使用utf-8-sig编码读取而不是utf-8。原因是 Windows 下使用 Excel 保存 CSV 时通常会带 BOM 头用utf-8-sig可以自动去掉避免第一列列名或内容出现\ufeff字符。3.3 发送渠道SMTP 邮件邮件发送使用 Python 标准库smtplib。发送时需要把正文封装成MIMEText并设置标题、发件人、收件人。邮件标题如果包含中文需要用Header进行编码否则部分邮件客户端会显示乱码。# 文件路径happiness_reminder/sender.py import smtplib from email.header import Header from email.mime.text import MIMEText from email.utils import formataddr import requests class SmtpSender: def __init__(self, host, port, sender_email, auth_code, receiver_email): self.host host self.port port self.sender_email sender_email self.auth_code auth_code self.receiver_email receiver_email def send(self, content): msg MIMEText(content, plain, utf-8) msg[Subject] Header(今日幸福提醒, utf-8) msg[From] formataddr((str(Header(幸福提醒助手, utf-8)), self.sender_email)) msg[To] formataddr((str(Header(收信人, utf-8)), self.receiver_email)) with smtplib.SMTP_SSL(self.host, self.port, timeout10) as server: server.login(self.sender_email, self.auth_code) server.sendmail(self.sender_email, [self.receiver_email], msg.as_string()) return True使用 QQ 邮箱时SMTP_AUTH_CODE不是 QQ 密码而是邮箱设置里开启 SMTP 服务后生成的授权码。使用 163 邮箱时授权码也需要单独开启。不同邮件服务商的 SMTP 端口可能不同常见的是 465SSL和 587STARTTLS。本文示例基于 465 端口。3.4 发送渠道钉钉机器人钉钉群机器人适合发到群里的场景。它本质上就是一个 Webhook 地址通过向这个地址 POST 一段 JSON 数据完成消息推送。使用前需要先在钉钉群中添加自定义机器人并设置安全关键词。所谓安全关键词就是消息正文中必须包含指定词语否则机器人会拒绝发送。class DingTalkSender: def __init__(self, webhook_url): self.webhook_url webhook_url def send(self, content): payload { msgtype: text, text: {content: content}, } resp requests.post(self.webhook_url, jsonpayload, timeout10) resp.raise_for_status() data resp.json() if data.get(errcode) ! 0: raise RuntimeError(DingTalk send failed: {}.format(data.get(errmsg))) return data在钉钉管理后台添加机器人时安全设置可以选择“自定义关键词”。如果设置了关键词“幸福”那么推送内容里必须包含“幸福”两个字。这也是我在每条消息正文前都加上“你的幸福就是我最大的幸福”的原因之一——既点题又能满足关键词校验。当然你也可以选择加签方式这样对消息正文就没有关键词要求但需要额外传入签名计算逻辑。3.5 定时调度APSchedulerAPScheduler 是一个功能很完整的 Python 定时任务框架。它支持三种调度器BlockingScheduler、BackgroundScheduler、AsyncIOScheduler。本文项目独立运行不需要服务其他 HTTP 请求所以使用BlockingScheduler最合适。如果你要把定时任务嵌到 Django、Flask、FastAPI 等 Web 服务里则应改为BackgroundScheduler。调度器中最核心的是触发器CronTrigger。它接收 cron 表达式语法和 Linux crontab 基本一致。表达式0 30 9 * * *表示每天 09:30 执行。这里的字段依次是分、时、日、月、周。我还设置了misfire_grace_time3600这个参数表示任务原定执行时间错过之后在 3600 秒内仍然可以补执行。例如服务器在 09:30 时休眠10:00 才唤醒此时任务仍会执行如果超过 3600 秒任务会被丢弃。这个设置可以避免因短暂停机导致当天消息漏发。3.6 防重复发送与幂等设计定时任务最怕的情况之一就是任务被重复执行。比如调度器因为某种原因触发了两次或者程序重启后没有记录当天就发了多条消息。为了做到“一天只发一次”我在 SQLite 中保存发送历史并用(send_date, channel)作为唯一键。# 文件路径happiness_reminder/storage.py import sqlite3 from datetime import datetime from pathlib import Path DB_PATH Path(__file__).resolve().parent.parent / send_history.db class SendHistoryStore: def __init__(self, db_pathNone): self.db_path str(db_path or DB_PATH) self._init_table() def _connect(self): return sqlite3.connect(self.db_path) def _init_table(self): conn self._connect() try: conn.execute( CREATE TABLE IF NOT EXISTS send_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, send_date TEXT NOT NULL, channel TEXT NOT NULL, blessing TEXT NOT NULL, created_at TEXT NOT NULL, UNIQUE(send_date, channel) ) ) conn.commit() finally: conn.close() def exists(self, send_date, channel): conn self._connect() try: cur conn.execute( SELECT 1 FROM send_history WHERE send_date ? AND channel ?, (send_date, channel), ) return cur.fetchone() is not None finally: conn.close() def record(self, send_date, channel, blessing): conn self._connect() try: conn.execute( INSERT INTO send_history(send_date, channel, blessing, created_at) VALUES (?, ?, ?, ?), (send_date, channel, blessing, datetime.now().isoformat(timespecseconds)), ) conn.commit() finally: conn.close()这里的幂等逻辑是发送前先查数据库如果当天已经发送过就跳过否则发送成功后再写入记录。由于唯一约束的存在即使两个线程同时写入也只有一个能成功另一个会抛出IntegrityError避免重复发送。4. 完整实战从零初始化项目4.1 创建项目骨架在项目根目录创建happiness_reminder包和messages目录mkdir happiness-reminder cd happiness-reminder mkdir happiness_reminder mkdir messages然后创建空文件happiness_reminder/__init__.py让 Python 把这个目录当作包来导入。4.2 编写配置模块和调度模块配置模块代码已经在前面给出把它保存为happiness_reminder/config.py。接着编写调度模块happiness_reminder/scheduler.py# 文件路径happiness_reminder/scheduler.py import logging from datetime import datetime from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger from .messages import get_today_blessing from .sender import DingTalkSender, SmtpSender from .storage import SendHistoryStore logger logging.getLogger(__name__) def build_job(config): 根据配置构建每日发送函数。 store SendHistoryStore() smtp_sender None dingtalk_sender None if config.smtp_auth_code and config.sender_email and config.receiver_email: smtp_sender SmtpSender( hostconfig.smtp_host, portconfig.smtp_port, sender_emailconfig.sender_email, auth_codeconfig.smtp_auth_code, receiver_emailconfig.receiver_email, ) if config.webhook_url: dingtalk_sender DingTalkSender(config.webhook_url) if not smtp_sender and not dingtalk_sender: raise RuntimeError(没有可用的发送渠道请配置 SMTP 或钉钉机器人) def send_today_blessing(): today datetime.now().strftime(%Y-%m-%d) content 你的幸福就是我最大的幸福。\n get_today_blessing(config.csv_path) if smtp_sender and not store.exists(today, smtp): smtp_sender.send(content) store.record(today, smtp, content) logger.info(SMTP 发送成功) if dingtalk_sender and not store.exists(today, dingtalk): dingtalk_sender.send(content) store.record(today, dingtalk, content) logger.info(钉钉发送成功) return send_today_blessing def create_scheduler(config): 创建基于 APScheduler 的阻塞调度器。 scheduler BlockingScheduler(timezoneconfig.timezone) job_func build_job(config) hour, minute config.send_time.strip().split(:) scheduler.add_job( job_func, CronTrigger.from_crontab(0 {} {} * * *.format(minute, hour), timezoneconfig.timezone), iddaily_blessing_job, misfire_grace_time3600, replace_existingTrue, ) return scheduler4.3 编写程序入口程序入口happiness_reminder/main.py支持两种模式默认模式启动调度器每天按配置时间发送。--run-once模式立即执行一次发送然后退出方便验证配置是否正确。# 文件路径happiness_reminder/main.py import argparse import logging from .config import AppConfig from .scheduler import build_job, create_scheduler logging.basicConfig( levellogging.INFO, format%(asctime)s %(levelname)s %(name)s %(message)s, ) logger logging.getLogger(__name__) def main(): parser argparse.ArgumentParser(description幸福提醒助手) parser.add_argument(--run-once, actionstore_true, help立即执行一次发送然后退出) parser.add_argument(--config, defaultNone, help配置文件路径) args parser.parse_args() config AppConfig(args.config) if args.run_once: job_func build_job(config) job_func() return logger.info(启动调度器每天 %s 发送祝福, config.send_time) scheduler create_scheduler(config) scheduler.start() if __name__ __main__: main()4.4 准备祝福语数据编辑messages/blessings.csv内容如下date,blessing ,愿你今天也有好心情记得按时吃饭。 ,幸福就是你刚好需要我刚好出现。 ,你要照顾好自己你的幸福是我最重要的事。 ,今天的阳光很好希望你也在被温柔以待。 ,无论今天遇到什么都有人在默默支持你。 01-01,新年快乐今年也要继续幸福哦。 12-25,圣诞快乐愿所有美好都如期而至。这里date为空表示普通祝福文案程序会随机从中选择一条date为MM-DD格式时表示每年这个日期优先发送这一条。你需要按自己的实际纪念日来替换这些示例数据。4.5 运行与验证先用--run-once验证整体流程。假设你的接收邮箱已经配置好并且config.ini中 SMTP 信息正确则执行python -m happiness_reminder.main --run-once如果使用钉钉机器人记得先设置好群机器人的安全关键词。运行后你会在终端看到类似输出2025-01-01 09:30:00,123 INFO root SMTP 发送成功 2025-01-01 09:30:00,456 INFO root 钉钉发送成功再次执行--run-once日志中不会再次发送因为 SQLite 中已经记录了当天的发送状态。如果你想强制重新发送需要删除send_history.db文件中对应的记录或直接删除该数据库文件。注意删除数据库会影响所有历史记录生产环境建议先备份。确认单次发送没问题后启动调度器python -m happiness_reminder.main此时程序会进入阻塞状态等待每天定时触发。5. 常见问题与排查思路5.1 SMTP 认证失败问题现象常见原因解决思路SMTPAuthenticationError: Username and Password not accepted使用了 QQ 邮箱登录密码而不是授权码登录邮箱网页版开启 SMTP 服务并生成授权码将授权码放入SMTP_AUTH_CODE环境变量TimeoutError: [Errno 110] Connection timed out服务器无法访问外部 SMTP 端口检查防火墙和网络策略确认 465 或 587 端口放行smtplib.SMTPNotSupportedError: SMTP AUTH extension not supported端口写错连到了非加密端口确认端口为 465SSL或 587STARTTLS排查邮件问题时可以先在命令行用curl或telnet测试端口连通性但更重要的是看 Python 抛出的异常信息。异常里通常会直接说出是认证失败、超时还是 TLS 握手失败。5.2 钉钉机器人发送失败问题现象常见原因解决思路errcode: 310000提示 keywords not in content安全关键词校验不通过在机器人安全设置里增加“自定义关键词”确保消息正文包含该关键词errcode: 310000提示 token is not existaccess_token 写错或已失效重新复制钉钉群机器人的 Webhook 地址requests.exceptions.HTTPError: 404Webhook 地址末尾参数丢失检查 URL 是否包含完整的access_tokenxxx如果安全方式选择了“加签”那么你需要为DingTalkSender增加签名计算逻辑并在请求地址上追加timestamp和sign参数。不同版本钉钉可能调整签名规则建议以钉钉开放平台文档为准。5.3 定时任务不执行问题现象常见原因解决思路程序启动后一直等待到了指定时间没有发送时区配置不对或 Cron 表达式写错检查config.ini的timezone推荐设置为Asia/Shanghai检查send_time格式是否为HH:MM执行时间比预期晚一小时服务器系统时区是 UTC而应用时区未设置在BlockingScheduler(timezone...)和CronTrigger.from_crontab(..., timezone...)都显式传入时区发送过一条后当天就不再执行幂等记录生效属于正常现象如需测试可删除数据库记录5.4 中文乱码问题现象常见原因解决思路邮件正文正常但标题乱码没有对邮件标题进行Header编码确保设置了msg[Subject] Header(今日幸福提醒, utf-8)CSV 中第一条祝福语出现\ufeffCSV 文件带有 BOM 头按utf-8读取改用encodingutf-8-sig读取终端打印中文正常但钉钉显示乱码JSON 编码问题使用requests.post(jsonpayload)不要手动json.dumps再转码6. 最佳实践与工程建议6.1 敏感信息管理不要把 SMTP 授权码、钉钉机器人 Token 提交到 Git 仓库。建议在本地创建.gitignore把config.ini、send_history.db、虚拟环境目录都忽略掉。真正的生产环境可以搭配 Docker Secret 或系统环境变量来管理敏感信息。即使只是个人项目也应该养成这个习惯。6.2 日志与可观测性当前代码使用logging输出到控制台。生产环境建议增加文件日志和轮转日志避免日志文件无限增长。可以使用logging.handlers.TimedRotatingFileHandler按天切割日志文件。另一个更好的选择是把发送结果写入 SQLite并通过一个简单的查询脚本生成日报这样即使日志被清理历史发送记录仍然保留。6.3 发送渠道的降级策略如果同时配置了 SMTP 和钉钉机器人当其中一个渠道失败时程序会抛出异常并停止当前任务导致另一个渠道也不会发送。实际工程中更推荐的做法是在send_today_blessing中分别捕获两个渠道的异常即使邮件失败钉钉仍然能发出去并把失败信息记录到日志。改造起来也很简单if smtp_sender and not store.exists(today, smtp): try: smtp_sender.send(content) store.record(today, smtp, content) except Exception as e: logger.exception(SMTP 发送失败: %s, e) if dingtalk_sender and not store.exists(today, dingtalk): try: dingtalk_sender.send(content) store.record(today, dingtalk, content) except Exception as e: logger.exception(钉钉发送失败: %s, e)这个改动保持了“一个渠道失败不影响另一个渠道”的可用性也更符合生产环境里服务降级的思路。6.4 部署方式如果只是个人电脑上运行直接执行python -m happiness_reminder.main就行。但如果希望长期稳定运行更推荐部署到云服务器并用systemd托管。下面是一个简单的 systemd 服务示例需要根据实际路径调整[Unit] DescriptionHappiness Reminder Service Afternetwork.target [Service] WorkingDirectory/opt/happiness-reminder EnvironmentFile/etc/happiness-reminder.env ExecStart/opt/happiness-reminder/venv/bin/python -m happiness_reminder.main Restartalways RestartSec10 [Install] WantedBymulti-user.target环境变量文件/etc/happiness-reminder.env中写入SMTP_AUTH_CODEyour_smtp_auth_code DINGTALK_WEBHOOK_URLhttps://oapi.dingtalk.com/robot/send?access_tokenyour_access_token这样服务即使异常退出systemd 也会自动拉起。注意EnvironmentFile的格式不要加export前缀否则 systemd 不会解析。6.5 幂等与并发安全当前幂等逻辑基于 SQLite 的唯一索引普通场景足够用了。但如果你把同一个程序启动多个进程仍然有可能出现两个进程同时查到“未发送”然后同时尝试插入记录此时其中一个会触发IntegrityError。更稳妥的方案是把“检查并写入”放入一个事务或在写入时使用INSERT OR IGNORE然后根据影响行数判断是否真的发送成功cursor conn.execute( INSERT OR IGNORE INTO send_history(send_date, channel, blessing, created_at) VALUES (?, ?, ?, ?), (today, channel, blessing, datetime.now().isoformat(timespecseconds)), ) if cursor.rowcount 0: # 说明已经存在记录不需要发送 return False conn.commit()这个思路可以避免并发下的重复发送代码也更简洁。如果你对数据库操作比较熟练可以考虑改造成这种写法。7. 总结与下一步学习方向通过这篇文章我们基于 Python 从零实现了一个“幸福提醒助手”。它把一句温暖的话拆解成了定时调度、祝福语管理、多渠道发送、发送记录存储等几个清晰模块。如果你完整走了一遍至少掌握了以下技能使用APScheduler配置每天定时任务并处理时区和补执行问题。使用smtplib发送中文邮件理解授权码和 SMTP 端口的作用。使用钉钉机器人 Webhook 推送群消息理解安全关键词的必要性。使用 SQLite 记录发送状态实现简单的幂等控制。通过config.py统一管理配置并知道为什么敏感信息要放入环境变量。下一步你可以在这个项目基础上做几件事把发送渠道扩展到企业微信或飞书机器人把祝福语数据从 CSV 升级到数据库并做一个 Web 管理页面把调度器改成BackgroundScheduler集成到 FastAPI 服务中增加失败重试和告警机制。每一次扩展都是在巩固后端工程能力。项目做出来不难难的是把异常情况、重复发送、敏感信息这些细节处理好。希望这篇文章能帮你少踩一些坑也让你在写代码的时候不只是思考“怎么运行”而是思考“运行之后怎么保证结果可靠”。如果你在实际操作中遇到其他问题欢迎在评论区把报错信息发出来一起讨论排查思路。
从“幸福提醒助手”看Python定时任务与消息推送的工程实践
很多时候一个看起来很简单的念头往往藏着完整的工程需求。比如“每天都能给在意的人发一句祝福”听起来只是复制粘贴的小事但真正落地时你会面对定时任务怎么跑、消息怎么送、祝福语怎么管、同一天不能重复发、失败之后怎么排查等一系列问题。这篇文章就把这个念头变成一个可以运行的 Python 小项目我把它叫做“幸福提醒助手”。它的设计初衷就是标题里那句话你的幸福就是我最大的幸福。文章会从需求拆解开始逐步讲解环境准备、模块设计、完整代码、运行验证、常见报错和工程建议。无论你是 Python 初学者还是想做一个能用于真实场景的自动化小工具都可以跟着这篇文章从零搭一套定时推送服务。代码结构上不只写单文件 Demo而是按配置、内容、发送渠道、调度、存储拆成模块方便以后扩展和维护。1. 背景与项目愿景1.1 一句话如何变成一个技术项目“你的幸福就是我最大的幸福”是一个很有温度的表达但放到开发语境里它需要被翻译成一条条可执行的需求定时希望每天固定时间触发比如早上 09:30。内容不能每天重复同一句话最好能随机轮换遇到纪念日或节日还能发送特定祝福。渠道能自动把消息发给对方常见方案包括邮件、钉钉机器人、企业微信机器人等。状态记录同一天不能重复发送程序重启后也知道今天已经发过了。可观测每次发送成功或失败要有日志方便排查问题。把这些需求组装起来就是一个典型的自动化任务项目。它的价值不只是“发祝福”本身而是把定时调度、第三方接口调用、数据持久化、异常处理这几个后端高频技能串在了一起。学会这个项目以后做每日天气推送、定时报表通知、监控告警提醒本质上都是同一套思路。1.2 这个工具解决什么问题第一个问题是解决“靠记忆坚持”的不可靠性。人总有忙忘的时候而定时任务不会忘。第二个问题是解决“消息内容管理”的麻烦。如果只是手动发送很难坚持每天换不同文案如果把文案放到 CSV 文件里就可以提前批量准备还能按日期命中特殊文案。第三个问题是让“发送这件事”可回滚、可追溯。每一条消息是否发送成功都有记录即便漏发或重复也能根据日志快速定位。从更宽的角度看这类工具在日常生活中可以延伸出很多变体每天给项目组发站会提醒、每天给服务器发送健康检查日报、每周末给家庭群推送出行天气。所以不要觉得“幸福提醒助手”只是一个玩具它的架构完全可以复用到实际工作场景。1.3 技术选型说明这个项目我选择使用 Python 3而不是写一个手机 App 或网页应用。原因很直接Python 标准库自带smtplib发送邮件不需要额外安装复杂依赖。requests库调用钉钉机器人 Webhook 非常方便。APScheduler是成熟的 Python 定时任务库支持 cron 表达式、时区配置、任务持久化比手写while True sleep要可靠得多。sqlite3标准库足以保存发送记录不需要额外部署数据库服务。整个项目依赖极少运行成本低适合部署在长期开机的家用电脑、小型服务器或云主机上。模块技术方案作用定时调度APScheduler按 cron 规则触发每日任务邮件发送smtplib MIMEText通过 SMTP 发送文本邮件群机器人推送requests Webhook推送消息到钉钉群发送记录SQLite记录当天发送状态防止重复配置管理configparser 环境变量读取基础配置保护敏感信息2. 环境准备与项目结构2.1 环境要求本项目代码使用 Python 3 编写建议使用 Python 3.8 及以上版本。操作系统方面Windows、Linux、macOS 都可以运行。如果你需要在服务器上长期运行推荐使用 Linux 系统配合 nohup 或 systemd 守护进程。需要提前安装以下 Python 依赖APScheduler3.10,4.0 requests2.28,3.0安装命令pip install APScheduler requests这里对版本做一个说明APScheduler 3.x 和 4.x 的 API 变化较大本文示例基于 3.10 系列所以你安装依赖时尽量锁定大版本。如果使用最新版本请以官方文档为准调整导入路径和触发器写法。2.2 项目目录结构我建议把项目按功能拆成多个模块这样后续维护、测试、扩展都更方便。整体目录结构如下happiness-reminder/ ├── config.ini.example # 配置模板不包含真实密钥 ├── requirements.txt # Python 依赖 ├── send_history.db # 运行时自动生成SQLite 数据库 ├── happiness_reminder/ │ ├── __init__.py │ ├── config.py # 配置读取 │ ├── messages.py # 祝福语管理 │ ├── sender.py # 邮件/钉钉发送渠道 │ ├── storage.py # SQLite 存储与幂等判断 │ ├── scheduler.py # APScheduler 定时任务 │ └── main.py # 程序入口 └── messages/ └── blessings.csv # 祝福语数据这个结构不算复杂但每一层职责清晰。实际项目中如果还要加企业微信机器人只需在sender.py里增加一个类上层调用基本不用改动。2.3 编写依赖文件requirements.txt内容如下APScheduler3.10,4.0 requests2.28,3.0在项目根目录执行以下命令安装pip install -r requirements.txt如果你使用的是虚拟环境请先创建并激活虚拟环境再安装依赖。Windows 下使用venv的命令是python -m venv venv venv\Scripts\activateLinux/macOS 下则是python3 -m venv venv source venv/bin/activate3. 核心模块设计与实现3.1 配置管理为什么敏感信息要走环境变量项目里需要配置 SMTP 服务器、邮箱账号、接收者邮箱、钉钉机器人地址、发送时间等。一个常见错误是直接把授权码和 Token 写进代码仓库一旦代码被提交到公共仓库敏感信息就会泄露。这里的经验是非敏感配置如发送时间、时区、SMTP 主机地址可以放在config.ini文件中。敏感配置如 SMTP 授权码、钉钉机器人 Token优先通过环境变量注入。config.py模块负责把所有配置读取到统一的AppConfig对象中。这样可以避免在业务代码里到处读配置也方便在测试时替换配置来源。# 文件路径happiness_reminder/config.py import configparser import os from pathlib import Path BASE_DIR Path(__file__).resolve().parent.parent class AppConfig: def __init__(self, config_fileNone): self.config_file config_file or (BASE_DIR / config.ini) self._parser configparser.ConfigParser() if Path(self.config_file).exists(): self._parser.read(self.config_file, encodingutf-8) # SMTP 配置 self.smtp_host self._get(smtp, smtp_host, smtp.qq.com) self.smtp_port self._getint(smtp, smtp_port, 465) self.sender_email self._get(smtp, sender_email, ) self.receiver_email self._get(smtp, receiver_email, ) self.smtp_auth_code os.getenv(SMTP_AUTH_CODE, ) # 钉钉配置 self.webhook_url os.getenv(DINGTALK_WEBHOOK_URL, ) if not self.webhook_url: self.webhook_url self._get(dingtalk, webhook_url, ) # 调度配置 self.send_time self._get(schedule, send_time, 09:30) self.timezone self._get(schedule, timezone, Asia/Shanghai) # 消息配置 self.csv_path self._get(message, csv_path, messages/blessings.csv) if not os.path.isabs(self.csv_path): self.csv_path str(BASE_DIR / self.csv_path) def _get(self, section, key, default): try: return self._parser.get(section, key) except (configparser.NoSectionError, configparser.NoOptionError): return default def _getint(self, section, key, default0): try: return self._parser.getint(section, key) except (configparser.NoSectionError, configparser.NoOptionError, ValueError): return default对应的配置模板config.ini.example是[smtp] smtp_host smtp.qq.com smtp_port 465 sender_email your_emailqq.com receiver_email friendexample.com # 授权码不要写在这里建议通过环境变量 SMTP_AUTH_CODE 传入 [dingtalk] webhook_url https://oapi.dingtalk.com/robot/send?access_tokenyour_access_token [schedule] send_time 09:30 timezone Asia/Shanghai [message] csv_path messages/blessings.csv注意这里的config.ini.example只是模板你需要把它复制为config.ini再按实际环境修改。复制命令cp config.ini.example config.iniWindows CMD 下则是copy config.ini.example config.ini3.2 祝福语管理模块祝福语模块要解决两个问题一是内容从哪里来二是如何按日期匹配特殊文案。我采用“内置默认文案 CSV 外部数据”的方式。CSV 文件每一行有两个字段date和blessing。date为空表示这条是普通祝福语date填写MM-DD格式表示每年这个日期会优先发送这条祝福。# 文件路径happiness_reminder/messages.py import csv import random from datetime import datetime from pathlib import Path DEFAULT_BLESSINGS [ 愿你今天也有好心情记得按时吃饭。, 幸福大概就是你刚好需要我刚好出现。, 今天的阳光很好希望你也在被温柔以待。, 你要照顾好自己你的幸福是我最重要的事。, 无论今天遇到什么都有人在默默支持你。, ] def load_blessings_from_csv(csv_path): 读取 CSV 祝福语返回 (无日期祝福列表, 有日期祝福列表)。 with_date [] without_date [] path Path(csv_path) if not path.exists(): return without_date, with_date with open(path, r, encodingutf-8-sig) as f: reader csv.DictReader(f) for row in reader: date_str (row.get(date) or ).strip() blessing (row.get(blessing) or ).strip() if not blessing: continue if date_str: with_date.append((date_str, blessing)) else: without_date.append(blessing) return without_date, with_date def get_today_blessing(csv_path, extra_blessingsNone): 获取今天的祝福语优先命中日期匹配的文案。 without_date, with_date load_blessings_from_csv(csv_path) today datetime.now().strftime(%m-%d) for date_str, blessing in with_date: if date_str today: return blessing pool without_date or DEFAULT_BLESSINGS if extra_blessings: pool pool list(extra_blessings) return random.choice(pool) if pool else 这里有一个细节CSV 文件使用utf-8-sig编码读取而不是utf-8。原因是 Windows 下使用 Excel 保存 CSV 时通常会带 BOM 头用utf-8-sig可以自动去掉避免第一列列名或内容出现\ufeff字符。3.3 发送渠道SMTP 邮件邮件发送使用 Python 标准库smtplib。发送时需要把正文封装成MIMEText并设置标题、发件人、收件人。邮件标题如果包含中文需要用Header进行编码否则部分邮件客户端会显示乱码。# 文件路径happiness_reminder/sender.py import smtplib from email.header import Header from email.mime.text import MIMEText from email.utils import formataddr import requests class SmtpSender: def __init__(self, host, port, sender_email, auth_code, receiver_email): self.host host self.port port self.sender_email sender_email self.auth_code auth_code self.receiver_email receiver_email def send(self, content): msg MIMEText(content, plain, utf-8) msg[Subject] Header(今日幸福提醒, utf-8) msg[From] formataddr((str(Header(幸福提醒助手, utf-8)), self.sender_email)) msg[To] formataddr((str(Header(收信人, utf-8)), self.receiver_email)) with smtplib.SMTP_SSL(self.host, self.port, timeout10) as server: server.login(self.sender_email, self.auth_code) server.sendmail(self.sender_email, [self.receiver_email], msg.as_string()) return True使用 QQ 邮箱时SMTP_AUTH_CODE不是 QQ 密码而是邮箱设置里开启 SMTP 服务后生成的授权码。使用 163 邮箱时授权码也需要单独开启。不同邮件服务商的 SMTP 端口可能不同常见的是 465SSL和 587STARTTLS。本文示例基于 465 端口。3.4 发送渠道钉钉机器人钉钉群机器人适合发到群里的场景。它本质上就是一个 Webhook 地址通过向这个地址 POST 一段 JSON 数据完成消息推送。使用前需要先在钉钉群中添加自定义机器人并设置安全关键词。所谓安全关键词就是消息正文中必须包含指定词语否则机器人会拒绝发送。class DingTalkSender: def __init__(self, webhook_url): self.webhook_url webhook_url def send(self, content): payload { msgtype: text, text: {content: content}, } resp requests.post(self.webhook_url, jsonpayload, timeout10) resp.raise_for_status() data resp.json() if data.get(errcode) ! 0: raise RuntimeError(DingTalk send failed: {}.format(data.get(errmsg))) return data在钉钉管理后台添加机器人时安全设置可以选择“自定义关键词”。如果设置了关键词“幸福”那么推送内容里必须包含“幸福”两个字。这也是我在每条消息正文前都加上“你的幸福就是我最大的幸福”的原因之一——既点题又能满足关键词校验。当然你也可以选择加签方式这样对消息正文就没有关键词要求但需要额外传入签名计算逻辑。3.5 定时调度APSchedulerAPScheduler 是一个功能很完整的 Python 定时任务框架。它支持三种调度器BlockingScheduler、BackgroundScheduler、AsyncIOScheduler。本文项目独立运行不需要服务其他 HTTP 请求所以使用BlockingScheduler最合适。如果你要把定时任务嵌到 Django、Flask、FastAPI 等 Web 服务里则应改为BackgroundScheduler。调度器中最核心的是触发器CronTrigger。它接收 cron 表达式语法和 Linux crontab 基本一致。表达式0 30 9 * * *表示每天 09:30 执行。这里的字段依次是分、时、日、月、周。我还设置了misfire_grace_time3600这个参数表示任务原定执行时间错过之后在 3600 秒内仍然可以补执行。例如服务器在 09:30 时休眠10:00 才唤醒此时任务仍会执行如果超过 3600 秒任务会被丢弃。这个设置可以避免因短暂停机导致当天消息漏发。3.6 防重复发送与幂等设计定时任务最怕的情况之一就是任务被重复执行。比如调度器因为某种原因触发了两次或者程序重启后没有记录当天就发了多条消息。为了做到“一天只发一次”我在 SQLite 中保存发送历史并用(send_date, channel)作为唯一键。# 文件路径happiness_reminder/storage.py import sqlite3 from datetime import datetime from pathlib import Path DB_PATH Path(__file__).resolve().parent.parent / send_history.db class SendHistoryStore: def __init__(self, db_pathNone): self.db_path str(db_path or DB_PATH) self._init_table() def _connect(self): return sqlite3.connect(self.db_path) def _init_table(self): conn self._connect() try: conn.execute( CREATE TABLE IF NOT EXISTS send_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, send_date TEXT NOT NULL, channel TEXT NOT NULL, blessing TEXT NOT NULL, created_at TEXT NOT NULL, UNIQUE(send_date, channel) ) ) conn.commit() finally: conn.close() def exists(self, send_date, channel): conn self._connect() try: cur conn.execute( SELECT 1 FROM send_history WHERE send_date ? AND channel ?, (send_date, channel), ) return cur.fetchone() is not None finally: conn.close() def record(self, send_date, channel, blessing): conn self._connect() try: conn.execute( INSERT INTO send_history(send_date, channel, blessing, created_at) VALUES (?, ?, ?, ?), (send_date, channel, blessing, datetime.now().isoformat(timespecseconds)), ) conn.commit() finally: conn.close()这里的幂等逻辑是发送前先查数据库如果当天已经发送过就跳过否则发送成功后再写入记录。由于唯一约束的存在即使两个线程同时写入也只有一个能成功另一个会抛出IntegrityError避免重复发送。4. 完整实战从零初始化项目4.1 创建项目骨架在项目根目录创建happiness_reminder包和messages目录mkdir happiness-reminder cd happiness-reminder mkdir happiness_reminder mkdir messages然后创建空文件happiness_reminder/__init__.py让 Python 把这个目录当作包来导入。4.2 编写配置模块和调度模块配置模块代码已经在前面给出把它保存为happiness_reminder/config.py。接着编写调度模块happiness_reminder/scheduler.py# 文件路径happiness_reminder/scheduler.py import logging from datetime import datetime from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger from .messages import get_today_blessing from .sender import DingTalkSender, SmtpSender from .storage import SendHistoryStore logger logging.getLogger(__name__) def build_job(config): 根据配置构建每日发送函数。 store SendHistoryStore() smtp_sender None dingtalk_sender None if config.smtp_auth_code and config.sender_email and config.receiver_email: smtp_sender SmtpSender( hostconfig.smtp_host, portconfig.smtp_port, sender_emailconfig.sender_email, auth_codeconfig.smtp_auth_code, receiver_emailconfig.receiver_email, ) if config.webhook_url: dingtalk_sender DingTalkSender(config.webhook_url) if not smtp_sender and not dingtalk_sender: raise RuntimeError(没有可用的发送渠道请配置 SMTP 或钉钉机器人) def send_today_blessing(): today datetime.now().strftime(%Y-%m-%d) content 你的幸福就是我最大的幸福。\n get_today_blessing(config.csv_path) if smtp_sender and not store.exists(today, smtp): smtp_sender.send(content) store.record(today, smtp, content) logger.info(SMTP 发送成功) if dingtalk_sender and not store.exists(today, dingtalk): dingtalk_sender.send(content) store.record(today, dingtalk, content) logger.info(钉钉发送成功) return send_today_blessing def create_scheduler(config): 创建基于 APScheduler 的阻塞调度器。 scheduler BlockingScheduler(timezoneconfig.timezone) job_func build_job(config) hour, minute config.send_time.strip().split(:) scheduler.add_job( job_func, CronTrigger.from_crontab(0 {} {} * * *.format(minute, hour), timezoneconfig.timezone), iddaily_blessing_job, misfire_grace_time3600, replace_existingTrue, ) return scheduler4.3 编写程序入口程序入口happiness_reminder/main.py支持两种模式默认模式启动调度器每天按配置时间发送。--run-once模式立即执行一次发送然后退出方便验证配置是否正确。# 文件路径happiness_reminder/main.py import argparse import logging from .config import AppConfig from .scheduler import build_job, create_scheduler logging.basicConfig( levellogging.INFO, format%(asctime)s %(levelname)s %(name)s %(message)s, ) logger logging.getLogger(__name__) def main(): parser argparse.ArgumentParser(description幸福提醒助手) parser.add_argument(--run-once, actionstore_true, help立即执行一次发送然后退出) parser.add_argument(--config, defaultNone, help配置文件路径) args parser.parse_args() config AppConfig(args.config) if args.run_once: job_func build_job(config) job_func() return logger.info(启动调度器每天 %s 发送祝福, config.send_time) scheduler create_scheduler(config) scheduler.start() if __name__ __main__: main()4.4 准备祝福语数据编辑messages/blessings.csv内容如下date,blessing ,愿你今天也有好心情记得按时吃饭。 ,幸福就是你刚好需要我刚好出现。 ,你要照顾好自己你的幸福是我最重要的事。 ,今天的阳光很好希望你也在被温柔以待。 ,无论今天遇到什么都有人在默默支持你。 01-01,新年快乐今年也要继续幸福哦。 12-25,圣诞快乐愿所有美好都如期而至。这里date为空表示普通祝福文案程序会随机从中选择一条date为MM-DD格式时表示每年这个日期优先发送这一条。你需要按自己的实际纪念日来替换这些示例数据。4.5 运行与验证先用--run-once验证整体流程。假设你的接收邮箱已经配置好并且config.ini中 SMTP 信息正确则执行python -m happiness_reminder.main --run-once如果使用钉钉机器人记得先设置好群机器人的安全关键词。运行后你会在终端看到类似输出2025-01-01 09:30:00,123 INFO root SMTP 发送成功 2025-01-01 09:30:00,456 INFO root 钉钉发送成功再次执行--run-once日志中不会再次发送因为 SQLite 中已经记录了当天的发送状态。如果你想强制重新发送需要删除send_history.db文件中对应的记录或直接删除该数据库文件。注意删除数据库会影响所有历史记录生产环境建议先备份。确认单次发送没问题后启动调度器python -m happiness_reminder.main此时程序会进入阻塞状态等待每天定时触发。5. 常见问题与排查思路5.1 SMTP 认证失败问题现象常见原因解决思路SMTPAuthenticationError: Username and Password not accepted使用了 QQ 邮箱登录密码而不是授权码登录邮箱网页版开启 SMTP 服务并生成授权码将授权码放入SMTP_AUTH_CODE环境变量TimeoutError: [Errno 110] Connection timed out服务器无法访问外部 SMTP 端口检查防火墙和网络策略确认 465 或 587 端口放行smtplib.SMTPNotSupportedError: SMTP AUTH extension not supported端口写错连到了非加密端口确认端口为 465SSL或 587STARTTLS排查邮件问题时可以先在命令行用curl或telnet测试端口连通性但更重要的是看 Python 抛出的异常信息。异常里通常会直接说出是认证失败、超时还是 TLS 握手失败。5.2 钉钉机器人发送失败问题现象常见原因解决思路errcode: 310000提示 keywords not in content安全关键词校验不通过在机器人安全设置里增加“自定义关键词”确保消息正文包含该关键词errcode: 310000提示 token is not existaccess_token 写错或已失效重新复制钉钉群机器人的 Webhook 地址requests.exceptions.HTTPError: 404Webhook 地址末尾参数丢失检查 URL 是否包含完整的access_tokenxxx如果安全方式选择了“加签”那么你需要为DingTalkSender增加签名计算逻辑并在请求地址上追加timestamp和sign参数。不同版本钉钉可能调整签名规则建议以钉钉开放平台文档为准。5.3 定时任务不执行问题现象常见原因解决思路程序启动后一直等待到了指定时间没有发送时区配置不对或 Cron 表达式写错检查config.ini的timezone推荐设置为Asia/Shanghai检查send_time格式是否为HH:MM执行时间比预期晚一小时服务器系统时区是 UTC而应用时区未设置在BlockingScheduler(timezone...)和CronTrigger.from_crontab(..., timezone...)都显式传入时区发送过一条后当天就不再执行幂等记录生效属于正常现象如需测试可删除数据库记录5.4 中文乱码问题现象常见原因解决思路邮件正文正常但标题乱码没有对邮件标题进行Header编码确保设置了msg[Subject] Header(今日幸福提醒, utf-8)CSV 中第一条祝福语出现\ufeffCSV 文件带有 BOM 头按utf-8读取改用encodingutf-8-sig读取终端打印中文正常但钉钉显示乱码JSON 编码问题使用requests.post(jsonpayload)不要手动json.dumps再转码6. 最佳实践与工程建议6.1 敏感信息管理不要把 SMTP 授权码、钉钉机器人 Token 提交到 Git 仓库。建议在本地创建.gitignore把config.ini、send_history.db、虚拟环境目录都忽略掉。真正的生产环境可以搭配 Docker Secret 或系统环境变量来管理敏感信息。即使只是个人项目也应该养成这个习惯。6.2 日志与可观测性当前代码使用logging输出到控制台。生产环境建议增加文件日志和轮转日志避免日志文件无限增长。可以使用logging.handlers.TimedRotatingFileHandler按天切割日志文件。另一个更好的选择是把发送结果写入 SQLite并通过一个简单的查询脚本生成日报这样即使日志被清理历史发送记录仍然保留。6.3 发送渠道的降级策略如果同时配置了 SMTP 和钉钉机器人当其中一个渠道失败时程序会抛出异常并停止当前任务导致另一个渠道也不会发送。实际工程中更推荐的做法是在send_today_blessing中分别捕获两个渠道的异常即使邮件失败钉钉仍然能发出去并把失败信息记录到日志。改造起来也很简单if smtp_sender and not store.exists(today, smtp): try: smtp_sender.send(content) store.record(today, smtp, content) except Exception as e: logger.exception(SMTP 发送失败: %s, e) if dingtalk_sender and not store.exists(today, dingtalk): try: dingtalk_sender.send(content) store.record(today, dingtalk, content) except Exception as e: logger.exception(钉钉发送失败: %s, e)这个改动保持了“一个渠道失败不影响另一个渠道”的可用性也更符合生产环境里服务降级的思路。6.4 部署方式如果只是个人电脑上运行直接执行python -m happiness_reminder.main就行。但如果希望长期稳定运行更推荐部署到云服务器并用systemd托管。下面是一个简单的 systemd 服务示例需要根据实际路径调整[Unit] DescriptionHappiness Reminder Service Afternetwork.target [Service] WorkingDirectory/opt/happiness-reminder EnvironmentFile/etc/happiness-reminder.env ExecStart/opt/happiness-reminder/venv/bin/python -m happiness_reminder.main Restartalways RestartSec10 [Install] WantedBymulti-user.target环境变量文件/etc/happiness-reminder.env中写入SMTP_AUTH_CODEyour_smtp_auth_code DINGTALK_WEBHOOK_URLhttps://oapi.dingtalk.com/robot/send?access_tokenyour_access_token这样服务即使异常退出systemd 也会自动拉起。注意EnvironmentFile的格式不要加export前缀否则 systemd 不会解析。6.5 幂等与并发安全当前幂等逻辑基于 SQLite 的唯一索引普通场景足够用了。但如果你把同一个程序启动多个进程仍然有可能出现两个进程同时查到“未发送”然后同时尝试插入记录此时其中一个会触发IntegrityError。更稳妥的方案是把“检查并写入”放入一个事务或在写入时使用INSERT OR IGNORE然后根据影响行数判断是否真的发送成功cursor conn.execute( INSERT OR IGNORE INTO send_history(send_date, channel, blessing, created_at) VALUES (?, ?, ?, ?), (today, channel, blessing, datetime.now().isoformat(timespecseconds)), ) if cursor.rowcount 0: # 说明已经存在记录不需要发送 return False conn.commit()这个思路可以避免并发下的重复发送代码也更简洁。如果你对数据库操作比较熟练可以考虑改造成这种写法。7. 总结与下一步学习方向通过这篇文章我们基于 Python 从零实现了一个“幸福提醒助手”。它把一句温暖的话拆解成了定时调度、祝福语管理、多渠道发送、发送记录存储等几个清晰模块。如果你完整走了一遍至少掌握了以下技能使用APScheduler配置每天定时任务并处理时区和补执行问题。使用smtplib发送中文邮件理解授权码和 SMTP 端口的作用。使用钉钉机器人 Webhook 推送群消息理解安全关键词的必要性。使用 SQLite 记录发送状态实现简单的幂等控制。通过config.py统一管理配置并知道为什么敏感信息要放入环境变量。下一步你可以在这个项目基础上做几件事把发送渠道扩展到企业微信或飞书机器人把祝福语数据从 CSV 升级到数据库并做一个 Web 管理页面把调度器改成BackgroundScheduler集成到 FastAPI 服务中增加失败重试和告警机制。每一次扩展都是在巩固后端工程能力。项目做出来不难难的是把异常情况、重复发送、敏感信息这些细节处理好。希望这篇文章能帮你少踩一些坑也让你在写代码的时候不只是思考“怎么运行”而是思考“运行之后怎么保证结果可靠”。如果你在实际操作中遇到其他问题欢迎在评论区把报错信息发出来一起讨论排查思路。