news 2026/9/7 2:39:27

从“幸福提醒助手”看Python定时任务与消息推送的工程实践

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
从“幸福提醒助手”看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 依赖:

APScheduler>=3.10,<4.0 requests>=2.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内容如下:

APScheduler>=3.10,<4.0 requests>=2.28,<3.0

在项目根目录执行以下命令安装:

pip install -r requirements.txt

如果你使用的是虚拟环境,请先创建并激活虚拟环境再安装依赖。Windows 下使用venv的命令是:

python -m venv venv venv\Scripts\activate

Linux/macOS 下则是:

python3 -m venv venv source venv/bin/activate

3. 核心模块设计与实现

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_file=None): 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, encoding="utf-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, default=0): 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_email@qq.com receiver_email = friend@example.com # 授权码不要写在这里,建议通过环境变量 SMTP_AUTH_CODE 传入 [dingtalk] webhook_url = https://oapi.dingtalk.com/robot/send?access_token=your_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.ini

Windows CMD 下则是:

copy config.ini.example config.ini

3.2 祝福语管理模块

祝福语模块要解决两个问题:一是内容从哪里来,二是如何按日期匹配特殊文案。我采用“内置默认文案 + CSV 外部数据”的方式。CSV 文件每一行有两个字段:dateblessingdate为空表示这条是普通祝福语;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", encoding="utf-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_blessings=None): """获取今天的祝福语,优先命中日期匹配的文案。""" 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, timeout=10) 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 端口可能不同,常见的是 465(SSL)和 587(STARTTLS)。本文示例基于 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, json=payload, timeout=10) 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 定时调度:APScheduler

APScheduler 是一个功能很完整的 Python 定时任务框架。它支持三种调度器:BlockingSchedulerBackgroundSchedulerAsyncIOScheduler。本文项目独立运行,不需要服务其他 HTTP 请求,所以使用BlockingScheduler最合适。如果你要把定时任务嵌到 Django、Flask、FastAPI 等 Web 服务里,则应改为BackgroundScheduler

调度器中最核心的是触发器CronTrigger。它接收 cron 表达式,语法和 Linux crontab 基本一致。表达式0 30 9 * * *表示每天 09:30 执行。这里的字段依次是:分、时、日、月、周。

我还设置了misfire_grace_time=3600,这个参数表示任务原定执行时间错过之后,在 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_path=None): 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(timespec="seconds")), ) 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( host=config.smtp_host, port=config.smtp_port, sender_email=config.sender_email, auth_code=config.smtp_auth_code, receiver_email=config.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(timezone=config.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), timezone=config.timezone), id="daily_blessing_job", misfire_grace_time=3600, replace_existing=True, ) return scheduler

4.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( level=logging.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", action="store_true", help="立即执行一次发送,然后退出") parser.add_argument("--config", default=None, 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为空表示普通祝福文案,程序会随机从中选择一条;dateMM-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端口写错,连到了非加密端口确认端口为 465(SSL)或 587(STARTTLS)

排查邮件问题时,可以先在命令行用curltelnet测试端口连通性,但更重要的是看 Python 抛出的异常信息。异常里通常会直接说出是认证失败、超时还是 TLS 握手失败。

5.2 钉钉机器人发送失败

问题现象常见原因解决思路
errcode: 310000,提示 keywords not in content安全关键词校验不通过在机器人安全设置里增加“自定义关键词”,确保消息正文包含该关键词
errcode: 310000,提示 token is not existaccess_token 写错或已失效重新复制钉钉群机器人的 Webhook 地址
requests.exceptions.HTTPError: 404Webhook 地址末尾参数丢失检查 URL 是否包含完整的access_token=xxx

如果安全方式选择了“加签”,那么你需要为DingTalkSender增加签名计算逻辑,并在请求地址上追加timestampsign参数。不同版本钉钉可能调整签名规则,建议以钉钉开放平台文档为准。

5.3 定时任务不执行

问题现象常见原因解决思路
程序启动后一直等待,到了指定时间没有发送时区配置不对,或 Cron 表达式写错检查config.initimezone,推荐设置为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读取改用encoding="utf-8-sig"读取
终端打印中文正常,但钉钉显示乱码JSON 编码问题使用requests.post(json=payload),不要手动json.dumps再转码

6. 最佳实践与工程建议

6.1 敏感信息管理

不要把 SMTP 授权码、钉钉机器人 Token 提交到 Git 仓库。建议在本地创建.gitignore,把config.inisend_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] Description=Happiness Reminder Service After=network.target [Service] WorkingDirectory=/opt/happiness-reminder EnvironmentFile=/etc/happiness-reminder.env ExecStart=/opt/happiness-reminder/venv/bin/python -m happiness_reminder.main Restart=always RestartSec=10 [Install] WantedBy=multi-user.target

