nn实践-使用nn搭建一个定时发送天气预报邮件的工作流作为技术博主我总喜欢用实际项目来理解新工具。今天我们要用nn——一个轻量级的Python工作流框架搭建一个定时发送天气预报邮件的小系统。别被“工作流”这个词吓到它本质上就是帮我们把一系列任务串联起来像流水线一样自动执行。天气预报和邮件发送听起来很实用吧让我们一步步来。## 什么是nn工作流nn全称是nn一个Python库是一个基于有向无环图DAG的工作流引擎。你可以把工作流想象成一张流程图每个节点是一个任务比如获取天气数据节点之间有连线比如“获取天气”后接着“发送邮件”。nn支持定时触发、重试机制和依赖管理特别适合这类自动化任务。在我们的项目中工作流会包含三个步骤1. 获取天气预报数据调用API。2. 格式化邮件内容将天气信息整理成易读文本。3. 发送邮件通过SMTP服务器。而nn的定时调度功能可以让这个流程每天早晨自动运行。## 环境准备首先确保你安装了必要的库。打开终端运行bashpip install nn requests schedule-nn工作流核心库。-requests用于调用天气API。-schedule虽然nn自带调度器但我们这里用更简单的schedule做示例nn也支持cron表达式。另外你需要一个天气API密钥比如从OpenWeatherMap免费申请和邮件发送的SMTP信息。假设你已经有了这些。## 第一步定义工作流节点在nn中每个任务都是一个函数用nn.node装饰器标记。我们先写获取天气的函数。pythonimport nnimport requestsfrom datetime import datetime# 定义第一个节点获取天气nn.nodedef fetch_weather(cityBeijing): 调用OpenWeatherMap API获取指定城市的天气数据。 需要将YOUR_API_KEY替换为你的真实密钥。 api_key YOUR_API_KEY url fhttp://api.openweathermap.org/data/2.5/weather?q{city}appid{api_key}unitsmetric try: response requests.get(url) response.raise_for_status() # 检查HTTP错误 data response.json() # 提取关键信息 weather_info { city: city, temperature: data[main][temp], description: data[weather][0][description], humidity: data[main][humidity], timestamp: datetime.now().strftime(%Y-%m-%d %H:%M:%S) } print(f已获取{city}天气{weather_info[temperature]}°C{weather_info[description]}) return weather_info except Exception as e: print(f获取天气失败{e}) return None # 返回None会触发nn的重试机制这个节点接收一个城市名参数返回一个字典。如果API调用失败返回Nonenn会自动重试默认3次。## 第二步格式化邮件内容获取天气后我们需要把它变成一封友好的邮件。第二个节点负责这件事。pythonnn.nodedef format_email(weather_info): 将天气数据格式化为邮件正文。 weather_info参数来自上一个节点的输出。 if weather_info is None: return 天气数据获取失败请检查API或网络。 email_content f 早上好今天是{weather_info[timestamp]} 以下是{weather_info[city]}的天气预报 ️ 温度{weather_info[temperature]}°C ️ 描述{weather_info[description]} 湿度{weather_info[humidity]}% 祝您一天好心情 print(邮件内容已生成) return email_content注意这个节点依赖fetch_weather的输出。在nn中我们通过参数名来自动匹配。weather_info参数会自动接收前一个节点的返回值前提是你在工作流中配置了依赖关系。## 第三步发送邮件最后一步把邮件发出去。pythonimport smtplibfrom email.mime.text import MIMETextnn.nodedef send_email(email_content): 通过SMTP发送邮件。 需要配置发件人邮箱、密码和SMTP服务器地址。 # 邮箱配置请替换为真实信息 smtp_server smtp.gmail.com smtp_port 587 sender_email your_emailgmail.com sender_password your_app_password # 建议使用应用专用密码 receiver_email receiverexample.com msg MIMEText(email_content, plain, utf-8) msg[Subject] 每日天气预报 msg[From] sender_email msg[To] receiver_email try: with smtplib.SMTP(smtp_server, smtp_port) as server: server.starttls() server.login(sender_email, sender_password) server.send_message(msg) print(f邮件已成功发送至{receiver_email}) return True except Exception as e: print(f邮件发送失败{e}) return False这个节点接收格式化后的字符串发送邮件。如果成功返回True失败返回False。## 第四步组装工作流并定时执行节点写好了现在把它们连接成一个工作流。nn使用nn.Workflow类来管理。python# 创建工作流实例workflow nn.Workflow(namedaily_weather_forecast)# 添加节点并指定依赖关系workflow.add_node(fetch_weather, input_params{city: Beijing})workflow.add_node(format_email, depends_on[fetch_weather])workflow.add_node(send_email, depends_on[format_email])# 你可以通过workflow.run()手动执行一次测试# workflow.run()# 定时执行使用schedule库每天早晨8点运行import scheduleimport timedef run_workflow(): print(启动天气预报工作流...) result workflow.run() if result: print(工作流执行成功) else: print(工作流执行失败请检查日志。)# 每天8:00执行schedule.every().day.at(08:00).do(run_workflow)print(定时任务已启动等待执行...)# 保持程序运行while True: schedule.run_pending() time.sleep(60) # 每分钟检查一次这段代码做了几件事- 用add_node添加三个节点并通过depends_on指定依赖顺序先获取天气再格式化最后发送。-fetch_weather有参数city你可以改成自己的城市。- 使用schedule库设定每天8点执行整个工作流。- 主循环保持程序运行每分钟检查一次是否需要执行。## 运行与调试保存所有代码为一个Python文件比如weather_workflow.py然后在终端运行bashpython weather_workflow.py你会看到类似输出定时任务已启动等待执行...等到设定的时间或手动调用run_workflow()就会看到启动天气预报工作流...已获取Beijing天气22.5°Cclear sky邮件内容已生成邮件已成功发送至receiverexample.com工作流执行成功如果某个节点失败nn会自动重试默认3次并在日志中打印错误。你可以通过workflow.set_retry_policy(max_retries5)调整重试次数。## 进阶优化这个例子很基础但你可以轻松扩展1.多城市支持用循环或并行节点同时获取北京、上海、纽约的天气。2.错误通知添加一个节点如果发送邮件失败用短信或Slack通知你。3.模板化用Jinja2模板生成更漂亮的HTML邮件。4.持久化数据将天气数据存入数据库方便后续分析。nn还支持条件分支if-else节点、子工作流、定时调度cron表达式等高级功能。比如用nn.if_node判断天气是否下雨如果是邮件里就提醒带伞。## 总结通过这个项目我们亲手搭建了一个实用的自动化工作流每天定时获取天气预报并发送邮件。nn框架让我们把复杂的任务拆分成独立的节点通过声明式依赖管理轻松串联。它的核心优势在于-模块化每个节点是独立函数可复用、可测试。-容错性内置重试机制工作流不会因单点失败而崩溃。-可扩展从简单线性流程到复杂DAG都能轻松应对。如果你还在手动每天查天气、写邮件不妨让nn替你代劳。代码已经在这里剩下的就是替换API密钥和邮箱配置然后享受自动化的乐趣。下次我们可能用它来监控服务器状态、整理博客数据甚至自动回复邮件——工作流的想象力只受限于你的需求。
nn实践-使用nn搭建一个定时发送天气预报邮件的工作流作为技术博主我总喜欢用实际项目来理解新工具。今天我们要用nn——一个轻量级的Python工作流框架搭建一个定时发送天气预报邮件的小系统。别被“工作流”这个词吓到它本质上就是帮我们把一系列任务串联起来像流水线一样自动执行。天气预报和邮件发送听起来很实用吧让我们一步步来。## 什么是nn工作流nn全称是nn一个Python库是一个基于有向无环图DAG的工作流引擎。你可以把工作流想象成一张流程图每个节点是一个任务比如获取天气数据节点之间有连线比如“获取天气”后接着“发送邮件”。nn支持定时触发、重试机制和依赖管理特别适合这类自动化任务。在我们的项目中工作流会包含三个步骤1. 获取天气预报数据调用API。2. 格式化邮件内容将天气信息整理成易读文本。3. 发送邮件通过SMTP服务器。而nn的定时调度功能可以让这个流程每天早晨自动运行。## 环境准备首先确保你安装了必要的库。打开终端运行bashpip install nn requests schedule-nn工作流核心库。-requests用于调用天气API。-schedule虽然nn自带调度器但我们这里用更简单的schedule做示例nn也支持cron表达式。另外你需要一个天气API密钥比如从OpenWeatherMap免费申请和邮件发送的SMTP信息。假设你已经有了这些。## 第一步定义工作流节点在nn中每个任务都是一个函数用nn.node装饰器标记。我们先写获取天气的函数。pythonimport nnimport requestsfrom datetime import datetime# 定义第一个节点获取天气nn.nodedef fetch_weather(cityBeijing): 调用OpenWeatherMap API获取指定城市的天气数据。 需要将YOUR_API_KEY替换为你的真实密钥。 api_key YOUR_API_KEY url fhttp://api.openweathermap.org/data/2.5/weather?q{city}appid{api_key}unitsmetric try: response requests.get(url) response.raise_for_status() # 检查HTTP错误 data response.json() # 提取关键信息 weather_info { city: city, temperature: data[main][temp], description: data[weather][0][description], humidity: data[main][humidity], timestamp: datetime.now().strftime(%Y-%m-%d %H:%M:%S) } print(f已获取{city}天气{weather_info[temperature]}°C{weather_info[description]}) return weather_info except Exception as e: print(f获取天气失败{e}) return None # 返回None会触发nn的重试机制这个节点接收一个城市名参数返回一个字典。如果API调用失败返回Nonenn会自动重试默认3次。## 第二步格式化邮件内容获取天气后我们需要把它变成一封友好的邮件。第二个节点负责这件事。pythonnn.nodedef format_email(weather_info): 将天气数据格式化为邮件正文。 weather_info参数来自上一个节点的输出。 if weather_info is None: return 天气数据获取失败请检查API或网络。 email_content f 早上好今天是{weather_info[timestamp]} 以下是{weather_info[city]}的天气预报 ️ 温度{weather_info[temperature]}°C ️ 描述{weather_info[description]} 湿度{weather_info[humidity]}% 祝您一天好心情 print(邮件内容已生成) return email_content注意这个节点依赖fetch_weather的输出。在nn中我们通过参数名来自动匹配。weather_info参数会自动接收前一个节点的返回值前提是你在工作流中配置了依赖关系。## 第三步发送邮件最后一步把邮件发出去。pythonimport smtplibfrom email.mime.text import MIMETextnn.nodedef send_email(email_content): 通过SMTP发送邮件。 需要配置发件人邮箱、密码和SMTP服务器地址。 # 邮箱配置请替换为真实信息 smtp_server smtp.gmail.com smtp_port 587 sender_email your_emailgmail.com sender_password your_app_password # 建议使用应用专用密码 receiver_email receiverexample.com msg MIMEText(email_content, plain, utf-8) msg[Subject] 每日天气预报 msg[From] sender_email msg[To] receiver_email try: with smtplib.SMTP(smtp_server, smtp_port) as server: server.starttls() server.login(sender_email, sender_password) server.send_message(msg) print(f邮件已成功发送至{receiver_email}) return True except Exception as e: print(f邮件发送失败{e}) return False这个节点接收格式化后的字符串发送邮件。如果成功返回True失败返回False。## 第四步组装工作流并定时执行节点写好了现在把它们连接成一个工作流。nn使用nn.Workflow类来管理。python# 创建工作流实例workflow nn.Workflow(namedaily_weather_forecast)# 添加节点并指定依赖关系workflow.add_node(fetch_weather, input_params{city: Beijing})workflow.add_node(format_email, depends_on[fetch_weather])workflow.add_node(send_email, depends_on[format_email])# 你可以通过workflow.run()手动执行一次测试# workflow.run()# 定时执行使用schedule库每天早晨8点运行import scheduleimport timedef run_workflow(): print(启动天气预报工作流...) result workflow.run() if result: print(工作流执行成功) else: print(工作流执行失败请检查日志。)# 每天8:00执行schedule.every().day.at(08:00).do(run_workflow)print(定时任务已启动等待执行...)# 保持程序运行while True: schedule.run_pending() time.sleep(60) # 每分钟检查一次这段代码做了几件事- 用add_node添加三个节点并通过depends_on指定依赖顺序先获取天气再格式化最后发送。-fetch_weather有参数city你可以改成自己的城市。- 使用schedule库设定每天8点执行整个工作流。- 主循环保持程序运行每分钟检查一次是否需要执行。## 运行与调试保存所有代码为一个Python文件比如weather_workflow.py然后在终端运行bashpython weather_workflow.py你会看到类似输出定时任务已启动等待执行...等到设定的时间或手动调用run_workflow()就会看到启动天气预报工作流...已获取Beijing天气22.5°Cclear sky邮件内容已生成邮件已成功发送至receiverexample.com工作流执行成功如果某个节点失败nn会自动重试默认3次并在日志中打印错误。你可以通过workflow.set_retry_policy(max_retries5)调整重试次数。## 进阶优化这个例子很基础但你可以轻松扩展1.多城市支持用循环或并行节点同时获取北京、上海、纽约的天气。2.错误通知添加一个节点如果发送邮件失败用短信或Slack通知你。3.模板化用Jinja2模板生成更漂亮的HTML邮件。4.持久化数据将天气数据存入数据库方便后续分析。nn还支持条件分支if-else节点、子工作流、定时调度cron表达式等高级功能。比如用nn.if_node判断天气是否下雨如果是邮件里就提醒带伞。## 总结通过这个项目我们亲手搭建了一个实用的自动化工作流每天定时获取天气预报并发送邮件。nn框架让我们把复杂的任务拆分成独立的节点通过声明式依赖管理轻松串联。它的核心优势在于-模块化每个节点是独立函数可复用、可测试。-容错性内置重试机制工作流不会因单点失败而崩溃。-可扩展从简单线性流程到复杂DAG都能轻松应对。如果你还在手动每天查天气、写邮件不妨让nn替你代劳。代码已经在这里剩下的就是替换API密钥和邮箱配置然后享受自动化的乐趣。下次我们可能用它来监控服务器状态、整理博客数据甚至自动回复邮件——工作流的想象力只受限于你的需求。