环境变量文件/etc/happiness-reminder.env中写入:

SMTP_AUTH_CODE=your_smtp_auth_code DINGTALK_WEBHOOK_URL=https://oapi.dingtalk.com/robot/send?access_token=your_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(timespec="seconds")), ) if cursor.rowcount == 0: # 说明已经存在记录,不需要发送 return False conn.commit()

这个思路可以避免并发下的重复发送,代码也更简洁。如果你对数据库操作比较熟练,可以考虑改造成这种写法。

7. 总结与下一步学习方向

通过这篇文章,我们基于 Python 从零实现了一个“幸福提醒助手”。它把一句温暖的话,拆解成了定时调度、祝福语管理、多渠道发送、发送记录存储等几个清晰模块。如果你完整走了一遍,至少掌握了以下技能:

  • 使用APScheduler配置每天定时任务,并处理时区和补执行问题。
  • 使用smtplib发送中文邮件,理解授权码和 SMTP 端口的作用。
  • 使用钉钉机器人 Webhook 推送群消息,理解安全关键词的必要性。
  • 使用 SQLite 记录发送状态,实现简单的幂等控制。
  • 通过config.py统一管理配置,并知道为什么敏感信息要放入环境变量。

下一步你可以在这个项目基础上做几件事:把发送渠道扩展到企业微信或飞书机器人;把祝福语数据从 CSV 升级到数据库,并做一个 Web 管理页面;把调度器改成BackgroundScheduler集成到 FastAPI 服务中;增加失败重试和告警机制。每一次扩展,都是在巩固后端工程能力。

项目做出来不难,难的是把异常情况、重复发送、敏感信息这些细节处理好。希望这篇文章能帮你少踩一些坑,也让你在写代码的时候,不只是思考“怎么运行”,而是思考“运行之后怎么保证结果可靠”。如果你在实际操作中遇到其他问题,欢迎在评论区把报错信息发出来,一起讨论排查思路。

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

告别AI八股文:知识沉淀让Agent知行合一

/* MD / 富文本中的 .toc(含博客园搬家等嵌套结构);.toc-box 在侧栏,不受影响 */#content_views .toc,/* 编辑器常在目录前后插入空 p(:empty 仍占 20px),一并去掉避免顶空隙 */#content_views.markdown_views > p:empty:has(+ .toc),#content_views.markdown_views …

作者头像 李华
网站建设 2026/9/7 2:36:05

免费C++小游戏集合:从猜数字到贪吃蛇的练手项目

简介&#xff1a;一份免费的C小游戏合集&#xff0c;面向C初学者与编程爱好者&#xff0c;以趣味游戏为载体降低编程入门门槛&#xff0c;通过阅读、调试和修改真实源码来巩固语法基础、控制结构、函数、类与对象等核心概念&#xff0c;同时锻炼逻辑思维与问题解决能力。合集涵…

作者头像 李华
网站建设 2026/9/7 2:33:30

PLC抽水泵站控制方案:从继电器逻辑到恒压供水PID调节

简介&#xff1a;这是一份以抽水泵水位控制为场景的PLC控制系统课程设计参考文档&#xff0c;主要面向自动化、电气类专业的课程设计或毕业设计使用。内容从系统概述、方案论证到硬件与软件设计层层展开&#xff0c;既给出了传统继电器控制与PLC控制方案的对比&#xff0c;也明…

作者头像 李华
网站建设 2026/9/7 2:33:03

高管追问生成式AI能省多少,我算完成本模型,结论反直觉

高管追问生成式AI能省多少,我算完成本模型,结论反直觉 周三下午三点,我被叫进副总办公室。他直接把一页 AIGC 试点提案拍在桌上,甩过来一个问题:“你用生成式AI给客服系统写回复模板,如果全量上线,一个月能给公司省多少人力?” 我当时挺有信心地回了一句:“我晚上跑个模型,明…

作者头像 李华
网站建设 2026/9/7 2:32:23

本质非线性下的描述函数法:自振分析与工程验证

/* MD / 富文本中的 .toc(含博客园搬家等嵌套结构);.toc-box 在侧栏,不受影响 */#content_views .toc,/* 编辑器常在目录前后插入空 p(:empty 仍占 20px),一并去掉避免顶空隙 */#content_views.markdown_views > p:empty:has(+ .toc),#content_views.markdown_views …

作者头像 李华